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

Initial support for SBv3 and MockStage

Browse source
This commit is contained in:
Joel Collins 2019-05-15 13:25:04 +01:00
parent 068ce9e751
commit ff534e6a51
10 changed files with 966 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

14
openflexure_microscope/api/app.py
View file

@ -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,

4
openflexure_microscope/api/exceptions.py
View file

@ -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

27
openflexure_microscope/microscope.py
View file

@ -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]

7
openflexure_microscope/plugins/loader.py
View file

@ -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] == '__']

58
openflexure_microscope/stage/base.py Normal file
View file

@ -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

59
openflexure_microscope/stage/mock.py Normal file
View file

@ -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

144
openflexure_microscope/stage/sanga.py Normal file
View file

@ -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

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

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

414
openflexure_microscope/stage/sangaboard/extensible_serial_instrument.py Normal file
View file

@ -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)

264
openflexure_microscope/stage/sangaboard/sangaboard.py Normal file
View file

@ -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))