diff --git a/openflexure_microscope/stage/sanga.py b/openflexure_microscope/stage/sanga.py index d5ffcf15..5fa147ae 100644 --- a/openflexure_microscope/stage/sanga.py +++ b/openflexure_microscope/stage/sanga.py @@ -2,7 +2,7 @@ import numpy as np import time import logging from collections.abc import Iterable -from .sangaboard import Sangaboard +from sangaboard import Sangaboard from openflexure_microscope.stage.base import BaseStage from openflexure_microscope.utilities import axes_to_array diff --git a/openflexure_microscope/stage/sangaboard/__init__.py b/openflexure_microscope/stage/sangaboard/__init__.py deleted file mode 100644 index 1904bd7f..00000000 --- a/openflexure_microscope/stage/sangaboard/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -### THIS WILL EVENTUALLY BE REPLACED WITH A DEPENDENCY ON PYSANGABOARD ### -from .sangaboard import Sangaboard - -__version__ = "0.1" diff --git a/openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py b/openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py deleted file mode 100644 index a999128a..00000000 --- a/openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py +++ /dev/null @@ -1,536 +0,0 @@ -# -*- coding: utf-8 -*- -""" -This module has been modified from the chopped-out of class from nplab_. -It is a serial instrument class to simplify the process of interfacing with -equipment that talks on a serial port. The idea is that your instrument can -subclass :class:`ExtensibleSerialInstrument` and provide methods to control the -hardware, which will mostly consist of `self.query()` commands. It also has -options for adding OptionalModules so that you don't have to have multiple -classes for lots of instruments with different configurations - -The :class:`QueriedProperty` class is a convenient shorthand to create a property -that is read and/or set with a single serial query (i.e. a read followed by a write). - -.. module_author: Richard Rowman (c) 2017, released under GNU GPL -.. _nplab: http://www.github.com/nanophotonics/nplab -""" - - -from __future__ import division -import re -from functools import partial -import threading -import serial -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 -import io -import time -import logging -import warnings - - -class ExtensibleSerialInstrument(object): - """ - An instrument that communicates by sending strings back and forth over serial - - This base class provides commonly-used mechanisms that support the use of - serial instruments. Most interactions with this class involve - a call to the `query` method. This writes a message and returns the reply. - This has been hacked together from the nplab_ MessageBusInstrument and SerialInstrument - classes. - - **Threading Notes** - - The message bus protocol includes a property, `communications_lock`. All - commands that use the communications bus should be protected by this lock. - It's also permissible to use it to protect sequences of calls to the bus - that must be atomic (e.g. a multi-part exchange of messages). However, try - not to hold it too long - or odd things might happen if other threads are - blocked for a long time. The lock is reentrant so there's no issue with - acquiring it twice. - """ - - termination_character = ( - "\n" - ) #: All messages to or from the instrument end with this character. - termination_line = ( - None - ) #: If multi-line responses are recieved, they must end with this string - ignore_echo = False - port_settings = {} - - def __init__(self, port=None, **kwargs): - """ - Set up the serial port and so on. - """ - logging.info("Updating ESI port settings") - logging.debug(kwargs) - self.port_settings.update(kwargs) - logging.info("Opening ESI connection to port {}".format(port)) - self.open(port, False) # Eventually this shouldn't rely on init... - logging.info("Opened ESI connection to port {}".format(port)) - - def open(self, port=None, quiet=True): - """Open communications with the serial port. - - If no port is specified, it will attempt to autodetect. If quiet=True - then we don't warn when ports are opened multiple times. - """ - with self.communications_lock: - logging.debug(f"Opening port {port}") - if hasattr(self, "_ser") and self._ser.isOpen(): - 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. Are you sure the instrument is connected?" - logging.info("Creating serial.Serial instance...") - self._ser = serial.Serial(port, **self.port_settings) - logging.info(f"Created {self._ser}") - # 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 - # be set to 1 byte for maximum responsiveness. - assert ( - self.test_communications() - ), "The instrument doesn't seem to be responding. Did you specify the right port?" - - def close(self): - """Cleanly close the device. Includes proper logging statements.""" - logging.debug("Closing serial connection") - with self.communications_lock: - try: - self._ser.close() - except Exception as e: - logging.warning("The serial port didn't close cleanly: {}".format(e)) - logging.debug("Connection closed") - - def __del__(self): - """Emergency close the device. Try to avoid having to use this.""" - if hasattr(self, "_ser") and self._ser.isOpen(): - with self.communications_lock: - print( - "Closing an open serial communication has been triggered by garbage collection!\n" - "Please close this device more sensibly in future." - ) - try: - self._ser.close() - except Exception as e: - print("The serial port didn't close cleanly: {}".format(e)) - - def __enter__(self): - return self - - def __exit__(self, type, value, traceback): - """Cleanly close down the instrument at end of a with block.""" - self.close() - - def write(self, query_string): - """Write a string to the serial port""" - with self.communications_lock: - assert ( - self._ser.isOpen() - ), "Attempted to write to the serial port before it was opened. Perhaps you need to call the 'open' method first?" - logging.debug("Encoding message...") - data = query_string + self.termination_character - data = data.encode() - logging.debug(f"Writing message: {data}") - self._ser.write(data) - logging.debug("Write successful") - - def flush_input_buffer(self): - """Make sure there's nothing waiting to be read, and clear the buffer if there is.""" - with self.communications_lock: - if self._ser.inWaiting() > 0: - self._ser.flushInput() - - def readline(self, timeout=None): - """Read one line from the serial port.""" - self._ser.timeout = timeout - with self.communications_lock: - return ( - self._ser.readline() - .decode("utf8") - .replace(self.termination_character, "\n") - ) - - _communications_lock = None - - @property - def communications_lock(self): - """A lock object used to protect access to the communications bus""" - # This requires initialisation but our init method won't be called - so - # the property initialises it on first use. - if self._communications_lock is None: - self._communications_lock = threading.RLock() - return self._communications_lock - - def read_multiline(self, termination_line=None, timeout=None): - """Read one line from the underlying bus. Must be overriden. - - This should not need to be reimplemented unless there's a more efficient - way of reading multiple lines than multiple calls to readline().""" - with self.communications_lock: - if termination_line is None: - termination_line = self.termination_line - assert isinstance( - termination_line, str - ), "If you perform a multiline query, you must specify a termination line either through the termination_line keyword argument or the termination_line property of the NPSerialInstrument." - response = "" - last_line = "dummy" - while ( - termination_line not in last_line and len(last_line) > 0 - ): # read until we get the termination line. - last_line = self.readline(timeout) - response += last_line - return response - - def query(self, queryString, multiline=False, termination_line=None, timeout=None): - """ - Write a string to the stage controller and return its response. - - It will block until a response is received. The multiline and termination_line commands - will keep reading until a termination phrase is reached. - """ - with self.communications_lock: - self.flush_input_buffer() - logging.debug(f"Writing query: {queryString}") - self.write(queryString) - logging.debug("Waiting for query response...") - if self.ignore_echo == True: # Needs Implementing for a multiline read! - first_line = self.readline(timeout).strip() - if first_line == queryString: - return self.readline(timeout).strip() - else: - logging.info("This command did not echo!!!") - return first_line - - if termination_line is not None: - multiline = True - if multiline: - return self.read_multiline(termination_line) - else: - line = self.readline( - timeout - ).strip() # question: should we strip the final newline? - return line - - def parsed_query( - self, - query_string, - response_string=r"%d", - re_flags=0, - parse_function=None, - **kwargs, - ): - """ - Perform a query, returning a parsed form of the response. - - First query the instrument with the given query string, then compare - the response against a template. The template may contain text and - placeholders (e.g. %i and %f for integer and floating point values - respectively). Regular expressions are also allowed - each group is - considered as one item to be parsed. However, currently it's not - supported to use both % placeholders and regular expressions at the - same time. - - If placeholders %i, %f, etc. are used, the returned values are - automatically converted to integer or floating point, otherwise you - must specify a parsing function (applied to all groups) or a list of - parsing functions (applied to each group in turn). - """ - - response_regex = response_string - noop = lambda x: x # placeholder null parse function - placeholders = [ # tuples of (regex matching placeholder, regex to replace it with, parse function) - (r"%c", r".", noop), - ( - r"%(\d+)c", - r".{\1}", - noop, - ), # TODO support %cn where n is a number of chars - (r"%d", r"[-+]?\\d+", int), - (r"%[eEfg]", r"[-+]?(?:\\d+(?:\.\\d*)?|\.\\d+)(?:[eE][-+]?\\d+)?", float), - ( - r"%i", - r"[-+]?(?:0[xX][\\dA-Fa-f]+|0[0-7]*|\\d+)", - lambda x: int(x, 0), - ), # 0=autodetect base - (r"%o", r"[-+]?[0-7]+", lambda x: int(x, 8)), # 8 means octal - (r"%s", r"\\s+", noop), - (r"%u", r"\\d+", int), - ( - r"%[xX]", - r"[-+]?(?:0[xX])?[\\dA-Fa-f]+", - lambda x: int(x, 16), - ), # 16 forces hexadecimal - ] - matched_placeholders = [] - for placeholder, regex, parse_fun in placeholders: - response_regex = re.sub( - placeholder, "(" + regex + ")", response_regex - ) # substitute regex for placeholder - matched_placeholders.extend( - [ - (parse_fun, m.start()) - for m in re.finditer(placeholder, response_string) - ] - ) # save the positions of the placeholders - if parse_function is None: - parse_function = [ - f for f, s in sorted(matched_placeholders, key=lambda m: m[1]) - ] # order parse functions by their occurrence in the original string - if not hasattr(parse_function, "__iter__"): - parse_function = [parse_function] # make sure it's a list. - - reply = self.query(query_string, **kwargs) # do the query - # if match this could be because another response entered the buffer between write and read. Sleep for short while then - # check if something is now in the buffer, while the buffer is not empty repeat regex - waited = False - res = re.search(response_regex, reply, flags=re_flags) - while res is None: - if not waited: - time.sleep(0.1) - waited = True - original_reply = reply - if self._ser.inWaiting(): - reply = self.readline().strip() - res = re.search(response_regex, reply, flags=re_flags) - if res is not None: - warnings.warn( - "Query suceeded after initially receieving unmatched response ('%s') to '%s'. Match pattern /%s/ (generated regex /%s/)" - % ( - original_reply, - query_string, - response_string, - response_regex, - ), - RuntimeWarning, - ) - else: - raise ValueError( - "Stage response to '%s' ('%s') wasn't matched by /%s/ (generated regex /%s/)" - % (query_string, original_reply, response_string, response_regex) - ) - try: - parsed_result = [ - f(g) for f, g in zip(parse_function, res.groups()) - ] # try to apply each parse function to its argument - if len(parsed_result) == 1: - return parsed_result[0] - else: - return parsed_result - except ValueError: - logging.info("Matched Groups: {}".format(res.groups())) - logging.info("Parsing Functions {}:".format(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)""" - return self.parsed_query(query_string, "%d", **kwargs) - - def float_query(self, query_string, **kwargs): - """Perform a query and return the result(s) as float(s) (see parsedQuery)""" - return self.parsed_query(query_string, "%f", **kwargs) - - def test_communications(self): - """Check if the device is available on the current port. - - This should be overridden by subclasses. Assume the port has been - successfully opened and the settings are as defined by self.port_settings. - Usually this function sends a command and checks for a known reply.""" - 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.""" - print("Auto-scanning ports") - 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): - """This allows a `ExtensibleSerialInstrument` to have optional features. - - OptionalModule is designed as a base class - for interfacing with optional modules which may or may not be included with - the serial instrument, and can be added or removed at run-time. - """ - - def __init__( - self, available, parent=None, module_type="Undefined", model="Generic" - ): - assert ( - type(available) is bool - ), "Option module availablity should be a boolean not a {}".format( - type(available) - ) - self._available = available - self._parent = parent - assert ( - type(module_type) is str - ), "Option module type should be a string not a {}".format(type(module_type)) - self.module_type = module_type - if available: - assert ( - type(model) is str - ), "Option module type should be a string not a {}".format(type(model)) - self.model = model - else: - self.model = None - - @property - def available(self): - return self._available - - def confirm_available(self): - """Check if module is available, no return, will raise exception if not available!""" - assert self._available, 'No "{}" supported on firmware'.format(self.module_type) - - def describe(self): - """Consistently spaced desciption for listing modules""" - return self.module_type + " " * (25 - len(self.module_type)) + "- " + self.model - - -class QueriedProperty(object): - """A Property interface that reads and writes from the instrument on the bus. - - This returns a property-like (i.e. a descriptor) object. You can use it - in a class definition just like a property. The property it creates will - interact with the instrument over the communication bus to set and retrieve - its value. It uses calls to `ExtensibleSerialInstrument.parsed_query` to set or - get the value of the property. - - `QueriedProperty` can be used to define properties on a `ExtensibleSerialInstrument` - or an `OptionalModule` (in which case the `ExtensibleSerialInstrument.parsed_query` - method of the parent object will be used). - - Arguments: - - :get_cmd: - the string sent to the instrument to obtain the value - :set_cmd: - the string used to set the value (use {} or % placeholders) - :validate: - a list of allowable values - :valrange: - a maximum and minimum value - :fdel: - a function to call when it's deleted - :doc: - the docstring - :response_string: - supply a % code (as you would for response_string in a - ``ExtensibleSerialInstrument.parsed_query``) - :ack_writes: - set to "readline" to discard a line of input after writing. - """ - - def __init__( - self, - get_cmd=None, - set_cmd=None, - validate=None, - valrange=None, - fdel=None, - doc=None, - response_string=None, - ack_writes="no", - ): - self.response_string = response_string - self.get_cmd = get_cmd - self.set_cmd = set_cmd - self.validate = validate - self.valrange = valrange - self.fdel = fdel - self.ack_writes = ack_writes - self.__doc__ = doc - - # TODO: standardise the return (single value only vs parsed result), consider bool - def __get__(self, obj, objtype=None): - if issubclass(type(obj), OptionalModule): - obj.confirm_available() - obj = obj._parent - if obj is None: - return self - assert issubclass(type(obj), ExtensibleSerialInstrument) - if self.get_cmd is None: - raise AttributeError("unreadable attribute") - # Allow certain "magic" values to set the response string - for key, val in [("float", r"%f"), ("int", r"%d")]: - if self.response_string == key: - self.response_string = val - if self.response_string in ["bool", "raw", None]: - value = obj.query(self.get_cmd) - if self.response_string == "bool": - value = bool(value) - else: - value = obj.parsed_query(self.get_cmd, self.response_string) - return value - - def __set__(self, obj, value): - if issubclass(type(obj), OptionalModule): - obj.confirm_available() - obj = obj._parent - assert issubclass(type(obj), ExtensibleSerialInstrument) - if self.set_cmd is None: - raise AttributeError("can't set attribute") - if self.validate is not None: - if value not in self.validate: - raise ValueError( - "invalid value supplied - value must be one of {}".format( - self.validate - ) - ) - if self.valrange is not None: - if value < min(self.valrange) or value > max(self.valrange): - raise ValueError( - "invalid value supplied - value must be in the range {}-{}".format( - *self.valrange - ) - ) - message = self.set_cmd - if "{0" in message: - message = message.format(value) - elif "%" in message: - message = message % value - obj.write(message) - if self.ack_writes == "readline": - obj.readline() - - def __delete__(self, obj): - if self.fdel is None: - raise AttributeError("can't delete attribute") - self.fdel(obj) diff --git a/openflexure_microscope/stage/sangaboard/sangaboard.py b/openflexure_microscope/stage/sangaboard/sangaboard.py deleted file mode 100644 index 62008c3d..00000000 --- a/openflexure_microscope/stage/sangaboard/sangaboard.py +++ /dev/null @@ -1,372 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -""" -Sangaboard module - -This Python code allows control of the Sangaboard - -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 - - -class Sangaboard(ExtensibleSerialInstrument): - """Class managing serial communications with a Sangaboard - - The `Sangaboard` class handles setting up communications with the sangaboard, - wraps the various serial commands in Python methods, and provides iterators and - context managers to simplify opening/closing the hardware connection and some - other tasks like conducting a linear scan. - - Arguments to the constructor are passed to the constructor of - :class:`Sangaboard.extensible_serial_instrument.ExtensibleSerialInstrument`, - most likely the only one necessary is `port` which should be set to the serial port - you will use to communicate with the motor controller. - - This class can be used as a context manager, i.e. it's encouraged to use it as:: - - with Sangaboard() as sb: - sb.move_rel([1000,0,0]) - - In that case, the serial port will automatically be closed at the end of the block, - 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": 115_200, - "bytesize": EIGHTBITS, - "parity": PARITY_NONE, - "stopbits": STOPBITS_ONE, - } - - # 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", - doc="Get or set the minimum time between steps of the motors in microseconds.\n\n" - "The step time is ``1000000/max speed`` in steps/second. It is saved to EEPROM on " - "the sangaboard, so it will be persistent even if the motor controller is turned off.", - ) - ramp_time = QueriedProperty( - get_cmd="ramp_time?", - set_cmd="ramp_time %d", - response_string="ramp time %d", - doc="Get or set the acceleration time in microseconds.\n\n" - "The motors will accelerate/decelerate between stationary and maximum speed over `ramp_time` " - "microseconds. Zero means the motor runs at full speed initially, with no accleration " - "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") - - # Once initialised, `firmware` is a string that identifies the firmware version - firmware = None - - def __init__(self, port=None, timeout: int = 2, **kwargs): - """Create a sangaboard object. - - Arguments are passed to the constructor of - :class:`Sangaboard.extensible_serial_instrument.ExtensibleSerialInstrument`, - most likely the only one necessary is `port` which should be set to the serial port - you will use to communicate with the motor controller. That's the first argument so - it doesn't need to be named. - """ - - # Initialise basic serial instrument with specified - logging.info(f"Initialising ExtensibleSerialInstrument on port {port}") - ExtensibleSerialInstrument.__init__(self, port, **kwargs) - - try: - # Make absolutely sure that whatever port we're on is valid - logging.info("Checking valid firmware...") - self.check_valid_firmware() - - # Bit messy: Defining all valid modules as not available, then overwriting with available information if available. - logging.info("Loading modules...") - self.light_sensor = LightSensor(False) - - for module in self.list_modules(): - module_type = module.split(":")[0].strip() - module_model = module.split(":")[1].strip() - if module_type.startswith("Light Sensor"): - if module_model in self.supported_light_sensors: - self.light_sensor = LightSensor( - True, parent=self, model=module_model - ) - else: - 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: - logging.warning( - 'Module type "{}" not recognised.'.format(module_type) - ) - - self.board = self.query("board", timeout=2).rstrip() - - except Exception as e: - # 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() - logging.error(e) - logging.warning( - "You may need to update the firmware running on the Sangaboard." - ) - raise e - - def test_communications(self): - """ - Overrides superclass, used in self.open(), and port scanning - """ - logging.info("Testing communication to SangaBoard") - return self.check_valid_firmware() - - def check_valid_firmware(self): - logging.info("Running firmware checks...") - - # Request firmware version from the board - self.firmware = self.query("version", timeout=2).rstrip() - logging.info("Firmware response: {}".format(self.firmware)) - # 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.warning( - 'Version string "{}" not recognised.'.format(self.firmware) - ) - return False - else: - logging.warning("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.warning( - "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. - - displacement: integer or array/list of 3 integers - axis: None (for 3-axis moves) or one of 'x','y','z' - """ - if axis is not None: - assert axis in self.axis_names, "axis must be one of {}".format( - self.axis_names - ) - self.query("mr{} {}".format(axis, int(displacement))) - else: - # TODO: assert displacement is 3 integers - self.query("mr {} {} {}".format(*list(displacement))) - - def release_motors(self): - """De-energise the stepper motor coils""" - self.query("release") - - def zero_position(self): - """Set the current position to zero""" - self.query("zero") - - def move_abs(self, final, **kwargs): - """Make an absolute move to a position - - NB the sangaboard only accepts relative move commands, so this first - queries the board for its position, then instructs it to make about - relative move. - """ - rel_mov = [f_pos - i_pos for f_pos, i_pos in zip(final, self.position)] - return self.move_rel(rel_mov, **kwargs) - - def query(self, message, *args, **kwargs): - """Send a message and read the response. See ExtensibleSerialInstrument.query()""" - time.sleep(0.001) # This is to protect the stage from us talking too fast! - return ExtensibleSerialInstrument.query(self, message, *args, **kwargs) - - def list_modules(self): - """Return a list of strings detailing optional modules. - - Each module will correspond to a string of the form ``Module Name: Model`` - """ - modules = self.query( - "list_modules", multiline=True, termination_line="--END--\r\n" - ).split("\r\n")[:-2] - return [str(module) for module in modules] - - def print_help(self): - """Print the stage's built-in help message.""" - print(self.query("help", multiline=True, termination_line="--END--\r\n")) - - -class LightSensor(OptionalModule): - """An optional module giving access to the light sensor. - - If a light sensor is enabled in the motor controller's firmware, then - the :class:`sangaboard.Sangaboard` will gain an optional - module which is an instance of this class. It can be used to access - the light sensor (usually via the I2C bus). - """ - - valid_gains = None - _valid_gains_int = None - 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.", - ) - intensity = QueriedProperty( - get_cmd="light_sensor_intensity?", - response_string="%d", - doc="Read the current intensity measured by the light sensor (arbitrary units).", - ) - - def __init__(self, available, parent=None, model="Generic"): - super(LightSensor, self).__init__( - available, parent=parent, module_type="LightSensor", model=model - ) - if available: - self.valid_gains = self.__get_gain_values() - self._valid_gains_int = [int(g) for g in self.valid_gains] - - @property - def gain(self): - """"Get or set the current gain value of the light sensor. - - Valid gain values are defined in the `valid_gains` property, and should be floating-point numbers.""" - self.confirm_available() - gain = self._parent.query("light_sensor_gain?") - M = re.search("[0-9\.]+(?=x)", gain) - assert M is not None, 'Cannot read gain string: "{}"'.format(gain) - # gain is a float as non integer gains exist but are set with floor of value - return float(M.group()) - - @gain.setter - def gain(self, val): - self.confirm_available() - assert ( - int(val) in self._valid_gains_int - ), "Gain {} not valid must be one of: {}".format(val, self.valid_gains) - gain = self._parent.query("light_sensor_gain %d" % (int(val))) - M = re.search("[0-9\.]+(?=x)", gain) - assert M is not None, 'Cannot read gain string: "{}"'.format(gain) - # gain is a float as non integer gains exist but are set with floor of value - assert int(val) == int( - float(M.group()) - ), 'Gain of {} set, "{}" returned'.format(val, gain) - - def __get_gain_values(self): - """Read the allowable values for the light sensor's gain. - - This function will attempt to return a list of floating-point numbers which may - be used as values of the `gain` property. If the stage returns non-floating-point - values, the list will be of strings. - """ - self.confirm_available() - gains = self._parent.query("light_sensor_gain_values?") - try: - M = re.findall("[0-9\.]+(?=x)", gains) - return [float(gain) for gain in M] - except: - # Fall back to strings if we don't get floats (unlikely) - gain_strings = gains[20:].split(", ") - return gain_strings - - -class Endstops(OptionalModule): - """An optional module for use with endstops. - - If endstops are installed in the firmware the :class:`sangaboard.Sangaboard` - will gain an optional module which is an instance of this class. It can be used to retrieve - the type, state of the endstops, read and write maximum positions, and home. - """ - - installed = [] - """List of installed endstop types (min, max, soft)""" - - def __init__(self, available, parent=None, model="min"): - super(Endstops, self).__init__(available, parent=parent, model="Endstops") - self.installed = model.split(" ") - - status = QueriedProperty( - get_cmd="endstops?", - response_string=r"%d %d %d", - doc="Get endstops status as {-1,0,1} for {min,no,max} endstop triggered for each axis", - ) - maxima = QueriedProperty( - get_cmd="max_p?", - set_cmd="max %d %d %d", - response_string="%d %d %d", - doc="Vector of maximum positions, homing to max endstops will measure this, " - + "can be set to a known value for use with max only and min+soft endstops", - ) - - def home(self, direction="min", axes=["x", "y", "z"]): - """ Home given/all axes in the given direction (min/max/both) - - :param direction: one of {min,max,both} - :param axes: list of axes e.g. ['x','y'] - """ - ax = 0 - if "x" in axes: - ax += 1 - if "y" in axes: - ax += 2 - if "z" in axes: - ax += 3 - - if direction == "min" or direction == "both": - self._parent.query("home_min {}".format(ax)) - if direction == "max" or direction == "both": - self._parent.query("home_max {}".format(ax)) diff --git a/poetry.lock b/poetry.lock index 1b086cf9..54a1e21e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -110,6 +110,7 @@ ofmclient = [] reference = "f1803f5f655b405a6a902f6c697c9f720f6a18da" type = "git" url = "https://gitlab.com/openflexure/microscope-extensions/camera-stage-mapping.git" + [[package]] category = "main" description = "Pure Python CBOR (de)serializer with extensive tag support" @@ -205,6 +206,14 @@ version = "3.0.8" Flask = ">=0.9" Six = "*" +[[package]] +category = "main" +description = "Clean single-source support for Python 3 and 2" +name = "future" +optional = false +python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*" +version = "0.18.2" + [[package]] category = "main" description = "Coroutine-based network library" @@ -425,6 +434,7 @@ test = ["coverage", "pytest", "mock", "pillow", "numpy"] reference = "" type = "url" url = "https://github.com/rwb27/picamera/releases/download/v1.13.2b0/picamera-1.13.2b0-py3-none-any.whl" + [[package]] category = "main" description = "Python Imaging Library (Fork)" @@ -558,6 +568,19 @@ optional = true python-versions = "*" version = "0.6.5" +[[package]] +category = "main" +description = "Communication to the Sangaboard unipolar motor driver" +name = "sangaboard" +optional = false +python-versions = "*" +version = "0.2.0" + +[package.dependencies] +future = "*" +numpy = "*" +pyserial = "*" + [[package]] category = "main" description = "SciPy: Scientific Library for Python" @@ -814,7 +837,7 @@ testing = ["coverage (>=5.0.3)", "zope.event", "zope.testing"] rpi = ["RPi.GPIO"] [metadata] -content-hash = "b7d729f575bc8438cc20864e1c899cbd456d8a452c8b809e95846abc974211c4" +content-hash = "1c577dec4c5e870bacd76d24991f49cb0c7fed212ace179bea075e2849c91e00" python-versions = "^3.6" [metadata.files] @@ -908,6 +931,9 @@ flask-cors = [ {file = "Flask-Cors-3.0.8.tar.gz", hash = "sha256:72170423eb4612f0847318afff8c247b38bd516b7737adfc10d1c2cdbb382d16"}, {file = "Flask_Cors-3.0.8-py2.py3-none-any.whl", hash = "sha256:f4d97201660e6bbcff2d89d082b5b6d31abee04b1b3003ee073a6fd25ad1d69a"}, ] +future = [ + {file = "future-0.18.2.tar.gz", hash = "sha256:b1bead90b70cf6ec3f0710ae53a525360fa360d306a86583adc6bf83a4db537d"}, +] gevent = [ {file = "gevent-20.5.1-cp27-cp27m-macosx_10_15_x86_64.whl", hash = "sha256:2504563f44bb188c1e48684e2ac7d2793f9f5b1e1cf119a8fdf8c36d2bf2eaf7"}, {file = "gevent-20.5.1-cp27-cp27m-manylinux2010_x86_64.whl", hash = "sha256:1c2ad11663597d785e06daa8b65978a1536347a42bc840cf32823b54a0209d15"}, @@ -1240,6 +1266,10 @@ rope = [ "rpi.gpio" = [ {file = "RPi.GPIO-0.6.5.tar.gz", hash = "sha256:a4210ad63bfe844e43995286de0d3950dfacfa0f3799bb9392770ac54a7d2e47"}, ] +sangaboard = [ + {file = "sangaboard-0.2.0-py3-none-any.whl", hash = "sha256:510cb95a56e070ef7a9214f6a0673a92f2269e71e125c54b8bb1c16b1abfb095"}, + {file = "sangaboard-0.2.0.tar.gz", hash = "sha256:3636203179e9748cd27a1856d44de2b1c7adf398d033a8d5d382748903a098f0"}, +] scipy = [ {file = "scipy-1.4.1-cp35-cp35m-macosx_10_6_intel.whl", hash = "sha256:c5cac0c0387272ee0e789e94a570ac51deb01c796b37fb2aad1fb13f85e2f97d"}, {file = "scipy-1.4.1-cp35-cp35m-manylinux1_i686.whl", hash = "sha256:a144811318853a23d32a07bc7fd5561ff0cac5da643d96ed94a4ffe967d89672"}, diff --git a/pyproject.toml b/pyproject.toml index 19a6ab0d..94617f35 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.masonry.api" [tool.poetry] name = "openflexure-microscope-server" -version = "2.1.2" +version = "2.1.3" description = "Python module, and Flask-based web API, to run the OpenFlexure Microscope." authors = [ @@ -34,12 +34,8 @@ Flask = "^1.0" numpy = "1.18.2" Pillow = "^5.4" scipy = "1.4.1" # Exact version so we can guarantee a wheel - "RPi.GPIO" = { version = "^0.6.5", optional = true } picamera = {url = "https://github.com/rwb27/picamera/releases/download/v1.13.2b0/picamera-1.13.2b0-py3-none-any.whl" } - -pyserial = "^3.4" # Used for sangaboard (basic_serial_instrument) until we move to sangaboard pip library - python-dateutil = "^2.8" psutil = "^5.6.7" # Autostorage extension opencv-python-headless = [ @@ -49,6 +45,7 @@ opencv-python-headless = [ labthings = "0.6.4" pynpm = "^0.1.2" camera-stage-mapping = {git = "https://gitlab.com/openflexure/microscope-extensions/camera-stage-mapping.git"} +sangaboard = "^0.2.0" [tool.poetry.extras] rpi = ["RPi.GPIO"]