csaxs_bec.devices.epics.specMotors.PmMonoBender#

class PmMonoBender(prefix='', *, concurrent=True, read_attrs=None, configuration_attrs=None, name, egu='', auto_target=True, **kwargs)[source]#

Bases: PseudoPositioner

Monochromator bender

Small wrapper to combine the four monochromator bender motors.

Methods

`add_instantiation_callback`	Register a callback which will receive each OphydObject instance.
`check_single`	Check if a new position for a single pseudo positioner is valid
`check_value`	Check if a new position for all pseudo positioners is valid
`clear_sub`	Remove a subscription, given the original callback function
`configure`	Configure the device for something during a run
`describe`	Provide schema and meta-data for `read()`.
`describe_configuration`	Provide schema & meta-data for `read_configuration()`
`destroy`	Disconnect and destroy all signals on the Device
`forward`	Calculate a RealPosition from a given PseudoPosition
`get`	Get the value of all components in the device
`get_device_tuple`	The device tuple type associated with an Device class
`get_instantiated_signals`	Yields all of the instantiated signals in a device hierarchy
`inverse`	Calculate a PseudoPosition from a given RealPosition
`move`	Move to a specified position, optionally waiting for motion to complete.
`move_single`	Move one PseudoSingle axis to a position
`pause`	Attempt to 'pause' the device.
`put`	Put a value to all components of the device
`read`	Read data from the device.
`read_configuration`	Dictionary mapping names to value dicts with keys: value, timestamp
`resume`	Resume a device from a 'paused' state.
`set`	Move to a new position asynchronously
`set_defaults`	Set class-wide defaults for device communications
`stage`	Stage the device for data collection.
`stop`	Stop the Device and all (instantiated) subdevices
`subscribe`	Subscribe to events this event_type generates.
`summary`
`to_pseudo_tuple`	Convert arguments to a PseudoPosition namedtuple and kwargs
`to_real_tuple`	Convert arguments to a RealPosition namedtuple and kwargs
`trigger`	Trigger the device and return status object.
`unstage`	Unstage the device.
`unsubscribe`	Remove a subscription
`unsubscribe_all`
`wait_for_connection`	Wait for signals to connect
`walk_components`	Walk all components in the Device hierarchy
`walk_signals`	Walk all signals in the Device hierarchy
`walk_subdevice_classes`	Walk all sub-Devices classes in the Device hierarchy
`walk_subdevices`	Walk all sub-Devices in the hierarchy

Attributes

`SUB_ACQ_DONE`
`SUB_DONE`
`SUB_READBACK`
`SUB_START`
`ai`	A descriptor representing a device component (or signal)
`attr_name`
`bend`	A descriptor representing a device component (or signal)
`bo`	A descriptor representing a device component (or signal)
`co`	A descriptor representing a device component (or signal)
`component_names`
`composite_egu`	The composite engineering units (EGU) from all PseudoSingles
`concurrent`	If concurrent is set, motors will move concurrently (in parallel)
`configuration_attrs`
`connected`	If the device is connected.
`connection_timeout`
`di`	A descriptor representing a device component (or signal)
`dotted_name`	Return the dotted name
`egu`	The engineering units (EGU) for positions
`event_types`	Events that can be subscribed to via `obj.subscribe`
`high_limit`	All PseudoSingle high limits as a namedtuple
`hints`
`kind`
`lazy_wait_for_connection`
`limits`	All PseudoSingle limits as a namedtuple
`low_limit`	All PseudoSingle low limits as a namedtuple
`moving`	Whether or not the motor is moving
`name`	name of the device
`parent`	The parent of the ophyd object.
`position`	Pseudo motor position namedtuple
`pseudo_positioners`	Pseudo positioners instances in a namedtuple
`read_attrs`
`real_position`	Real motor position namedtuple
`real_positioners`	Real positioners instances in a namedtuple
`report`	A report on the object.
`root`	Walk parents to find ultimate ancestor (parent's parent...).
`sequential`	If sequential is set, motors will move in the sequence they were defined in (i.e., in series)
`settle_time`	Amount of time to wait after moves to report status completion
`signal_names`
`subscriptions`
`target`	Last commanded target positions
`timeout`	Amount of time to wait before to considering a motion as failed
`trigger_signals`

class OphydAttrList(device, kind, remove_kind, recurse_key)#

Bases: MutableSequence

list proxy to migrate away from Device.read_attrs and Device.config_attrs

append(value)#: S.append(value) – append value to the end of the sequence

clear() → None -- remove all items from S#

count(value) → integer -- return number of occurrences of value#

extend(values)#: S.extend(iterable) – extend sequence by appending elements from the iterable

index(value[, start[, stop]]) → integer -- return first index of value.#

Raises ValueError if the value is not present.

Supporting start and stop arguments is optional, but recommended.

insert(index, object)#: S.insert(index, value) – insert value before index

pop([index]) → item -- remove and return item at index (default last).#: Raise IndexError if list is empty or index is out of range.

remove(value)#: S.remove(value) – remove first occurrence of value. Raise ValueError if the value is not present.

reverse()#: S.reverse() – reverse IN PLACE

classmethod add_instantiation_callback(callback, fail_if_late=False)#

Register a callback which will receive each OphydObject instance.

Parameters:

callback (callable) – Expected signature: f(ophydobj_instance)
fail_if_late (boolean) – If True, verify that OphydObj has not yet been instantiated and raise RuntimeError if it has, as a way of verify that no instances will be “missed” by this registry. False by default.

check_single(pseudo_single, single_pos)#: Check if a new position for a single pseudo positioner is valid

check_value(pseudo_pos)#

Check if a new position for all pseudo positioners is valid

First checks limits against those set for individual pseudo axes. Second, calculates forward(pseudo_pos) => real_pos and checks it against the real positioners.

NOTE: If you have limits that are coupled together or are somehow more complicated than the above procedure, you should redefine this method in your subclass.

clear_sub(cb, event_type=None)#

Remove a subscription, given the original callback function

See also subscribe(), unsubscribe()

Parameters:

cb (callable) – The callback
event_type (str, optional) – The event to unsubscribe from (if None, removes it from all event types)

property composite_egu#: The composite engineering units (EGU) from all PseudoSingles

property concurrent#: If concurrent is set, motors will move concurrently (in parallel)

configure(d: Dict[str, Any]) → Tuple[Dict[str, Any], Dict[str, Any]]#

Configure the device for something during a run

This default implementation allows the user to change any of the configuration_attrs. Subclasses might override this to perform additional input validation, cleanup, etc.

Parameters:

d (dict) – The configuration dictionary. To specify the order that the changes should be made, use an OrderedDict.

Returns:

(old, new) tuple of dictionaries
Where old and new are pre- and post-configure configuration states.

property connected#

If the device is connected.

Subclasses should override this

describe() → OrderedDictType[str, Dict[str, Any]]#

Provide schema and meta-data for read().

This keys in the OrderedDict this method returns must match the keys in the OrderedDict return by read().

This provides schema related information, (ex shape, dtype), the source (ex PV name), and if available, units, limits, precision etc.

Returns:: data_keys – The keys must be strings and the values must be dict-like with the event_model.event_descriptor.data_key schema.
Return type:: OrderedDict

describe_configuration() → OrderedDictType[str, Dict[str, Any]]#

Provide schema & meta-data for read_configuration()

This keys in the OrderedDict this method returns must match the keys in the OrderedDict return by read().

This provides schema related information, (ex shape, dtype), the source (ex PV name), and if available, units, limits, precision etc.

Returns:: data_keys – The keys must be strings and the values must be dict-like with the event_model.event_descriptor.data_key schema.
Return type:: OrderedDict

destroy()#: Disconnect and destroy all signals on the Device

property dotted_name: str#: Return the dotted name

property egu#: The engineering units (EGU) for positions

property event_types#: Events that can be subscribed to via obj.subscribe

forward(pseudo_pos)[source]#

Calculate a RealPosition from a given PseudoPosition

Must be defined on the subclass.

Parameters:: pseudo_pos (PseudoPosition) – The pseudo position input
Returns:: real_position – The real position output
Return type:: RealPosition

get(**kwargs)#

Get the value of all components in the device

Keyword arguments are passed onto each signal.get(). Components beginning with an underscore will not be included.

classmethod get_device_tuple()#

The device tuple type associated with an Device class

This is a tuple representing the full state of all components and dynamic device sub-components.

get_instantiated_signals(*, attr_prefix=None)#

Yields all of the instantiated signals in a device hierarchy

Parameters:: attr_prefix (string, optional) – The attribute prefix. If None, defaults to self.name
Yields:: (fully_qualified_attribute_name, signal_instance)

property high_limit#: All PseudoSingle high limits as a namedtuple

inverse(real_pos)[source]#

Calculate a PseudoPosition from a given RealPosition

Must be defined on the subclass.

Parameters:: real_position (RealPosition) – The real position input
Returns:: pseudo_pos – The pseudo position output
Return type:: PseudoPosition

property limits#: All PseudoSingle limits as a namedtuple

property low_limit#: All PseudoSingle low limits as a namedtuple

move(position, wait=True, timeout=None, moved_cb=None)#

Move to a specified position, optionally waiting for motion to complete.

Parameters:

position – Position to move to
moved_cb (callable) – Call this callback when movement has finished. This callback must accept one keyword argument: ‘obj’ which will be set to this positioner instance.
wait (bool, optional) – Wait until motion has completed
timeout (float, optional) – Maximum time to wait for a motion

Returns:

status

Return type:

MoveStatus

Raises:

TimeoutError – When motion takes longer than timeout
ValueError – On invalid positions
RuntimeError – If motion fails other than timing out

move_single(pseudo, position, **kwargs)#

Move one PseudoSingle axis to a position

All other positioners will use their current setpoint/target value, if available. Failing that, their current readback value will be used (see PseudoSingle.sync and PseudoSingle.target).

Parameters:

pseudo (PseudoSingle) – PseudoSingle positioner to move
position (float) – Position only for the PseudoSingle
kwargs (dict) – Passed onto move

property moving#

Whether or not the motor is moving

Returns:: moving
Return type:: bool

property name#: name of the device

property parent#

The parent of the ophyd object.

If at the top of its hierarchy, parent will be None

pause() → None#

Attempt to ‘pause’ the device.

This is called when ever the RunEngine is interrupted.

A device may have internal state that means plans can not safely be re-wound. This method may: put the device in a ‘paused’ state and/or raise NoReplayAllowed to indicate that the plan can not be rewound.

Raises:: bluesky.run_engine.NoReplayAllowed –

property position#: Pseudo motor position namedtuple

property pseudo_positioners#

Pseudo positioners instances in a namedtuple

Returns:: positioner_instances
Return type:: PseudoPosition

put(dev_t, **kwargs)#

Put a value to all components of the device

Keyword arguments are passed onto each signal.put()

Parameters:: dev_t (DeviceTuple or tuple) – The device tuple with the value(s) to put (see get_device_tuple)

read() → OrderedDictType[str, Dict[str, Any]]#

Read data from the device.

This method is expected to be as instantaneous as possible, with any substantial acquisition time taken care of in trigger().

The OrderedDict returned by this method must have identical keys (in the same order) as the OrderedDict returned by describe().

By convention, the first key in the return is the ‘primary’ key and maybe used by heuristics in bluesky.

The values in the ordered dictionary must be dict (-likes) with the keys {'value', 'timestamp'}. The 'value' may have any type, the timestamp must be a float UNIX epoch timestamp in UTC.

Returns:: data – The keys must be strings and the values must be dict-like with the keys {'value', 'timestamp'}
Return type:: OrderedDict

read_configuration() → OrderedDictType[str, Dict[str, Any]]#

Dictionary mapping names to value dicts with keys: value, timestamp

To control which fields are included, change the Component kinds on the device, or modify the configuration_attrs list.

property real_position#: Real motor position namedtuple

property real_positioners#

Real positioners instances in a namedtuple

Returns:: positioner_instances
Return type:: RealPosition

property report#: A report on the object.

resume() → None#

Resume a device from a ‘paused’ state.

This is called by the bluesky.run_engine.RunEngine when it resumes from an interruption and is responsible for ensuring that the device is ready to take data again.

property root#: Walk parents to find ultimate ancestor (parent’s parent…).

property sequential#: If sequential is set, motors will move in the sequence they were defined in (i.e., in series)

set(position, **kwargs)#

Move to a new position asynchronously

Parameters:: position (PseudoPosition) – Position for the all of the pseudo axes
Returns:: status
Return type:: MoveStatus

classmethod set_defaults(*, connection_timeout=10.0)#

Set class-wide defaults for device communications

This may be called only before any instances of Device are made.

This setting applies to the class it is called on and all its subclasses. For example,

>>> Device.set_defaults(...)

will apply to any Device subclass.

Parameters:: connection_timeout (float, optional) – Time (seconds) allocated for establishing a connection with the IOC.
Raises:: RuntimeError – If called after EpicsSignalBase has been instantiated for the first time.

property settle_time#: Amount of time to wait after moves to report status completion

stage() → List[object]#

Stage the device for data collection.

This method is expected to put the device into a state where repeated calls to trigger() and read() will ‘do the right thing’.

Staging not idempotent and should raise RedundantStaging if staged twice without an intermediate unstage().

This method should be as fast as is feasible as it does not return a status object.

The return value of this is a list of all of the (sub) devices stage, including it’s self. This is used to ensure devices are not staged twice by the RunEngine.

This is an optional method, if the device does not need staging behavior it should not implement stage (or unstage).

Returns:: devices – list including self and all child devices staged
Return type:: list

stop(success=False)#: Stop the Device and all (instantiated) subdevices

subscribe(callback, event_type=None, run=True)#

Subscribe to events this event_type generates.

The callback will be called as cb(*args, **kwargs) with the values passed to _run_subs with the following additional keys:

sub_type : the string value of the event_type obj : the host object, added if ‘obj’ not already in kwargs

if the key ‘timestamp’ is in kwargs _and_ is None, then it will be replaced with the current time before running the callback.

The *args, **kwargs passed to _run_subs will be cached as shallow copies, be aware of passing in mutable data.

Warning

If the callback raises any exceptions when run they will be silently ignored.

Parameters:

callback (callable) –
A callable function (that takes kwargs) to be run when the event is generated. The expected signature is
```
def cb(*args, obj: OphydObject, sub_type: str, **kwargs) -> None:
```
The exact args/kwargs passed are whatever are passed to _run_subs
event_type (str, optional) –
The name of the event to subscribe to (if None, defaults to the default sub for the instance - obj._default_sub)

This maps to the sub_type kwargs in _run_subs
run (bool, optional) – Run the callback now