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

Merge remote-tracking branch 'origin/master' into deltastage

Browse source
This commit is contained in:
Samuel McDermott 2020-08-27 15:46:17 +01:00
commit 5e9e8cb2a3
222 changed files with 30104 additions and 7329 deletions
Download patch file Download diff file Expand all files Collapse all files

2
openflexure_microscope/stage/__init__.py
View file

@ -1,3 +1,3 @@
__all__ = ['base', 'mock', 'sanga']
__all__ = ["base", "mock", "sanga"]
from . import base, mock, sanga

71
openflexure_microscope/stage/base.py
View file

@ -1,34 +1,38 @@
import numpy as np
from abc import ABCMeta, abstractmethod
from openflexure_microscope.lock import StrictLock
from labthings import StrictLock
class BaseStage(metaclass=ABCMeta):
"""
Attributes:
lock (:py:class:`openflexure_microscope.lock.StrictLock`): Strict lock controlling thread
lock (:py:class:`labthings.StrictLock`): Strict lock controlling thread
access to camera hardware
"""
def __init__(self):
self.lock = StrictLock(timeout=5)
self.lock = StrictLock(name="Stage", timeout=None)
@abstractmethod
def apply_config(self, config: dict):
def update_settings(self, config: dict):
"""Update settings from a config dictionary"""
pass
@abstractmethod
def read_config(self):
def read_settings(self):
"""Return the current settings as a dictionary"""
pass
@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,
"""
"""The general state dictionary of the board."""
pass
@property
@abstractmethod
def configuration(self):
"""The general stage configuration."""
pass
@property
@ -43,6 +47,10 @@ class BaseStage(metaclass=ABCMeta):
"""The current position, as a list"""
pass
@property
def position_map(self):
return {"x": self.position[0], "y": self.position[1], "z": self.position[2]}
@property
@abstractmethod
def backlash(self):
@ -68,7 +76,50 @@ class BaseStage(metaclass=ABCMeta):
"""Make an absolute move to a position"""
pass
@abstractmethod
def zero_position(self):
"""Set the current position to zero"""
pass
@abstractmethod
def close(self):
"""Cleanly close communication with the stage"""
pass
def scan_linear(self, rel_positions, backlash=True, return_to_start=True):
"""
Scan through a list of (relative) positions (generator fn)
rel_positions should be an nx3-element array (or list of 3 element arrays).
Positions should be relative to the starting position - not a list of relative moves.
backlash argument is passed to move_rel
if return_to_start is True (default) we return to the starting position after a
successful scan. NB we always attempt to return to the starting position if an
exception occurs during the scan..
"""
starting_position = self.position
rel_positions = np.array(rel_positions)
assert rel_positions.shape[1] == 3, ValueError(
"Positions should be 3 elements long."
)
try:
self.move_rel(rel_positions[0], backlash=backlash)
yield 0
for i, step in enumerate(np.diff(rel_positions, axis=0)):
self.move_rel(step, backlash=backlash)
yield i + 1
except Exception as e:
return_to_start = True # always return to start if it went wrong.
raise e
finally:
if return_to_start:
self.move_abs(starting_position, backlash=backlash)
def scan_z(self, dz, **kwargs):
"""Scan through a list of (relative) z positions (generator fn)
This function takes a 1D numpy array of Z positions, relative to
the position at the start of the scan, and converts it into an
array of 3D positions with x=y=0. This, along with all the
keyword arguments, is then passed to ``scan_linear``.
"""
return self.scan_linear([[0, 0, z] for z in dz], **kwargs)

72
openflexure_microscope/stage/mock.py
View file

@ -3,47 +3,41 @@ from openflexure_microscope.utilities import axes_to_array
from collections.abc import Iterable
import numpy as np
import time
import logging
class MockStage(BaseStage):
class MissingStage(BaseStage):
def __init__(self, port=None, **kwargs):
BaseStage.__init__(self)
self._position = [0, 0, 0]
self._n_axis = 3
self._backlash = None
self.axis_names = ["x", "y", "z"] # Assume all sangaboards are 3 axis
@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'
}
state = {"position": self.position_map}
return state
def apply_config(self, config: dict):
"""Update settings from a config dictionary"""
@property
def configuration(self):
return {}
def update_settings(self, config: dict):
"""Update settings from a config dictionary"""
# Set backlash. Expects a dictionary with axis labels
if 'backlash' in config:
if "backlash" in config:
# Construct backlash array
backlash = axes_to_array(config['backlash'], ['x', 'y', 'z'], [0, 0, 0])
backlash = axes_to_array(config["backlash"], ["x", "y", "z"], [0, 0, 0])
self.backlash = backlash
def read_config(self) -> dict:
def read_settings(self) -> dict:
"""Return the current settings as a dictionary"""
blsh = self.backlash.tolist()
config = {
'backlash': {
'x': blsh[0],
'y': blsh[1],
'z': blsh[2],
}
}
config = {"backlash": {"x": blsh[0], "y": blsh[1], "z": blsh[2]}}
return config
@property
@ -69,13 +63,37 @@ class MockStage(BaseStage):
assert len(blsh) == self.n_axes
self._backlash = np.array(blsh)
else:
self._backlash = np.array([int(blsh)]*self.n_axes, dtype=np.int)
self._backlash = np.array([int(blsh)] * self.n_axes, dtype=np.int)
def move_rel(self, displacement, axis=None, backlash=True):
pass
def move_rel(
self, displacement: list, axis=None, backlash=True, simulate_time: bool = True
):
if simulate_time:
time.sleep(0.5)
if axis is not None:
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
def move_abs(self, final, **kwargs):
pass
initial_move = np.array(displacement, dtype=np.int)
self._position = list(np.array(self._position) + np.array(initial_move))
logging.debug(np.array(self._position) + np.array(initial_move))
logging.debug(f"New position: {self._position}")
def move_abs(self, final, simulate_time: bool = True, **kwargs):
if simulate_time:
time.sleep(0.5)
self._position = list(final)
logging.debug(f"New position: {self._position}")
def zero_position(self):
"""Set the current position to zero"""
self._position = [0, 0, 0]
def close(self):
pass

120
openflexure_microscope/stage/sanga.py
View file

@ -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
@ -18,28 +18,31 @@ class SangaStage(BaseStage):
board (:py:class:`openflexure_microscope.stage.sangaboard.Sangaboard`): Parent Sangaboard object.
_backlash (list): 3-element (element-per-axis) list of backlash compensation in steps.
"""
def __init__(self, port=None, **kwargs):
"""Class managing serial communications with the motors for an Openflexure stage"""
BaseStage.__init__(self)
self.port = port
self.board = Sangaboard(port, **kwargs)
self._backlash = None # Initialise backlash storage, used by property setter/getter
self.axis_names = ['x', 'y', 'z'] # Assume all sangaboards are 3 axis
self._backlash = (
None
) # Initialise backlash storage, used by property setter/getter
self.axis_names = ["x", "y", "z"] # Assume all sangaboards are 3 axis
@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 {"position": self.position_map}
@property
def configuration(self):
return {
"port": self.port,
"board": self.board.board,
"firmware": self.board.firmware,
}
return state
@property
def n_axes(self):
@ -48,7 +51,8 @@ class SangaStage(BaseStage):
@property
def position(self):
return self.board.position
with self.lock(timeout=None):
return self.board.position
@property
def backlash(self):
@ -80,37 +84,25 @@ class SangaStage(BaseStage):
assert len(blsh) == self.n_axes
self._backlash = np.array(blsh)
else:
self._backlash = np.array([int(blsh)]*self.n_axes, dtype=np.int)
self._backlash = np.array([int(blsh)] * self.n_axes, dtype=np.int)
def apply_config(self, config: dict):
def update_settings(self, config: dict):
"""Update settings from a config dictionary"""
# Set backlash. Expects a dictionary with axis labels
if 'backlash' in config:
if "backlash" in config:
# Construct backlash array
backlash = axes_to_array(config['backlash'], ['x', 'y', 'z'], [0, 0, 0])
backlash = axes_to_array(config["backlash"], ["x", "y", "z"], [0, 0, 0])
self.backlash = backlash
def read_config(self) -> dict:
def read_settings(self) -> dict:
"""Return the current settings as a dictionary"""
blsh = self.backlash.tolist()
config = {
'backlash': {
'x': blsh[0],
'y': blsh[1],
'z': blsh[2],
}
}
config = {"backlash": {"x": blsh[0], "y": blsh[1], "z": blsh[2]}}
return config
def steps_to_array(self, displacement, axis):
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)
return move
def move_rel(self, displacement, axis=None, backlash=True):
def move_rel(self, displacement: list, 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'
@ -121,8 +113,18 @@ class SangaStage(BaseStage):
displacement = self.steps_to_array(displacement, axis)
with self.lock:
logging.debug(f"Moving sangaboard by {displacement}")
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
@ -136,7 +138,7 @@ class SangaStage(BaseStage):
initial_move -= np.where(
self.backlash * displacement < 0,
self.backlash,
np.zeros(self.n_axes, dtype=self.backlash.dtype)
np.zeros(self.n_axes, dtype=self.backlash.dtype),
)
self.board.move_rel(initial_move)
if np.any(displacement - initial_move != 0):
@ -148,47 +150,17 @@ class SangaStage(BaseStage):
"""Make an absolute move to a position
"""
with self.lock:
logging.debug(f"Moving sangaboard to {final}")
self.board.move_abs(final, **kwargs)
def scan_linear(self, rel_positions, backlash=True, return_to_start=True):
"""
Scan through a list of (relative) positions (generator fn)
rel_positions should be an nx3-element array (or list of 3 element arrays).
Positions should be relative to the starting position - not a list of relative moves.
backlash argument is passed to move_rel
if return_to_start is True (default) we return to the starting position after a
successful scan. NB we always attempt to return to the starting position if an
exception occurs during the scan..
"""
starting_position = self.position
rel_positions = np.array(rel_positions)
assert rel_positions.shape[1] == 3, ValueError("Positions should be 3 elements long.")
try:
self.move_rel(rel_positions[0], backlash=backlash)
yield 0
for i, step in enumerate(np.diff(rel_positions, axis=0)):
self.move_rel(step, backlash=backlash)
yield i + 1
except Exception as e:
return_to_start = True # always return to start if it went wrong.
raise e
finally:
if return_to_start:
self.move_abs(starting_position, backlash=backlash)
def scan_z(self, dz, **kwargs):
"""Scan through a list of (relative) z positions (generator fn)
This function takes a 1D numpy array of Z positions, relative to
the position at the start of the scan, and converts it into an
array of 3D positions with x=y=0. This, along with all the
keyword arguments, is then passed to ``scan_linear``.
"""
return self.scan_linear([[0, 0, z] for z in dz], **kwargs)
def zero_position(self):
"""Set the current position to zero"""
with self.lock:
self.board.zero_position()
def close(self):
"""Cleanly close communication with the stage"""
if hasattr(self, 'board'):
if hasattr(self, "board"):
self.board.close()
# Methods specific to Sangaboard
@ -207,13 +179,17 @@ class SangaStage(BaseStage):
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")
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(
"A further exception occurred when resetting position: {}".format(e)
)
print("Move completed, raising exception...")
raise value # Propagate the exception

4
openflexure_microscope/stage/sangaboard/__init__.py
View file

@ -1,4 +0,0 @@
### THIS WILL EVENTUALLY BE REPLACED WITH A DEPENDENCY ON PYSANGABOARD ###
from .sangaboard import Sangaboard
__version__ = "0.1"

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

@ -1,422 +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")
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:
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?"
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):
"""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?"
#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:
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:
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:
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."""
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_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):
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)

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

@ -1,311 +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':115200, '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, **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
ExtensibleSerialInstrument.__init__(self, port, **kwargs)
try:
# 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)
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.error("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
"""
return self.check_valid_firmware()
def check_valid_firmware(self):
logging.debug("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.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.
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(self):
"""Zero the stored motor positions"""
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))