Startup
This commit is contained in:
0
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/__init__.py
Executable file
0
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/__init__.py
Executable file
322
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/command.py
Executable file
322
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/command.py
Executable file
@@ -0,0 +1,322 @@
|
||||
###
|
||||
# Copyright 2008-2011 Diamond Light Source Ltd.
|
||||
# This file is part of Diffcalc.
|
||||
#
|
||||
# Diffcalc is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# Diffcalc is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with Diffcalc. If not, see <http://www.gnu.org/licenses/>.
|
||||
###
|
||||
|
||||
#try:
|
||||
# from gda.device import Scannable
|
||||
#except ImportError:
|
||||
# from diffcalc.gdasupport.minigda.scannable import Scannable
|
||||
from diffcalc.gdasupport.minigda.scannable import Scannable
|
||||
from diffcalc.util import getMessageFromException, allnum, bold
|
||||
import math
|
||||
|
||||
|
||||
ROOT_NAMESPACE_DICT = {}
|
||||
|
||||
class Pos(object):
|
||||
|
||||
def __init__(self):
|
||||
self.__name__ = 'pos'
|
||||
|
||||
def __call__(self, *posargs):
|
||||
if len(posargs) == 0:
|
||||
|
||||
keys = dict(ROOT_NAMESPACE_DICT).keys()
|
||||
keys.sort()
|
||||
for key in keys:
|
||||
val = ROOT_NAMESPACE_DICT[key]
|
||||
if isinstance(val, Scannable):
|
||||
print self.posReturningReport(val)
|
||||
else:
|
||||
print self.posReturningReport(*posargs)
|
||||
|
||||
def posReturningReport(self, *posargs):
|
||||
# report position of this scannable
|
||||
if len(posargs) == 1:
|
||||
scannable = posargs[0]
|
||||
self._assert_scannable(scannable)
|
||||
return self._generatePositionReport(scannable)
|
||||
|
||||
# Move the scannable and report
|
||||
elif len(posargs) == 2:
|
||||
scannable = posargs[0]
|
||||
self._assert_scannable(scannable)
|
||||
# Move it
|
||||
scannable.asynchronousMoveTo(posargs[1])
|
||||
# TODO: minigda assumes all moves complete instantly, so no need
|
||||
# yet to check the move is complete
|
||||
return self._generatePositionReport(scannable)
|
||||
|
||||
else:
|
||||
raise ValueError(
|
||||
"Invlaid arguements: 'pos [ scannable [ value ] ]'")
|
||||
|
||||
def _assert_scannable(self, obj):
|
||||
if not isinstance(obj, Scannable):
|
||||
raise TypeError(
|
||||
"The first argument to the pos command must be scannable. "
|
||||
"Not: " + str(type(obj)))
|
||||
|
||||
def _generatePositionReport(self, scannable):
|
||||
fieldNames = (tuple(scannable.getInputNames()) +
|
||||
tuple(scannable.getExtraNames()))
|
||||
# All scannables
|
||||
result = "%s:" % scannable.getName()
|
||||
result = result.ljust(10)
|
||||
try:
|
||||
pos = scannable.getPosition()
|
||||
except Exception, e:
|
||||
return result + "Error: %s" % getMessageFromException(e)
|
||||
if pos is None:
|
||||
return result + "---"
|
||||
# Single field scannable:
|
||||
if len(fieldNames) == 1:
|
||||
try:
|
||||
result += "%s" % scannable.formatPositionFields(pos)[0]
|
||||
except AttributeError:
|
||||
result += str(scannable())
|
||||
# Multi field scannable:
|
||||
else:
|
||||
try:
|
||||
formatted = scannable.formatPositionFields(pos)
|
||||
for name, formattedValue in zip(fieldNames, formatted):
|
||||
result += "%s: %s " % (name, formattedValue)
|
||||
except AttributeError:
|
||||
result += str(scannable())
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class ScanDataHandler:
|
||||
def __init__(self):
|
||||
self.scannables = None
|
||||
|
||||
def callAtScanStart(self, scannables):
|
||||
pass
|
||||
|
||||
def callWithScanPoint(self, PositionDictIndexedByScannable):
|
||||
pass
|
||||
|
||||
def callAtScanEnd(self):
|
||||
pass
|
||||
|
||||
|
||||
class ScanDataPrinter(ScanDataHandler):
|
||||
|
||||
def __init__(self):
|
||||
self.first_point_printed = False
|
||||
self.widths = []
|
||||
self.scannables = []
|
||||
|
||||
def callAtScanStart(self, scannables):
|
||||
self.first_point_printed = False
|
||||
self.scannables = scannables
|
||||
|
||||
def print_first_point(self, position_dict):
|
||||
# also sets self.widths
|
||||
header_strings = []
|
||||
for scn in self.scannables:
|
||||
field_names = list(scn.getInputNames()) + list(scn.getExtraNames())
|
||||
if len(field_names) == 1:
|
||||
header_strings.append(scn.getName())
|
||||
else:
|
||||
for field_name in field_names:
|
||||
header_strings.append(field_name)
|
||||
|
||||
first_row_strings = []
|
||||
for scn in self.scannables:
|
||||
pos = position_dict[scn]
|
||||
first_row_strings.extend(scn.formatPositionFields(pos))
|
||||
|
||||
self.widths = []
|
||||
for header, pos_string in zip(header_strings, first_row_strings):
|
||||
self.widths.append(max(len(header), len(pos_string)))
|
||||
|
||||
header_cells = []
|
||||
for heading, width in zip(header_strings, self.widths):
|
||||
header_cells.append(heading.rjust(width))
|
||||
|
||||
underline_cells = ['-' * w for w in self.widths]
|
||||
|
||||
first_row_cells = []
|
||||
for pos, width in zip(first_row_strings, self.widths):
|
||||
first_row_cells.append(pos.rjust(width))
|
||||
|
||||
#table_width = sum(self.widths) + len(self.widths * 2) - 2
|
||||
lines = []
|
||||
#lines.append('=' * table_width)
|
||||
lines.append(bold(' '.join(header_cells)))
|
||||
lines.append(' '.join(underline_cells))
|
||||
lines.append(' '.join(first_row_cells))
|
||||
print '\n'.join(lines)
|
||||
|
||||
def callWithScanPoint(self, position_dict):
|
||||
if not self.first_point_printed:
|
||||
self.print_first_point(position_dict)
|
||||
self.first_point_printed = True
|
||||
else:
|
||||
row_strings = []
|
||||
for scn in self.scannables:
|
||||
pos = position_dict[scn]
|
||||
row_strings.extend(scn.formatPositionFields(pos))
|
||||
|
||||
row_cells = []
|
||||
for pos, width in zip(row_strings, self.widths):
|
||||
row_cells.append(pos.rjust(width))
|
||||
|
||||
print ' '.join(row_cells)
|
||||
|
||||
def callAtScanEnd(self):
|
||||
#table_width = sum(self.widths) + len(self.widths * 2) - 2
|
||||
#print '=' * table_width
|
||||
pass
|
||||
|
||||
|
||||
class Scan(object):
|
||||
class Group:
|
||||
def __init__(self, scannable):
|
||||
self.scannable = scannable
|
||||
self.args = []
|
||||
|
||||
def __cmp__(self, other):
|
||||
return(self.scannable.getLevel() - other.scannable.getLevel())
|
||||
|
||||
def __repr__(self):
|
||||
return "Group(%s, %s)" % (self.scannable.getName(), str(self.args))
|
||||
|
||||
def shouldTriggerLoop(self):
|
||||
return len(self.args) == 3
|
||||
|
||||
def __init__(self, scanDataHandlers):
|
||||
# scanDataHandlers should be list
|
||||
if type(scanDataHandlers) not in (tuple, list):
|
||||
scanDataHandlers = (scanDataHandlers,)
|
||||
self.dataHandlers = scanDataHandlers
|
||||
|
||||
def __call__(self, *scanargs):
|
||||
groups = self._parseScanArgsIntoScannableArgGroups(scanargs)
|
||||
groups = self._reorderInnerGroupsAccordingToLevel(groups)
|
||||
# Configure data handlers for a new scan
|
||||
for handler in self.dataHandlers: handler.callAtScanStart(
|
||||
[grp.scannable for grp in groups])
|
||||
# Perform the scan
|
||||
self._performScan(groups, currentRecursionLevel=0)
|
||||
# Inform data handlers of scan completion
|
||||
for handler in self.dataHandlers: handler.callAtScanEnd()
|
||||
|
||||
def _parseScanArgsIntoScannableArgGroups(self, scanargs):
|
||||
"""
|
||||
-> [ Group(scnA, (a1, a2, a2)), Group((scnB), (b1)), ...
|
||||
... Group((scnC),()), Group((scnD),(d1))]
|
||||
"""
|
||||
result = []
|
||||
if not isinstance(scanargs[0], Scannable):
|
||||
raise TypeError("First scan argument must be a scannable")
|
||||
|
||||
# Parse out scannables followed by non-scannable args
|
||||
for arg in scanargs:
|
||||
if isinstance(arg, Scannable):
|
||||
result.append(Scan.Group(arg))
|
||||
else:
|
||||
result[-1].args.append(arg)
|
||||
return result
|
||||
|
||||
def _reorderInnerGroupsAccordingToLevel(self, groups):
|
||||
# Find the first group not to trigger a loop
|
||||
for idx, group in enumerate(groups):
|
||||
if not group.shouldTriggerLoop():
|
||||
break
|
||||
latter = groups[idx:]; latter.sort() # Horrible hack not needed in python 3!
|
||||
return groups[:idx] + latter
|
||||
|
||||
def _performScan(self, groups, currentRecursionLevel):
|
||||
# groups[currentRecursionLevel:] will start with either:
|
||||
# a) A loop triggering group
|
||||
# b) A number (possibly 0) of non-loop triggering groups
|
||||
unprocessedGroups = groups[currentRecursionLevel:]
|
||||
|
||||
# 1) If first remaining group should trigger a loop, perform this loop,
|
||||
# recursively calling this method on the remaining groups
|
||||
if len(unprocessedGroups) > 0:
|
||||
first = unprocessedGroups[0]
|
||||
# If groups starts with a request to loop:
|
||||
if first.shouldTriggerLoop():
|
||||
posList = self._frange(first.args[0], first.args[1], first.args[2])
|
||||
for pos in posList:
|
||||
first.scannable.asynchronousMoveTo(pos)
|
||||
# TODO: Should wait. minigda assumes all moves complete immediately
|
||||
self._performScan(groups, currentRecursionLevel + 1)
|
||||
return
|
||||
|
||||
# 2) Move all non-loop triggering groups (may be zero)
|
||||
self._moveNonLoopTriggeringGroups(unprocessedGroups)
|
||||
|
||||
# 3) Sample position of all scannables
|
||||
posDict = self._samplePositionsOfAllScannables(groups)
|
||||
|
||||
# 4) Inform the data handlers that this point has been recorded
|
||||
for handler in self.dataHandlers: handler.callWithScanPoint(posDict)
|
||||
|
||||
def _moveNonLoopTriggeringGroups(self, groups):
|
||||
# TODO: Should wait. minigda assumes all moves complete immediately. groups could be zero lengthed.
|
||||
for grp in groups:
|
||||
if len(grp.args) == 0:
|
||||
pass
|
||||
elif len(grp.args) == 1:
|
||||
grp.scannable.asynchronousMoveTo(grp.args[0])
|
||||
elif len(grp.args) == 2:
|
||||
raise Exception("Scannables followed by two args not supported by minigda's scan command ")
|
||||
else:
|
||||
raise Exception("Scannable: %s args%s" % (grp.scannable, str(grp.args)))
|
||||
|
||||
def _samplePositionsOfAllScannables(self, groups):
|
||||
posDict = {}
|
||||
for grp in groups:
|
||||
posDict[grp.scannable] = grp.scannable.getPosition()
|
||||
return posDict
|
||||
|
||||
def _frange(self, limit1, limit2, increment):
|
||||
"""Range function that accepts floats (and integers).
|
||||
"""
|
||||
# limit1 = float(limit1)
|
||||
# limit2 = float(limit2)
|
||||
try:
|
||||
increment = float(increment)
|
||||
except TypeError:
|
||||
raise TypeError(
|
||||
"Only scaler values are supported, not GDA format vectors.")
|
||||
count = int(math.ceil(((limit2 - limit1) + increment / 100.) / increment))
|
||||
result = []
|
||||
for n in range(count):
|
||||
result.append(limit1 + n * increment)
|
||||
return result
|
||||
|
||||
|
||||
def sim(scn, hkl):
|
||||
"""sim hkl scn -- simulates moving scannable (not all)
|
||||
"""
|
||||
if not isinstance(hkl, (tuple, list)):
|
||||
raise TypeError()
|
||||
|
||||
if not allnum(hkl):
|
||||
raise TypeError()
|
||||
|
||||
try:
|
||||
print scn.simulateMoveTo(hkl)
|
||||
except AttributeError:
|
||||
raise TypeError(
|
||||
"The first argument does not support simulated moves")
|
||||
511
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/scannable.py
Executable file
511
script/__Lib/diffcalc-2.1/diffcalc/gdasupport/minigda/scannable.py
Executable file
@@ -0,0 +1,511 @@
|
||||
###
|
||||
# Copyright 2008-2011 Diamond Light Source Ltd.
|
||||
# This file is part of Diffcalc.
|
||||
#
|
||||
# Diffcalc is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# Diffcalc is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with Diffcalc. If not, see <http://www.gnu.org/licenses/>.
|
||||
###
|
||||
|
||||
import time
|
||||
|
||||
try:
|
||||
from gda.device.scannable import ScannableBase
|
||||
except ImportError:
|
||||
class Scannable(object):
|
||||
pass
|
||||
|
||||
class ScannableBase(Scannable):
|
||||
"""Implemtation of a subset of OpenGDA's Scannable interface
|
||||
"""
|
||||
|
||||
level = 5
|
||||
inputNames = []
|
||||
extraNames = []
|
||||
outputFormat = []
|
||||
|
||||
def isBusy(self):
|
||||
raise NotImplementedError()
|
||||
|
||||
def rawGetPosition(self):
|
||||
raise NotImplementedError()
|
||||
|
||||
def rawAsynchronousMoveTo(self, newpos):
|
||||
raise NotImplementedError()
|
||||
|
||||
def waitWhileBusy(self):
|
||||
while self.isBusy():
|
||||
time.sleep(.1)
|
||||
|
||||
def getPosition(self):
|
||||
return self.rawGetPosition()
|
||||
|
||||
def asynchronousMoveTo(self, newpos):
|
||||
self.rawAsynchronousMoveTo(newpos)
|
||||
|
||||
def atScanStart(self):
|
||||
pass
|
||||
|
||||
def atScanEnd(self):
|
||||
pass
|
||||
|
||||
def atCommandFailure(self):
|
||||
pass
|
||||
|
||||
###
|
||||
|
||||
def __repr__(self):
|
||||
pos = self.getPosition()
|
||||
formattedValues = self.formatPositionFields(pos)
|
||||
if len(tuple(self.getInputNames()) + tuple(self.getExtraNames())) > 1:
|
||||
result = self.getName() + ': '
|
||||
else:
|
||||
result = ''
|
||||
|
||||
names = tuple(self.getInputNames()) + tuple(self.getExtraNames())
|
||||
for name, val in zip(names, formattedValues):
|
||||
result += ' ' + name + ': ' + val
|
||||
return result
|
||||
###
|
||||
|
||||
def formatPositionFields(self, pos):
|
||||
"""Returns position as array of formatted strings"""
|
||||
# Make sure pos is a tuple or list
|
||||
if type(pos) not in (tuple, list):
|
||||
pos = tuple([pos])
|
||||
|
||||
# Sanity check
|
||||
if len(pos) != len(self.getOutputFormat()):
|
||||
raise Exception(
|
||||
"In scannable '%s':number of position fields differs from "
|
||||
"number format strings specified" % self.getName())
|
||||
|
||||
result = []
|
||||
for field, format in zip(pos, self.getOutputFormat()):
|
||||
if field is None:
|
||||
result.append('???')
|
||||
else:
|
||||
s = (format % field)
|
||||
## if width!=None:
|
||||
## s = s.ljust(width)
|
||||
result.append(s)
|
||||
|
||||
return result
|
||||
|
||||
def getName(self):
|
||||
return self.name
|
||||
|
||||
def setName(self, value):
|
||||
self.name = value
|
||||
|
||||
def getLevel(self):
|
||||
return self.level
|
||||
|
||||
def setLevel(self, value):
|
||||
self.level = value
|
||||
|
||||
def getInputNames(self):
|
||||
return self.inputNames
|
||||
|
||||
def setInputNames(self, value):
|
||||
self.inputNames = value
|
||||
|
||||
def getExtraNames(self):
|
||||
return self.extraNames
|
||||
|
||||
def setExtraNames(self, value):
|
||||
self.extraNames = value
|
||||
|
||||
def getOutputFormat(self):
|
||||
return self.outputFormat
|
||||
|
||||
def setOutputFormat(self, value):
|
||||
if type(value) not in (tuple, list):
|
||||
raise TypeError(
|
||||
"%s.setOutputFormat() expects tuple or list; not %s" %
|
||||
(self.getName(), str(type(value))))
|
||||
self.outputFormat = value
|
||||
|
||||
def __call__(self, newpos=None):
|
||||
if newpos is None:
|
||||
return self.getPosition()
|
||||
self.asynchronousMoveTo(newpos)
|
||||
|
||||
class ScannableAdapter(Scannable):
|
||||
'''Wrap up a Scannable and give it a new name and optionally an offset
|
||||
(added to the delegate when reading up and subtracting when setting down
|
||||
'''
|
||||
|
||||
def __init__(self, delegate_scn, name, offset=0):
|
||||
assert len(delegate_scn.getInputNames()) == 1
|
||||
assert len(delegate_scn.getExtraNames()) == 0
|
||||
self.delegate_scn = delegate_scn
|
||||
self.name = name
|
||||
self.offset = offset
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self.delegate_scn, name)
|
||||
|
||||
def getName(self):
|
||||
return self.name
|
||||
|
||||
def getInputNames(self):
|
||||
return [self.name]
|
||||
|
||||
def getPosition(self):
|
||||
return self.delegate_scn.getPosition() + self.offset
|
||||
|
||||
def asynchronousMoveTo(self, newpos):
|
||||
self.delegate_scn.asynchronousMoveTo(newpos - self.offset)
|
||||
|
||||
def __repr__(self):
|
||||
pos = self.getPosition()
|
||||
formatted_values = self.delegate_scn.formatPositionFields(pos)
|
||||
return self.name + ': ' + formatted_values[0] + ' ' + self.get_hint()
|
||||
|
||||
def get_hint(self):
|
||||
if self.offset:
|
||||
offset_hint = ' + ' if self.offset >= 0 else ' - '
|
||||
offset_hint += str(self.offset)
|
||||
else:
|
||||
offset_hint = ''
|
||||
return '(%s%s)' % (self.delegate_scn.name, offset_hint)
|
||||
|
||||
def __call__(self, newpos=None):
|
||||
if newpos is None:
|
||||
return self.getPosition()
|
||||
self.asynchronousMoveTo(newpos)
|
||||
|
||||
class SingleFieldDummyScannable(ScannableBase):
|
||||
|
||||
def __init__(self, name, initial_position=0.):
|
||||
self.name = name
|
||||
self.inputNames = [name]
|
||||
self.outputFormat = ['% 6.4f']
|
||||
self.level = 3
|
||||
self._current_position = float(initial_position)
|
||||
|
||||
def isBusy(self):
|
||||
return False
|
||||
|
||||
def waitWhileBusy(self):
|
||||
return
|
||||
|
||||
def asynchronousMoveTo(self, new_position):
|
||||
self._current_position = float(new_position)
|
||||
|
||||
def getPosition(self):
|
||||
return self._current_position
|
||||
|
||||
|
||||
class DummyPD(SingleFieldDummyScannable):
|
||||
"""For compatability with the gda's dummy_pd module"""
|
||||
pass
|
||||
|
||||
|
||||
class MultiInputExtraFieldsDummyScannable(ScannableBase):
|
||||
'''Multi input Dummy PD Class supporting input and extra fields'''
|
||||
def __init__(self, name, inputNames, extraNames):
|
||||
self.setName(name)
|
||||
self.setInputNames(inputNames)
|
||||
self.setExtraNames(extraNames)
|
||||
self.setOutputFormat(['%6.4f'] * (len(inputNames) + len(extraNames)))
|
||||
self.setLevel(3)
|
||||
self.currentposition = [0.0] * len(inputNames)
|
||||
|
||||
def isBusy(self):
|
||||
return 0
|
||||
|
||||
def asynchronousMoveTo(self, new_position):
|
||||
if type(new_position) == type(1) or type(new_position) == type(1.0):
|
||||
new_position = [new_position]
|
||||
msg = "Wrong new_position size"
|
||||
assert len(new_position) == len(self.currentposition), msg
|
||||
for i in range(len(new_position)):
|
||||
if new_position[i] != None:
|
||||
self.currentposition[i] = float(new_position[i])
|
||||
|
||||
def getPosition(self):
|
||||
extraValues = range(100, 100 + (len(self.getExtraNames())))
|
||||
return self.currentposition + map(float, extraValues)
|
||||
|
||||
|
||||
class ZeroInputExtraFieldsDummyScannable(ScannableBase):
|
||||
'''Zero input/extra field dummy pd
|
||||
'''
|
||||
def __init__(self, name):
|
||||
self.setName(name)
|
||||
self.setInputNames([])
|
||||
self.setOutputFormat([])
|
||||
|
||||
def isBusy(self):
|
||||
return 0
|
||||
|
||||
def asynchronousMoveTo(self, new_position):
|
||||
pass
|
||||
|
||||
def getPosition(self):
|
||||
pass
|
||||
|
||||
|
||||
class ScannableGroup(ScannableBase):
|
||||
"""wraps up motors. Simulates motors if non given."""
|
||||
|
||||
def __init__(self, name, motorList):
|
||||
|
||||
self.setName(name)
|
||||
# Set input format
|
||||
motorNames = []
|
||||
for scn in motorList:
|
||||
motorNames.append(scn.getName())
|
||||
self.setInputNames(motorNames)
|
||||
# Set output format
|
||||
format = []
|
||||
for motor in motorList:
|
||||
format.append(motor.getOutputFormat()[0])
|
||||
self.setOutputFormat(format)
|
||||
self.__motors = motorList
|
||||
|
||||
def asynchronousMoveTo(self, position):
|
||||
# if input has any Nones, then replace these with the current positions
|
||||
if None in position:
|
||||
position = list(position)
|
||||
current = self.getPosition()
|
||||
for idx, val in enumerate(position):
|
||||
if val is None:
|
||||
position[idx] = current[idx]
|
||||
|
||||
for scn, pos in zip(self.__motors, position):
|
||||
scn.asynchronousMoveTo(pos)
|
||||
|
||||
def getPosition(self):
|
||||
return [scn.getPosition() for scn in self.__motors]
|
||||
|
||||
def isBusy(self):
|
||||
for scn in self.__motors:
|
||||
if scn.isBusy():
|
||||
return True
|
||||
return False
|
||||
|
||||
def configure(self):
|
||||
pass
|
||||
|
||||
|
||||
class ScannableMotionWithScannableFieldsBase(ScannableBase):
|
||||
'''
|
||||
This extended version of ScannableMotionBase contains a
|
||||
completeInstantiation() method which adds a dictionary of
|
||||
MotionScannableParts to an instance. Each part allows one of the
|
||||
instances fields to be interacted with like it itself is a scannable.
|
||||
Fields are dynamically added to the instance linking to these parts
|
||||
allowing dotted access from Jython. They may also be accessed using
|
||||
Jython container access methods (via the __getitem__() method). To acess
|
||||
them from Jave use the getComponent(name) method.
|
||||
|
||||
When moving a part (via either a pos or scan command), the part calls
|
||||
the parent to perform the actual task. The parts asynchronousMoveto
|
||||
command will call the parent with a list of None values except for the
|
||||
field it represents which will be passed the desired position value.
|
||||
|
||||
The asynchronousMoveTo method in class that inherats from this base
|
||||
class then must handle these Nones. In some cases the method may
|
||||
actually be able to move the underlying system assoiciated with one
|
||||
field individually from others. If this is not possible the best
|
||||
behaviour may be to simply not support this beahviour and exception or
|
||||
alternatively to substitute the None values with actual current position
|
||||
of parent's scannables associated fields.
|
||||
|
||||
ScannableMotionBaseWithMemory() inherats from this calss and provides a
|
||||
solution useful for some scenarious: it keeps track of the last position
|
||||
moved to, and replaces the Nones in an asynchronousMoveTo request with
|
||||
these values. There are a number of dangers associated with this which
|
||||
are addressed in that class's documentation, but it provides a way to
|
||||
move one axis within a group of non-orthogonal axis while keeping the
|
||||
others still.
|
||||
'''
|
||||
childrenDict = {}
|
||||
numInputFields = None
|
||||
numExtraFields = None
|
||||
|
||||
def completeInstantiation(self):
|
||||
'''This method should be called at the end of all user defined
|
||||
consructors'''
|
||||
# self.validate()
|
||||
self.numInputFields = len(self.getInputNames())
|
||||
self.numExtraFields = len(self.getExtraNames())
|
||||
self.addScannableParts()
|
||||
self.autoCompletePartialMoveToTargets = False
|
||||
self.positionAtScanStart = None
|
||||
|
||||
def setAutoCompletePartialMoveToTargets(self, b):
|
||||
self.autoCompletePartialMoveToTargets = b
|
||||
|
||||
def atScanStart(self):
|
||||
self.positionAtScanStart = self.getPosition()
|
||||
|
||||
def atCommandFailure(self):
|
||||
self.positionAtScanStart = None
|
||||
|
||||
def atScanEnd(self):
|
||||
self.positionAtScanStart = None
|
||||
|
||||
###
|
||||
|
||||
def __repr__(self):
|
||||
pos = self.getPosition()
|
||||
formattedValues = self.formatPositionFields(pos)
|
||||
if len(tuple(self.getInputNames()) + tuple(self.getExtraNames())) > 1:
|
||||
result = self.getName() + ': '
|
||||
else:
|
||||
result = ''
|
||||
|
||||
names = tuple(self.getInputNames()) + tuple(self.getExtraNames())
|
||||
for name, val in zip(names, formattedValues):
|
||||
result += ' ' + name + ': ' + val
|
||||
return result
|
||||
###
|
||||
|
||||
def formatPositionFields(self, pos):
|
||||
"""Returns position as array of formatted strings"""
|
||||
# Make sure pos is a tuple or list
|
||||
if type(pos) not in (tuple, list):
|
||||
pos = tuple([pos])
|
||||
|
||||
# Sanity check
|
||||
if len(pos) != len(self.getOutputFormat()):
|
||||
raise Exception(
|
||||
"In scannable '%s':number of position fields differs from "
|
||||
"number format strings specified" % self.getName())
|
||||
|
||||
result = []
|
||||
for field, format in zip(pos, self.getOutputFormat()):
|
||||
if field is None:
|
||||
result.append('???')
|
||||
else:
|
||||
s = (format % field)
|
||||
## if width!=None:
|
||||
## s = s.ljust(width)
|
||||
result.append(s)
|
||||
|
||||
return result
|
||||
|
||||
###
|
||||
|
||||
def addScannableParts(self):
|
||||
'''
|
||||
Creates an array of MotionScannableParts each of which allows access to
|
||||
the scannable's fields. See this class's documentation for more info.
|
||||
'''
|
||||
self.childrenDict = {}
|
||||
# Add parts to access the input fields
|
||||
for index in range(len(self.getInputNames())):
|
||||
scannableName = self.getInputNames()[index]
|
||||
self.childrenDict[scannableName] = self.MotionScannablePart(
|
||||
scannableName, index, self, isInputField=1)
|
||||
|
||||
# Add parts to access the extra fields
|
||||
for index in range(len(self.getExtraNames())):
|
||||
scannableName = self.getExtraNames()[index]
|
||||
self.childrenDict[scannableName] = self.MotionScannablePart(
|
||||
scannableName, index + len(self.getInputNames()),
|
||||
self, isInputField=0)
|
||||
|
||||
def asynchronousMoveTo(self, newpos):
|
||||
if self.autoCompletePartialMoveToTargets:
|
||||
newpos = self.completePosition(newpos)
|
||||
ScannableBase.asynchronousMoveTo(self, newpos)
|
||||
|
||||
def completePosition(self, position):
|
||||
'''
|
||||
If position contains any null or None values, these are replaced with
|
||||
the corresponding fields from the scannables current position and then
|
||||
returned.'''
|
||||
# Just return position if it does not need padding
|
||||
if None not in position:
|
||||
return position
|
||||
if self.positionAtScanStart is not None:
|
||||
basePosition = self.positionAtScanStart
|
||||
else:
|
||||
basePosition = self.getPosition()[:self.numInputFields]
|
||||
for i in range(self.numInputFields):
|
||||
if position[i] is None:
|
||||
position[i] = basePosition[i]
|
||||
return position
|
||||
|
||||
def __getattr__(self, name):
|
||||
try:
|
||||
return self.childrenDict[name]
|
||||
except:
|
||||
raise AttributeError("No child named:" + name)
|
||||
|
||||
def __getitem__(self, key):
|
||||
'''Provides container like access from Jython'''
|
||||
return self.childrenDict[key]
|
||||
|
||||
def getPart(self, name):
|
||||
'''Returns the a compnent scannable'''
|
||||
return self.childrenDict[name]
|
||||
|
||||
class MotionScannablePart(ScannableBase):
|
||||
'''
|
||||
A scannable to be placed in the parent's childrenDict that allows
|
||||
access to the parent's individual fields.'''
|
||||
|
||||
def __init__(self, scannableName, index, parentScannable,
|
||||
isInputField):
|
||||
self.setName(scannableName)
|
||||
if isInputField:
|
||||
self.setInputNames([scannableName])
|
||||
else:
|
||||
self.setExtraNames([scannableName])
|
||||
self.index = index
|
||||
self.parentScannable = parentScannable
|
||||
self.setOutputFormat(
|
||||
[self.parentScannable.getOutputFormat()[index]])
|
||||
|
||||
def isBusy(self):
|
||||
return self.parentScannable.isBusy()
|
||||
|
||||
def asynchronousMoveTo(self, new_position):
|
||||
if self.parentScannable.isBusy():
|
||||
raise Exception(
|
||||
self.parentScannable.getName() + "." + self.getName() +
|
||||
" cannot be moved because " +
|
||||
self.parentScannable.getName() + " is already moving")
|
||||
|
||||
toMoveTo = [None] * len(self.parentScannable.getInputNames())
|
||||
toMoveTo[self.index] = new_position
|
||||
self.parentScannable.asynchronousMoveTo(toMoveTo)
|
||||
|
||||
def moveTo(self, new_position):
|
||||
self.asynchronousMoveTo(new_position)
|
||||
self.waitWhileBusy()
|
||||
|
||||
def getPosition(self):
|
||||
return self.parentScannable.getPosition()[self.index]
|
||||
|
||||
def __str__(self):
|
||||
return self.__repr__()
|
||||
|
||||
def __repr__(self):
|
||||
# Get the name of this field
|
||||
# (assume its an input field first and correct if wrong)
|
||||
name = self.getInputNames()[0]
|
||||
|
||||
if name == 'value':
|
||||
name = self.getExtraNames()[0]
|
||||
parentName = self.parentScannable.getName()
|
||||
return parentName + "." + name + " : " + str(self.getPosition())
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user