rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Sangaboard now handles port scanning and firmware checking

Browse source
This commit is contained in:
jtc42 2019-06-28 14:48:36 +01:00
parent 5f4af1e17a
commit 5823a87341
3 changed files with 124 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

1
openflexure_microscope/stage/sanga.py
View file

@ -186,6 +186,7 @@ class SangaStage(BaseStage):
def close(self):
"""Cleanly close communication with the stage"""
if hasattr(self, 'board'):
self.board.close()
# Methods specific to Sangaboard

55
openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py
View file

@ -16,12 +16,11 @@ that is read and/or set with a single serial query (i.e. a read followed by a wr
"""
from __future__ import print_function, division
from __future__ import division
import re
from functools import partial
import threading
import serial
import serial.tools.list_ports
from serial import FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS
from serial import PARITY_NONE, PARITY_EVEN, PARITY_ODD, PARITY_MARK, PARITY_SPACE
from serial import STOPBITS_ONE, STOPBITS_ONE_POINT_FIVE, STOPBITS_TWO
@ -55,17 +54,17 @@ class ExtensibleSerialInstrument(object):
ignore_echo = False
port_settings = {}
def __init__(self, port=None, **kwargs):
def __init__(self, port, **kwargs):
"""
Set up the serial port and so on.
"""
logging.info("Updating BSI port settings")
logging.info("Updating ESI port settings")
self.port_settings.update(kwargs)
logging.info("Opening BSI connection to port {}".format(port))
logging.info("Opening ESI connection to port {}".format(port))
self.open(port, False) # Eventually this shouldn't rely on init...
logging.info("Opened BSI connection to port {}".format(port))
logging.info("Opened ESI connection to port {}".format(port))
def open(self, port=None, quiet=True):
def open(self, port, quiet=True):
"""Open communications with the serial port.
If no port is specified, it will attempt to autodetect. If quiet=True
@ -73,10 +72,9 @@ class ExtensibleSerialInstrument(object):
"""
with self.communications_lock:
if hasattr(self,'_ser') and self._ser.isOpen():
if not quiet: print("Warning: attempted to open an already-open port!")
if not quiet: logging.warning("Attempted to open an already-open port!")
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 and autodetection failed. 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)
#the block above wraps the serial IO layer with a text IO layer
#this allows us to read/write in neat lines. NB the buffer size must
@ -90,7 +88,7 @@ class ExtensibleSerialInstrument(object):
try:
self._ser.close()
except Exception as e:
print("The serial port didn't close cleanly:", e)
logging.warning("The serial port didn't close cleanly:", e)
logging.debug("Connection closed")
def __del__(self):
@ -170,7 +168,7 @@ class ExtensibleSerialInstrument(object):
if first_line == queryString:
return self.readline(timeout).strip()
else:
print('This command did not echo!!!')
logging.info('This command did not echo!!!')
return first_line
if termination_line is not None:
@ -244,9 +242,9 @@ class ExtensibleSerialInstrument(object):
else:
return parsed_result
except ValueError:
print("Parsing Error")
print("Matched Groups:", res.groups())
print("Parsing Functions:", parse_function)
logging.error("Parsing Error")
logging.info("Matched Groups:", res.groups())
logging.info("Parsing Functions:", parse_function)
raise ValueError("Stage response to %s ('%s') couldn't be parsed by the supplied function" % (query_string, reply))
def int_query(self, query_string, **kwargs):
"""Perform a query and return the result(s) as integer(s) (see parsedQuery)"""
@ -264,31 +262,6 @@ class ExtensibleSerialInstrument(object):
with self.communications_lock:
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:
print("Trying port",port_name)
self.open(port_name)
success = True
print("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):
"""This allows a `ExtensibleSerialInstrument` to have optional features.
@ -369,7 +342,6 @@ class QueriedProperty(object):
# TODO: standardise the return (single value only vs parsed result), consider bool
def __get__(self, obj, objtype=None):
#print 'get', obj, objtype
if issubclass(type(obj),OptionalModule):
obj.confirm_available()
obj=obj._parent
@ -392,7 +364,6 @@ class QueriedProperty(object):
return value
def __set__(self, obj, value):
#print 'set', obj, value
if issubclass(type(obj),OptionalModule):
obj.confirm_available()
obj=obj._parent

139
openflexure_microscope/stage/sangaboard/sangaboard.py
View file

@ -10,6 +10,7 @@ It is (c) Richard Bowman & Julian Stirling 2019 and released under GNU GPL v3
from __future__ import print_function, division
import time
from .extensible_serial_instrument import ExtensibleSerialInstrument, OptionalModule, QueriedProperty, EIGHTBITS, PARITY_NONE, STOPBITS_ONE
import serial, serial.tools.list_ports
import re
import warnings
import logging
@ -36,10 +37,20 @@ class Sangaboard(ExtensibleSerialInstrument):
even if an error occurs. Otherwise, be sure to call the
:meth:`~.ExtensibleSerialInstrument.close()` method to release the serial port.
"""
# List of valid and deprecated firmwares, for communication checks
valid_firmwares = [
(0, 5),
(0, 4)
]
deprecated_firmwares = [
(0, 4)
]
# These are the settings for the sangaboards serial port, and can usually be left as default.
port_settings = {'baudrate':115200, 'bytesize':EIGHTBITS, 'parity':PARITY_NONE, 'stopbits':STOPBITS_ONE}
"""These are the settings for the sangaboards serial port, and can usually be left as default."""
# position, step time and ramp time are get/set using simple serial
# commands.
# position, step time and ramp time are get/set using simple serial commands.
position = QueriedProperty(get_cmd="p?", response_string=r"%d %d %d",
doc="Get the position of the axes as a tuple of 3 integers.")
step_time = QueriedProperty(get_cmd="dt?", set_cmd="dt %d", response_string="minimum step delay %d",
@ -53,10 +64,12 @@ class Sangaboard(ExtensibleSerialInstrument):
"control. Small moves may last less than `2*ramp_time`, in which case the acceleration "
"will be the same, but the motor will never reach full speed. It is saved to EEPROM on "
"the sangaboard, so it will be persistent even if the motor controller is turned off.")
# The names of the sangaboard's axes. NB this also defines the number of axes
axis_names = ('x', 'y', 'z')
"""The names of the sangaboard's axes. NB this also defines the number of axes."""
# Once initialised, `firmware` is a string that identifies the firmware version
firmware = None
"""Once initialised, `firmware` is a string that identifies the firmware version."""
def __init__(self, port=None, **kwargs):
"""Create a sangaboard object.
@ -67,32 +80,18 @@ class Sangaboard(ExtensibleSerialInstrument):
you will use to communicate with the motor controller. That's the first argument so
it doesn't need to be named.
"""
ExtensibleSerialInstrument.__init__(self, port, **kwargs)
# 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
ExtensibleSerialInstrument.__init__(self, scanned_port, **kwargs)
try:
logging.debug("Running firmware checks")
# Request firmware version from the board
self.firmware = self.query("version",timeout=2).rstrip()
# The slightly complicated regexp below will match the version string,
# and store the version number in the "groups" of the regexp. The version
# number should be in the format 1.2 and the groups will be "1.2", "1", "2"
# (for any number of elements).
try:
match = re.match(r"Sangaboard Firmware v(([\d]+)(?:\.([\d]+))+)", self.firmware)
assert match, "Version string \"{}\" not recognised.".format(self.firmware)
except AssertionError:
match = re.match(r"OpenFlexure Motor Board v(([\d]+)(?:\.([\d]+))+)", self.firmware)
assert match, "Version string \"{}\" not recognised.".format(self.firmware)
version = [int(g) for g in match.groups()[1:]]
self.firmware_version = match.group(1)
support_str = "This version of the Python module requires firmware v0.5 (with legacy support for v0.4)"
assert version[0] == 0, support_str
if version[1] == 4:
warnings.warn("Warning this Sangaboard is using v0.4 of the firmware, this will soon be depreciated!\n"+
"Please consider updating the Sangaboard firmware",Warning)
else:
assert version[1] == 5, support_str
# Make absolutely sure that whatever port we're on is valid
self.check_valid_firmware()
#Bit messy: Defining all valid modules as not available, then overwriting with available information if available.
self.light_sensor = LightSensor(False)
@ -104,11 +103,11 @@ class Sangaboard(ExtensibleSerialInstrument):
if module_model in self.supported_light_sensors:
self.light_sensor = LightSensor(True,parent=self,model=module_model)
else:
warnings.warn("Light sensor model \"%s\" not recognised."%(module_model),RuntimeWarning)
logging.warning("Light sensor model \"%s\" not recognised."%(module_model))
elif module_type.startswith('Endstops'):
self.endstops = Endstops(True, parent=self, model=module_model)
else:
warnings.warn("Module type \"{}\" not recognised.".format(module_type),RuntimeWarning)
logging.warning("Module type \"{}\" not recognised.".format(module_type))
self.board = self.query("board",timeout=2).rstrip()
@ -116,9 +115,79 @@ class Sangaboard(ExtensibleSerialInstrument):
# If an error occurred while setting up (e.g. because the board isn't connected or something)
# make sure we close the serial port cleanly (otherwise it hangs open).
self.close()
warnings.warn("You may need to update the firmware running on the Sangaboard.")
logging.error(e)
logging.error("You may need to update the firmware running on the Sangaboard.")
raise e
def scan_ports(self):
"""Iterate through the available serial ports and query them to see
if our instrument is there."""
logging.debug("Running Sangaboard port scanner")
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("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):
logging.debug("Running firmware checks")
# Request firmware version from the board
self.firmware = self.query("version",timeout=2).rstrip()
# Check for valid firmware string
if self.firmware:
match = re.match(r"Sangaboard Firmware v(([\d]+)(?:\.([\d]+))+)", self.firmware)
if not match:
# Try old firmware format
match = re.match(r"OpenFlexure Motor Board v(([\d]+)(?:\.([\d]+))+)", self.firmware)
if not match:
logging.error("Version string \"{}\" not recognised.".format(self.firmware))
return False
else:
logging.error("No firmware version string was returned.")
return False
# Check for matching/valid version number
version = [int(g) for g in match.groups()[1:]]
version_tuple = (version[0], version[1])
self.firmware_version = match.group(1)
if version_tuple not in Sangaboard.valid_firmwares:
logging.error("This version of the Python module requires firmware v0.5 (with legacy support for v0.4)")
return False
if version_tuple in Sangaboard.deprecated_firmwares:
logging.warning(
"Warning this Sangaboard is using v0.4 of the firmware, this will soon be depreciated! "
"Please consider updating the Sangaboard firmware"
)
return True
def move_rel(self, displacement, axis=None):
"""Make a relative move.
@ -173,10 +242,12 @@ class LightSensor(OptionalModule):
"""
valid_gains = None
_valid_gains_int = None
integration_time = QueriedProperty(get_cmd="light_sensor_integration_time?",
integration_time = QueriedProperty(
get_cmd="light_sensor_integration_time?",
set_cmd="light_sensor_integration_time %d",
response_string="light sensor integration time %d ms",
doc="Get or set the integration time of the light sensor in milliseconds.")
doc="Get or set the integration time of the light sensor in milliseconds."
)
intensity = QueriedProperty(get_cmd="light_sensor_intensity?", response_string="%d",
doc="Read the current intensity measured by the light sensor (arbitrary units).")