From ff534e6a51de6e8301d83dd532b645ed1d483953 Mon Sep 17 00:00:00 2001 From: Joel Collins Date: Wed, 15 May 2019 13:25:04 +0100 Subject: [PATCH] Initial support for SBv3 and MockStage --- openflexure_microscope/api/app.py | 14 +- openflexure_microscope/api/exceptions.py | 4 +- openflexure_microscope/microscope.py | 27 +- openflexure_microscope/plugins/loader.py | 7 +- openflexure_microscope/stage/base.py | 58 +++ openflexure_microscope/stage/mock.py | 59 +++ openflexure_microscope/stage/sanga.py | 144 ++++++ .../stage/sangaboard/__init__.py | 4 + .../extensible_serial_instrument.py | 414 ++++++++++++++++++ .../stage/sangaboard/sangaboard.py | 264 +++++++++++ 10 files changed, 966 insertions(+), 29 deletions(-) create mode 100644 openflexure_microscope/stage/base.py create mode 100644 openflexure_microscope/stage/mock.py create mode 100644 openflexure_microscope/stage/sanga.py create mode 100644 openflexure_microscope/stage/sangaboard/__init__.py create mode 100644 openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py create mode 100644 openflexure_microscope/stage/sangaboard/sangaboard.py diff --git a/openflexure_microscope/api/app.py b/openflexure_microscope/api/app.py index eb2837a8..6f859760 100644 --- a/openflexure_microscope/api/app.py +++ b/openflexure_microscope/api/app.py @@ -13,7 +13,8 @@ from openflexure_microscope.api.utilities import list_routes from openflexure_microscope import Microscope from openflexure_microscope.camera.pi import StreamingCamera -from openflexure_microscope.stage.openflexure import Stage +from openflexure_microscope.stage.sanga import SangaStage +from openflexure_microscope.stage.mock import MockStage import atexit import logging @@ -71,11 +72,12 @@ def attach_microscope(): logging.debug("Creating stage object...") try: - api_stage = Stage("/dev/ttyUSB0") - except SerialException as e: - api_camera.close() - raise e - else: + api_stage = SangaStage() + except (SerialException, OSError) as e: + logging.error(e) + logging.warning("No valid stage hardware found. Falling back to mock stage!") + api_stage = MockStage() + finally: logging.debug("Attaching devices to microscope...") api_microscope.attach( api_camera, diff --git a/openflexure_microscope/api/exceptions.py b/openflexure_microscope/api/exceptions.py index 95e03fd8..d3a8f048 100644 --- a/openflexure_microscope/api/exceptions.py +++ b/openflexure_microscope/api/exceptions.py @@ -1,4 +1,4 @@ -from flask import jsonify +from flask import jsonify, escape from werkzeug.exceptions import default_exceptions from werkzeug.exceptions import HTTPException @@ -21,7 +21,7 @@ class JSONExceptionHandler(object): response = { 'status_code': status_code, - 'message': message + 'message': escape(message) } return jsonify(response), status_code diff --git a/openflexure_microscope/microscope.py b/openflexure_microscope/microscope.py index fcc3030b..d50131cc 100644 --- a/openflexure_microscope/microscope.py +++ b/openflexure_microscope/microscope.py @@ -163,26 +163,10 @@ class Microscope(object): dict: Dictionary containing position data, and :py:attr:`openflexure_microscope.camera.base.BaseCamera.state` """ state = { - 'camera': {}, - 'stage': {}, - 'plugin': {} + 'camera': self.camera.state, + 'stage': self.stage.state, + 'plugin': self.plugin.state } - - # Add stage position - if self.stage: # If stage exists, populate with real values - position = self.stage.position - else: # Else, zero - position = [0, 0, 0] - - state['stage']['position'] = { - 'x': position[0], - 'y': position[1], - 'z': position[2], - } - - # Add camera state - state['camera'] = self.camera.state - return state def write_config(self, config: dict): @@ -216,7 +200,10 @@ class Microscope(object): # If attached to a stage if self.stage: - backlash = self.stage.backlash.tolist() + if hasattr(self.stage.backlash, 'tolist'): + backlash = self.stage.backlash.tolist() + else: + backlash = self.stage.backlash else: backlash = [0, 0, 0] diff --git a/openflexure_microscope/plugins/loader.py b/openflexure_microscope/plugins/loader.py index dfcaf63e..5ef9819c 100644 --- a/openflexure_microscope/plugins/loader.py +++ b/openflexure_microscope/plugins/loader.py @@ -131,11 +131,16 @@ class PluginMount(object): self.plugins = [] print("Creating plugin mount") + @property + def state(self): + return [m[0] for m in self.members] + @property def members(self): + ignores = ['state', 'members', 'attach'] plugin_array = [] for obj_name in dir(self): - if not obj_name == "plugins" and not obj_name[:2] == '__': + if not obj_name in ignores and not obj_name[:2] == '__': obj = getattr(self, obj_name) if isinstance(obj, MicroscopePlugin): plugin_members = [member for member in inspect.getmembers(obj) if not member[0][:2] == '__'] diff --git a/openflexure_microscope/stage/base.py b/openflexure_microscope/stage/base.py new file mode 100644 index 00000000..4236e7ee --- /dev/null +++ b/openflexure_microscope/stage/base.py @@ -0,0 +1,58 @@ +from abc import ABCMeta, abstractmethod +from openflexure_microscope.lock import StrictLock + +class BaseStage(metaclass=ABCMeta): + def __init__(self): + self.lock = StrictLock(timeout=5) #: :py:class:`openflexure_microscope.lock.StrictLock`: Strict lock controlling thread access to camera hardware + + @property + @abstractmethod + def state(self): + """The general state dictionary of the board. + Should at least contain 'position', and 'board' keys. + Note: A None/Null value for 'board' will disable stage + movement in the OpenFlexure eV client software, + """ + pass + + @property + @abstractmethod + def n_axes(self): + """The number of axes this stage has.""" + pass + + @property + @abstractmethod + def position(self): + """The current position, as a list""" + pass + + @property + @abstractmethod + def backlash(self): + """Get the distance used for backlash compensation.""" + pass + + @backlash.setter + @abstractmethod + def backlash(self): + """Set the distance used for backlash compensation.""" + pass + + @abstractmethod + def move_rel(self, displacement, backlash=True): + """Make a relative move, optionally correcting for backlash. + displacement: integer or array/list of 3 integers + backlash: (default: True) whether to correct for backlash. + """ + pass + + @abstractmethod + def move_abs(self, final, **kwargs): + """Make an absolute move to a position""" + pass + + @abstractmethod + def close(self): + """Cleanly close communication with the stage""" + pass diff --git a/openflexure_microscope/stage/mock.py b/openflexure_microscope/stage/mock.py new file mode 100644 index 00000000..f706e1b9 --- /dev/null +++ b/openflexure_microscope/stage/mock.py @@ -0,0 +1,59 @@ +from .sangaboard import Sangaboard +from openflexure_microscope.stage.base import BaseStage +from openflexure_microscope.lock import StrictLock + +import logging + +class MockStage(BaseStage): + def __init__(self, port=None, **kwargs): + BaseStage.__init__(self) + self._position = [0, 0, 0] + self._n_axis = 3 + self._backlash = None + + @property + def state(self): + """The general state dictionary of the board.""" + state = { + 'position': { + 'x': self.position[0], + 'y': self.position[1], + 'z': self.position[2], + }, + 'board': None, + 'version': '0' + } + return state + + + @property + def n_axes(self): + return self._n_axis + + @property + def position(self): + return self._position + + + @property + def backlash(self): + return self._backlash if self._backlash else [0]*self.n_axes + + @backlash.setter + def backlash(self, blsh): + if blsh is None: + self._backlash = None + try: + assert len(blsh) == self.n_axes + self._backlash = blsh + except: + self._backlash = [int(blsh)]*self.n_axes + + def move_rel(self, displacement, axis=None, backlash=True): + pass + + def move_abs(self, final, **kwargs): + pass + + def close(self): + pass \ No newline at end of file diff --git a/openflexure_microscope/stage/sanga.py b/openflexure_microscope/stage/sanga.py new file mode 100644 index 00000000..b14eff1b --- /dev/null +++ b/openflexure_microscope/stage/sanga.py @@ -0,0 +1,144 @@ +import numpy as np +from .sangaboard import Sangaboard +from openflexure_microscope.stage.base import BaseStage +from openflexure_microscope.lock import StrictLock + +import logging + +class SangaStage(BaseStage): + def __init__(self, port=None, **kwargs): + """Class managing serial communications with the motors for an Openflexure stage""" + BaseStage.__init__(self) + + self.board = Sangaboard(port, **kwargs) + self._backlash = None + + @property + def state(self): + """The general state dictionary of the board.""" + state = { + 'position': { + 'x': self.position[0], + 'y': self.position[1], + 'z': self.position[2], + }, + 'board': self.board.board, + 'firmware': self.board.firmware + } + return state + + @property + def n_axes(self): + """The number of axes this stage has.""" + return len(self.board.axis_names) + + @property + def position(self): + return self.board.position + + @property + def backlash(self): + """The distance used for backlash compensation. + Software backlash compensation is enabled by setting this property to a value + other than `None`. The value can either be an array-like object (list, tuple, + or numpy array) with one element for each axis, or a single integer if all axes + are the same. + The property will always return an array with the same length as the number of + axes. + The backlash compensation algorithm is fairly basic - it ensures that we always + approach a point from the same direction. For each axis that's moving, the + direction of motion is compared with ``backlash``. If the direction is opposite, + then the stage will overshoot by the amount in ``-backlash[i]`` and then move + back by ``backlash[i]``. This is computed per-axis, so if some axes are moving + in the same direction as ``backlash``, they won't do two moves. + """ + return self._backlash if self._backlash else np.array([0]*self.n_axes) + + @backlash.setter + def backlash(self, blsh): + if blsh is None: + self._backlash = None + try: + assert len(blsh) == self.n_axes + self._backlash = np.array(blsh, dtype=np.int) + except: + self._backlash = np.array([int(blsh)]*self.n_axes, dtype=np.int) + + def move_rel(self, displacement, axis=None, backlash=True): + """Make a relative move, optionally correcting for backlash. + displacement: integer or array/list of 3 integers + axis: None (for 3-axis moves) or one of 'x','y','z' + backlash: (default: True) whether to correct for backlash. + """ + with self.lock: + if not backlash or self.backlash is None: + return self.board.move_rel(displacement, axis=axis) + + if axis is not None: + # backlash correction is easier if we're always in 3D + # so this code just converts single-axis moves into all-axis moves. + assert axis in self.axis_names, "axis must be one of {}".format(self.axis_names) + move = np.zeros(self.n_axes, dtype=np.int) + move[np.argmax(np.array(self.axis_names) == axis)] = int(displacement) + displacement = move + + initial_move = np.array(displacement, dtype=np.int) + # Backlash Correction + # This backlash correction strategy ensures we're always approaching the + # end point from the same direction, while minimising the amount of extra + # motion. It's a good option if you're scanning in a line, for example, + # as it will kick in when moving to the start of the line, but not for each + # point on the line. + # For each axis where we're moving in the *opposite* + # direction to self.backlash, we deliberately overshoot: + initial_move -= np.where(self.backlash*displacement < 0, + self.backlash, + np.zeros(self.n_axes, dtype=self.backlash.dtype)) + self.board.move_rel(initial_move) + if np.any(displacement - initial_move != 0): + # If backlash correction has kicked in and made us overshoot, move + # to the correct end position (i.e. the move we were asked to make) + self.board.move_rel(displacement - initial_move) + + def move_abs(self, final, **kwargs): + """Make an absolute move to a position + """ + with self.lock: + self.board.move_abs(final, **kwargs) + + def close(self): + """Cleanly close communication with the stage""" + self.board.close() + + # Methods specific to Sangaboard + def release_motors(self): + """De-energise the stepper motor coils""" + self.board.release_motors() + + def __del__(self): + """Close the port when the object is deleted + + NB if the object is created in a with statement, this will cause + the port to be closed at the end of the with block.""" + self.close() + + def __enter__(self): + """When we use this in a with statement, remember where we started.""" + self._position_on_enter = self.position + return self + + def __exit__(self, type, value, traceback): + """The end of the with statement. Reset position if it went wrong. + NB the instrument is closed when the object is deleted, so we don't + need to worry about that here. + """ + if type is not None: + print("An exception occurred inside a with block, resetting " + + "position to its value at the start of the with block") + try: + time.sleep(0.5) + self.move_abs(self._position_on_enter) + except Exception as e: + print("A further exception occurred when resetting position: {}".format(e)) + print("Move completed, raising exception...") + raise value #propagate the exception \ No newline at end of file diff --git a/openflexure_microscope/stage/sangaboard/__init__.py b/openflexure_microscope/stage/sangaboard/__init__.py new file mode 100644 index 00000000..1904bd7f --- /dev/null +++ b/openflexure_microscope/stage/sangaboard/__init__.py @@ -0,0 +1,4 @@ +### 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 new file mode 100644 index 00000000..0e6df7d7 --- /dev/null +++ b/openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py @@ -0,0 +1,414 @@ +# -*- 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 print_function, 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 +import io +import time +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. + """ + self.port_settings.update(kwargs) + self.open(port, False) # Eventually this shouldn't rely on init... + + 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: + if hasattr(self,'_ser') and self._ser.isOpen(): + if not quiet: print("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?" + 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 + #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): + """Release the serial port""" + with self.communications_lock: + try: + self._ser.close() + except Exception as e: + print("The serial port didn't close cleanly:", e) + + def __del__(self): + """Close the port when the object is deleted + + NB if the object is created in a with statement, this will cause + the port to be closed at the end of the with block.""" + self.close() + + def __enter__(self): + return self + + def __exit__(self, type, value, traceback): + """Close down the instrument. This happens in __del__ though.""" + pass + + 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?" + #TODO: Check if this code is needed and if not kill it +# try: +# if self._ser.outWaiting()>0: self._ser.flushOutput() #ensure there's nothing waiting +# except AttributeError: +# if self._ser.out_waiting>0: self._ser.flushOutput() #ensure there's nothing waiting + data=query_string+self.termination_character + data=data.encode() + self._ser.write(data) + + 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.""" + 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() + self.write(queryString) + 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: + print('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: + return self.readline(timeout).strip() #question: should we strip the final newline? + + 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(.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: + print("Parsing Error") + print("Matched Groups:", res.groups()) + print("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)""" + 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.""" + 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. + + 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_typ)) + 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): + #print 'get', obj, objtype + 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): + #print 'set', 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 new file mode 100644 index 00000000..ad30e105 --- /dev/null +++ b/openflexure_microscope/stage/sangaboard/sangaboard.py @@ -0,0 +1,264 @@ +#!/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 re +import warnings + +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. + """ + 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 = 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.") + axis_names = ('x', 'y', 'z') + """The names of the sangaboard's axes. NB this also defines the number of axes.""" + firmware = None + """Once initialised, `firmware` is a string that identifies the firmware version.""" + + def __init__(self, port=None, **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. + """ + super(Sangaboard, self).__init__(port, **kwargs) + try: + # 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 + + #Bit messy: Defining all valid modules as not available, then overwriting with available information if available. + 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: + warnings.warn("Light sensor model \"%s\" not recognised."%(module_model),RuntimeWarning) + 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) + + 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() + warnings.warn("You may need to update the firmware running on the Sangaboard.") + raise e + + 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 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)) +