csaxs_bec.devices.epics.delay_generator_csaxs.ddg_1.DDG1#

class DDG1(name: str, prefix: str = '', scan_info: ScanInfo | None = None, device_manager: DeviceManagerBase | None = None, **kwargs)[source]#

Bases: PSIDeviceBase, DelayGeneratorCSAXS

Implementation of the DelayGenerator DDG1 for the cSAXS beamline. It is the main trigger source for the cSAXS beamline, and will be triggered by BEC through a software trigger or by a hardware trigger from a beamline device (e.g. Galil stages). Specific implementation of the cabling logic expected for this device are described in the module README, the attached PDF ‘trigger_scheme_ddg1_ddg2.pdf’ and the module docstring.

The IOC prefix is ‘X12SA-CPCL-DDG1:’.

Parameters:

name (str) – Name of the device.
prefix (str, optional) – EPICS prefix for the device. Defaults to ‘’.
scan_info (ScanInfo | None, optional) – Scan info object. Defaults to None.
device_manager (DeviceManagerBase | None, optional) – Device manager. Defaults to None.

Initialize the PSI Device Base class.

Parameters:

name (str) – Name of the device
scan_info (ScanInfo) – The scan info to use.

Methods

`add_instantiation_callback`	Register a callback which will receive each OphydObject instance.
`burst_disable`	Disable burst mode
`burst_enable`	Enable burst mode with valid parameters.
`cancel_on_stop`	Register a status object to be cancelled when the device is stopped.
`check_value`	Check if the value is valid for this object
`clear_sub`	Remove a subscription, given the original callback function
`complete`	Complete the device.
`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`	Destroy the device.
`get`	Get the value of all components in the device
`get_device_tuple`	The device tuple type associated with an Device class
`get_error_msg`	Get the error message from the delay generator.
`get_instantiated_signals`	Yields all of the instantiated signals in a device hierarchy
`get_shutter_to_open_delay`	Get the current delay that is set to open the shutter before the exposure time.
`keep_shutter_open_during_scan`	Method to configure the delay generator for keeping the shutter open during a scans.
`kickoff`	Kickoff the device.
`on_complete`	Called to inquire if a device has completed a scans.
`on_connected`	This method is called after the device is initialized and all signals are connected.
`on_destroy`	Clean up resources when the device is destroyed.
`on_init`	Called when the device is initialized.
`on_kickoff`	Called to kickoff a device for a fly scan.
`on_pre_scan`	Called right before the scan starts on all devices automatically.
`on_stage`	This method is called in preparation for a scan.
`on_stop`	Stop the delay generator HW and polling thread when the device is stopped.
`on_trigger`	This method is called from BEC as a software trigger.
`on_unstage`	Called while unstaging the device.
`pause`	Attempt to 'pause' the device.
`pre_scan`	Pre-scan function.
`prepare_mcs_on_trigger`	This method is used by the DDG1 on_trigger method to prepare the MCS card for the next trigger.
`proc_event_status`	The reading must be manually triggered to update the event status.
`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
`reset_error`	Reset the error status message of the delay generator.
`resume`	Resume a device from a 'paused' state.
`set_channel_reference`	Set the reference channel for a specific channel.
`set_defaults`	Set class-wide defaults for device communications
`set_delay_pairs`	Set the delay and width for a specific channel pair.
`set_io_values`	Set the IO values for the static pair.
`set_references_for_channels`	Set the reference channels for multiple channels.
`set_trigger`	Set the trigger source.
`stage`	Stage the device.
`stop`	Stop the device.
`stop_ddg`	Stop the delay generator by setting the burst mode to 0
`subscribe`	Subscribe to events this event_type generates.
`summary`
`trigger`	Trigger the device.
`unstage`	Unstage the device.
`unsubscribe`	Remove a subscription
`unsubscribe_all`
`wait_for_condition`	Utility method to easily wait for signals or methods to reach an expected state.
`wait_for_connection`	Wait for signals to connect
`wait_for_event_status`	Wait for a specific event status.
`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_DEVICE_MONITOR_1D`
`SUB_DEVICE_MONITOR_2D`
`SUB_DONE_MOVING`
`SUB_FILE_EVENT`
`SUB_MOTOR_IS_MOVING`
`SUB_PROGRESS`
`SUB_READBACK`
`SUB_VALUE`
`USER_ACCESS`
`ab`	A descriptor representing a device component (or signal)
`attr_name`
`burst_config`	A descriptor representing a device component (or signal)
`burst_count`	A descriptor representing a device component (or signal)
`burst_delay`	A descriptor representing a device component (or signal)
`burst_mode`	A descriptor representing a device component (or signal)
`burst_period`	A descriptor representing a device component (or signal)
`cd`	A descriptor representing a device component (or signal)
`component_names`
`configuration_attrs`
`connected`	If the device is connected.
`connection_timeout`
`destroyed`	Check if the device has been destroyed.
`dotted_name`	Return the dotted name
`ef`	A descriptor representing a device component (or signal)
`event_types`	Events that can be subscribed to via `obj.subscribe`
`fast_shutter_control`	A descriptor representing a device component (or signal)
`fast_shutter_readback`	A descriptor representing a device component (or signal)
`gh`	A descriptor representing a device component (or signal)
`hints`
`kind`
`lazy_wait_for_connection`
`name`	name of the device
`parent`	The parent of the ophyd object.
`read_attrs`
`report`	A report on the object.
`root`	Walk parents to find ultimate ancestor (parent's parent...).
`signal_names`
`staged`	Check if the device has been staged.
`state`	A descriptor representing a device component (or signal)
`status_msg`	A descriptor representing a device component (or signal)
`status_msg_clear`	A descriptor representing a device component (or signal)
`stopped`	Check if the device has been stopped.
`subscriptions`
`t0`	A descriptor representing a device component (or signal)
`trigger_holdoff`	A descriptor representing a device component (or signal)
`trigger_inhibit`	A descriptor representing a device component (or signal)
`trigger_level`	A descriptor representing a device component (or signal)
`trigger_rate`	A descriptor representing a device component (or signal)
`trigger_shot`	A descriptor representing a device component (or signal)
`trigger_signals`
`trigger_source`	A descriptor representing a device component (or signal)

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.

burst_disable() → None#: Disable burst mode

burst_enable(count: int, delay: float, period: float, config: Literal['all', 'first'] | BURSTCONFIG = 'first') → None#

Enable burst mode with valid parameters.

Parameters:

count (int) – Number of bursts >0
delay (float) – Delay before bursts start in seconds >=0
period (float) – Period of the bursts in seconds >0
config (str) – Configuration of T0 duiring burst. In addition, to simplify triggering of other instruments synchronously with the burst, the T0 output may be configured to fire on the first delay cycle of the burst, rather than for all delay cycles as is normally the case. BURSTCONFIG

cancel_on_stop(status: StatusBase) → None#

Register a status object to be cancelled when the device is stopped.

Parameters:: status (StatusBase) – The status object to be cancelled.

check_value(value, **kwargs)#

Check if the value is valid for this object

This function does no normalization, but may raise if the value is invalid.

Raises:: ValueError –

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)

complete() → DeviceStatus | StatusBase#: Complete the device.

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()#: Destroy the device.

property destroyed: bool#: Check if the device has been destroyed.

property dotted_name: str#: Return the dotted name

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

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_error_msg() → str#: Get the error message from the delay generator.

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)

get_shutter_to_open_delay() → float[source]#: Get the current delay that is set to open the shutter before the exposure time.

keep_shutter_open_during_scan(open: True) → None[source]#

Method to configure the delay generator for keeping the shutter open during a scans. This means that the additional delay to open the shutter needs to be removed (2e-3) from the timing of the signals.

Parameters:: open (bool) – If True, the shutter will be kept open during the scan. If False, the shutter will be opened and closed for each trigger cycle.

kickoff() → DeviceStatus | StatusBase#: Kickoff the device.

property name#: name of the device

on_complete() → DeviceStatus | StatusBase | None#: Called to inquire if a device has completed a scans.

on_connected() → None[source]#

This method is called after the device is initialized and all signals are connected. This happens when a device configuration is loaded in BEC.

It sets the default values for this device - intended to overwrite everything to a usable default state. For this purpose, we use the DEFAULT SETTINGS defined at the top of this module.

To ensure that this process is robust, we follow these steps: - First, we stop any ongoing burst mode operation. - Then, we set the DEFAULT_IO_CONFIG for each channel, the trigger source to DEFAULT_TRIGGER_SOURCE,

and the channel references to DEFAULT_REFERENCES.

We set the state proc_status to be event based. This triggers readouts of the EventStatusLI bit based on events. This was empirically found to be a stable solution in combination with the poll loop of the state.
Finally, we set the burst delay to 0, to set it to be of no delay.

on_destroy() → None[source]#: Clean up resources when the device is destroyed.

on_init() → None#

Called when the device is initialized.

No signals are connected at this point. If you like to set default values on signals, please use on_connected instead.

on_kickoff() → DeviceStatus | StatusBase | None#: Called to kickoff a device for a fly scan. Has to be called explicitly.

on_pre_scan() → DeviceStatus | StatusBase | None#: Called right before the scan starts on all devices automatically.

on_stage() → None[source]#

This method is called in preparation for a scan. All information about the upcoming scan is available in self.scan_info.msg at this point. We use this information to configure the DDG1 for the upcoming scan.

The DDG is operated in burst mode for the scan, but with only a single burst pulse. THe length of the pulse is set to the expected exposure time for a single trigger, which includes any burst acquisitions if frames_per_trigger > 1.

The logic is as follows: - We check if any default burst parameters need to be set, and set them if needed. - We calculate the burst pulse width based on the exposure time and frames_per_trigger. - We set the burst_period and the shutter signal (delay pairs cd) to be

exposure_time * frames_per_trigger + 3ms (2ms for shutter to open, 1ms to close).

We set the delay pairs ab to be 2ms delayed (to allow the shutter to open) with a width of 1us to trigger DDG2.
We set the delay pairs ef to be triggered after the shutter closes with a width of 1us to trigger the MCS card.
Finally, we add a short sleep to ensure that the IOC and DDG HW process the values properly.

on_stop() → None[source]#: Stop the delay generator HW and polling thread when the device is stopped.

on_trigger() → DeviceStatus[source]#

This method is called from BEC as a software trigger. Here the logic is as follows:

We first check if the trigger_source is set to SINGLE_SHOT. Only then will we received, otherwise we return a status object directly as the DDG is triggered by an external source which will have to implement its own logic to wait for trigger signals to be received.

I SINGLE_SHOT, the implementation here will send a software trigger. Now there are two options to wait for the trigger (burst) cycle to be done. One is to rely on the signal of the “mcs” card if it is present. However, this is only possible if the scan_type is not “fly” as in fly scans the ef channel is not triggered to send the last pulse to the card (but the card is finishing its acquisition in complete itself). Then we rely on the polling of the event status register to check if the burst cycle is done.

It follows a specific procedure to ensure that the DDG1 and MCS card are properly handled on a trigger event. The established logic is as follows:

Stop polling the event status register to avoid that the polling loop is still active
before sending the software trigger. This needs to be done to avoid conflicts in reading the event status register.
Wait for the _poll_thread_poll_loop_done event to ensure that the polling loop is no
longer active. A timeout of 1s is plenty as sleeps of 20ms are used in the poll loop.
Add an extra sleep of 20ms to make sure that the HW is again ready to process new commands.
This has been found empirically after long testing to improve stability.
If the MCS card is present in the current session of BEC, prepare the card for the next trigger.
Prepare a status StatusBitsCompareStatus that will be resolved once the burst is done.
Start the polling loop again to monitor the event status register.
Send the software trigger to the DDG1
Return the status object to BEC which will automatically resolve once the status register has
the END_OF_BURST bit set. The callback of the status object will also stop the polling loop.

on_unstage() → DeviceStatus | StatusBase | None#: Called while unstaging 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 –

pre_scan() → DeviceStatus | StatusBase | None#: Pre-scan function.

prepare_mcs_on_trigger() → CompareStatus[source]#

This method is used by the DDG1 on_trigger method to prepare the MCS card for the next trigger. It checks that the MCS card is properly prepared before BEC sends a software trigger to the DDG1, which is needed for step scans.

It relies on the MCS card implementation and needs to be adapted if the MCS card logic changes.

proc_event_status() → None#: The reading must be manually triggered to update the event status.

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 report#: A report on the object.

reset_error() → None#: Reset the error status message of the delay generator.

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…).

set_channel_reference(channel: Literal['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'], reference_channel: CHANNELREFERENCE)#

Set the reference channel for a specific channel.

Parameters:

channel (LiteralChannels) – The channel to set the reference for.
reference_channel (CHANNELREFERENCE) – The reference channel to set.

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.

Set the delay and width for a specific channel pair.

Parameters:

channel (str) – Channel pair to set the delay and width for. Can be “ab”, “cd”, “ef”, “gh”.
delay (float) – Delay in seconds to set for the channel pair.
width (float) – Width in seconds to set for the channel pair.

Set the IO values for the static pair.

Parameters:

channel (str | list[str]) – Channel(s) to set the IO values for. Can be “t0”, “ab”, “cd”, “ef”, “gh” or a list of these. If a list is provided, the same values will be set for all channels.
amplitude (float) – Amplitude of the output signal in volts.
offset (float) – Offset of the output signal in volts.
polarity (OUTPUTPOLARITY | int) – Polarity of the output signal.
ttl_mode (bool) – If True, set the output to TTL mode.
nim_mode (bool) – If True, set the output to NIM mode. If both ttl_mode and nim_mode are set to True, a ValueError is raised.

set_references_for_channels(channels_and_refs: list[tuple[Literal['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'], CHANNELREFERENCE]])#

Set the reference channels for multiple channels.

Parameters:: channels_and_refs (list[tuple[LiteralChannels, CHANNELREFERENCE]]) – A list of tuples where each tuple contains a channel and its corresponding reference channel.

set_trigger(source: TRIGGERSOURCE | int) → None#

Set the trigger source.

Parameters:: source (TriggerSource | int) – The trigger source INTERNAL = 0 EXT_RISING_EDGE = 1 EXT_FALLING_EDGE = 2 SS_EXT_RISING_EDGE = 3 SS_EXT_FALLING_EDGE = 4 SINGLE_SHOT = 5 LINE = 6

stage() → list[object] | DeviceStatus | StatusBase#: Stage the device.

property staged: Staged#: Check if the device has been staged.

stop(*, success: bool = False) → None#

Stop the device.

Parameters:: success (bool) – True if the action was successful, False otherwise.

stop_ddg() → None#: Stop the delay generator by setting the burst mode to 0

property stopped: bool#: Check if the device has been stopped.

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