Merge branch 'og-comms-tests' into 'master'

Original serial port scanning and testing

See merge request openflexure/openflexure-microscope-server!30
This commit is contained in:
Joel Collins 2019-07-15 16:56:25 +00:00
commit 25171d7306
2 changed files with 36 additions and 40 deletions

View file

@ -54,7 +54,7 @@ class ExtensibleSerialInstrument(object):
ignore_echo = False ignore_echo = False
port_settings = {} port_settings = {}
def __init__(self, port, **kwargs): def __init__(self, port=None, **kwargs):
""" """
Set up the serial port and so on. Set up the serial port and so on.
""" """
@ -64,7 +64,7 @@ class ExtensibleSerialInstrument(object):
self.open(port, False) # Eventually this shouldn't rely on init... self.open(port, False) # Eventually this shouldn't rely on init...
logging.info("Opened ESI connection to port {}".format(port)) logging.info("Opened ESI connection to port {}".format(port))
def open(self, port, quiet=True): def open(self, port=None, quiet=True):
"""Open communications with the serial port. """Open communications with the serial port.
If no port is specified, it will attempt to autodetect. If quiet=True If no port is specified, it will attempt to autodetect. If quiet=True
@ -74,6 +74,8 @@ class ExtensibleSerialInstrument(object):
if hasattr(self,'_ser') and self._ser.isOpen(): if hasattr(self,'_ser') and self._ser.isOpen():
if not quiet: logging.warning("Attempted to open an already-open port!") if not quiet: logging.warning("Attempted to open an already-open port!")
return return
if port is None:
port=self.find_port()
assert port is not None, "We don't have a serial port to open, meaning you didn't specify a valid port. Are you sure the instrument is connected?" assert port is not None, "We don't have a serial port to open, meaning you didn't specify a valid port. Are you sure the instrument is connected?"
self._ser = serial.Serial(port,**self.port_settings) self._ser = serial.Serial(port,**self.port_settings)
#the block above wraps the serial IO layer with a text IO layer #the block above wraps the serial IO layer with a text IO layer
@ -262,6 +264,31 @@ class ExtensibleSerialInstrument(object):
with self.communications_lock: with self.communications_lock:
return True return True
def find_port(self):
"""Iterate through the available serial ports and query them to see
if our instrument is there."""
with self.communications_lock:
success = False
for port_name, _, _ in serial.tools.list_ports.comports(): #loop through serial ports, apparently 256 is the limit?!
try:
logging.info("Trying port {}".format(port_name))
self.open(port_name)
success = True
logging.info("Success!")
except:
pass
finally:
try:
self.close()
except:
pass #we don't care if there's an error closing the port...
if success:
break #again, make sure this happens *after* closing the port
if success:
return port_name
else:
return None
class OptionalModule(object): class OptionalModule(object):
"""This allows a `ExtensibleSerialInstrument` to have optional features. """This allows a `ExtensibleSerialInstrument` to have optional features.

View file

@ -81,13 +81,8 @@ class Sangaboard(ExtensibleSerialInstrument):
it doesn't need to be named. it doesn't need to be named.
""" """
# If no port is specified
if not port:
# Scan all available ports, and check for valid firmware
scanned_port = self.scan_ports()
# Initialise basic serial instrument with specified # Initialise basic serial instrument with specified
ExtensibleSerialInstrument.__init__(self, scanned_port, **kwargs) ExtensibleSerialInstrument.__init__(self, port, **kwargs)
try: try:
# Make absolutely sure that whatever port we're on is valid # Make absolutely sure that whatever port we're on is valid
@ -119,37 +114,11 @@ class Sangaboard(ExtensibleSerialInstrument):
logging.error("You may need to update the firmware running on the Sangaboard.") logging.error("You may need to update the firmware running on the Sangaboard.")
raise e raise e
def scan_ports(self): def test_communications(self):
"""Iterate through the available serial ports and query them to see """
if our instrument is there.""" Overrides superclass, used in self.open(), and port scanning
logging.debug("Running Sangaboard port scanner") """
with self.communications_lock: return self.check_valid_firmware()
success = False
for port_name, _, _ in serial.tools.list_ports.comports(): #loop through serial ports, apparently 256 is the limit?!
try:
logging.info("Trying port {}".format(port_name))
self.open(port_name)
success = True
logging.info("Checking firmware")
fw = self.check_valid_firmware()
if not fw:
success = False
except Exception as e:
logging.warning("Error on port {}".format(port_name))
logging.warning(e)
pass
finally:
try:
self.close()
except:
pass #we don't care if there's an error closing the port...
if success:
break #again, make sure this happens *after* closing the port
if success:
return port_name
else:
return None
def check_valid_firmware(self): def check_valid_firmware(self):
logging.debug("Running firmware checks") logging.debug("Running firmware checks")