Utility

Callbacks

Registry

Note

The functionality in this module can only be used if ophyd-registry is installed. If you installed the package via pip, either adding the registry extra, or using all ought to work.

sophys.common.utils.registry.create_named_registry(registry_name: str)[source]

Create a new Registry associated with registry_name, clearing an existing one if it does.

The created instance has auto_registrer set to False, so you must explicitly register devices to it. Check register_devices() for some convenience.

Returns the created / cleared Registry instance.

sophys.common.utils.registry.get_all_devices(as_dict: bool = False, include_components: bool = False, use_registry_name: bool = True, use_dotted_name: bool = True)[source]

Return all the devices from all the registries currently instantiated.

Parameters:

  • as_dict (bool, optional) – If True, return a dictionary, as per to_variable_dict(). Otherwise, return a list of all devices. Defaults to False.

  • include_components (bool, optional) – If True, return all registered devices and components.

  • use_registry_name (bool, optional) – If True, prepend the containing registry name to the device’s name in the dict key. Defaults to True.

  • use_dotted_name (bool, optional) – If True, use the dotted name, with hierarchical information, in the dict key. Otherwise, use name. Defaults to True.

sophys.common.utils.registry.get_all_registries()[source]

Return all the Registry instances currently instantiated.

sophys.common.utils.registry.get_all_root_devices(as_dict: bool = False, use_registry_name: bool = True)[source]

Return all the root devices from all the registries currently instantiated.

Parameters:

as_dict (bool, optional) – If True, return a dictionary, as per to_variable_dict(). Otherwise, return a list of all root devices. Defaults to False.

sophys.common.utils.registry.get_named_registry(registry_name: str, create_if_missing: bool = False)[source]

Get the instance of ophyd-registry’s Registry associated with register_name.

If create_if_missing is True, create a new Registry for that name if it is missing, otherwise raise a RuntimeError.

sophys.common.utils.registry.get_registry_name(registry)[source]

Return the name associated with this registry.

sophys.common.utils.registry.in_autoregister_context() → bool[source]

Returns whether the current execution context is auto registering new devices to a registry.

sophys.common.utils.registry.register_devices(registry_name: str)[source]

Context Manager for registering instantiated devices inside it to registry_name.

For instance, to register devices a and b to registry beamline_a, and devices c, d, and e to registry beamline_b, you could do something like:

# ...

with register_devices("beamline_a"):
    DeviceAClass(name="a")
    DeviceBClass(name="b")

with register_devices("beamline_b"):
    DeviceCClass(name="c")
    DeviceDClass(name="d")
    DeviceEClass(name="e")
sophys.common.utils.registry.remove_all_named(name: str)[source]

Remove all devices with a given name from all registries.

sophys.common.utils.registry.to_variable_dict(registries: Iterable, use_registry_name: bool = True)[source]

Turns a registry (or set of registries) into a dictionary, intended of being used as such:

  • globals().update(<return value>)

  • locals().update(<return value>)

Signals

class sophys.common.utils.signals.EpicsSignalCmd(prefix, **kwargs)[source]

Bases: EpicsSignal

A EpicsSignal with Cmd added to prefix.

class sophys.common.utils.signals.EpicsSignalMon(prefix, **kwargs)[source]

Bases: EpicsSignalRO

class sophys.common.utils.signals.EpicsSignalWithCustomReadout(read_pv, write_pv, enforce_type=None, **kwargs)[source]

Bases: EpicsSignal

An EpicsSignal subclass extending the validation of the result of a ‘set’ operation.

This is useful in cases where the readout of a particular ‘set’ is different from the value you set, like when making a command to an IOC, and expecting a “Done” string in return.

Parameters:

enforce_type (type, optional) – Whether to try to apply a type conversion to the readout value. If not set, defaults to not trying any type conversion (the default EpicsSignal behavior).

set(value, *, expected_readout=None, timeout=5.0, settle_time=None)[source]

Set the value of the Signal and return a Status object.

If put completion is used for this EpicsSignal, the status object will complete once EPICS reports the put has completed.

Otherwise the readback will be polled until equal to ‘expected_readout’, and if ‘enforce_type’ was set in the constructor, both of the values will be cast to that type, raising an Exception if the conversion is not possible.

Parameters:

  • value (any) –

  • expected_readout (any, optional) – Expected value of the ‘read_pv’ after successfully setting the value. If not set, defaults to ‘value’.

  • timeout (float, optional) – Maximum time to wait.

  • settle_time (float, optional) – Delay after the set() has completed to indicate completion to the caller

Returns:

st

Return type:

Status

See also

EpicsSignal.set

class sophys.common.utils.signals.EpicsSignalWithCustomReadoutRBV(write_pv, **kwargs)[source]

Bases: EpicsSignalWithCustomReadout

An EpicsSignalWithCustomReadout subclass setting the read_pv to ‘write_pv + _RBV’ by default.

class sophys.common.utils.signals.EpicsSignalWithRBSP(prefix, **kwargs)[source]

Bases: EpicsSignal

A simple Signal with a similar logic of EpicsSignalWithRBV, but pvname is -RB and write_pv is -SP.

sophys.common.utils.signals.add_components_to_device(obj: Device, components: Iterable[tuple[str, Component]], *, for_each_sig: Callable | None = None)[source]

Add a collection of components to a device, after it has been initialized.

Parameters:

  • obj (Device) – The device to which the components will be added.

  • components (iterable of (component name, component) tuples) – The components that will be added to obj.

  • for_each_sig (callable, optional) –

    Callback that is called on each signal addition, with signature (name: str, sig: Signal) -> Any.

    By default, it does nothing.

    One common usage is to call setattr of the signal to its parent.

Examples

Add four signals named channel_{x} to obj, each with prefix CH{x}:

components = (
    (
        f"channel_{i}",
        Component(EpicsSignal, f"CH{i}:"),
    )
    for i in range(1, 5)
)
add_components_to_device(
    obj, components, for_each_sig=lambda name, sig: setattr(self, name, sig)
)

Add an arbitrary number of signals named scale_{x} to obj, each with prefix SC{x}, while also adding them to a list scales:

def for_each_sig(name, sig):
    setattr(self, name, sig)
    self.scales.append(sig)

n_scales = 8
components = (
    (f"scale_{i}", Component(Scale, f"SC{i}:")) for i in range(n_scales)
)
add_components_to_device(self, components, for_each_sig=for_each_sig)

Data access

Acessing recently generated data inside a plan can sometimes be useful, allowing an additional degree of liberty in automating procedures in a plan. However, it conflicts with the fact that local storage is not the end location of that data: for information to be processed and analized for its scientific merits, it needs to be accessible at a later time, and for that to happen, it’s end storage medium is Ibira. And, as with any other EPICS-based data collection procedure, area detectors save directly to Ibira, bypassing the orchestration layer.

Because of that, local reasoning based on generated data from a recent run is limited in scope, and can only do so much. It is, nonetheless, still useful in some cases, and is supported directly in Bluesky, through the bluesky.plan_stubs.subscribe() and bluesky.plan_stubs.unsubscribe() stub plans, which allows us to configure callbacks in the bluesky.RunEngine inside a plan, so we can attach a local databroker instance to collect the data right there.

Below is an example of such:

from bluesky import RunEngine, plan_stubs as bps
from bluesky.plans import count

from ophyd.sim import hw

from databroker.v2 import temp as db_temp

from sophys.common.plans.annotated_default_plans import DETECTORS_TYPE, MD_TYPE


def example_plan(detectors: DETECTORS_TYPE, *, num=10, md: MD_TYPE = None):
    md = md or {}
    _md = {**md, "my_other_thing_key": "my_other_thing_value"}

    db = db_temp()

    # Make all Bluesky events go to the databroker catalog
    token = (yield from bps.subscribe("all", db.v1.insert))

    uid = (yield from count(detectors, num=num, delay=0, md=_md))

    # Remove the databroker catalog callback
    yield from bps.unsubscribe(token)

    data_from_count = db[uid]

    # Do some (basic) stuff with our data!
    data_from_count = data_from_count.primary.read()

    data_med = float(sum(data_from_count.rand) / len(data_from_count.rand))
    print("Median (n={}): {:.5f}".format(num, data_med))

    if abs(data_med - 0.5) > 0.0025:
        new_num = num * 2
        print("Not enough data! Will retry with n={}".format(new_num))

        yield from example_plan(detectors, num=new_num, md=md)


if __name__ == "__main__":
    RE = RunEngine({})

    rand = hw().rand
    rand.kind = "hinted"     # Specific to the simulated random device
    rand.start_simulation()  # Specific to the simulated random device

    RE(example_plan([rand]))

Misc.

class sophys.common.utils.DebugOptions(file: Union[io.IOBase, str, NoneType] = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, datefmt: str = '%H:%M:%S', color: bool = True, level: Union[str, int] = 'DEBUG')[source]

Bases: object

color: bool = True

Whether to use colored output in the logs.

datefmt: str = '%H:%M:%S'

What formatting to use for the date of an event.

file: IOBase | str | None = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>

Where to save the log.

static info_level()[source]

Create a DebugOptions object with no DEBUG messages processed.

level: str | int = 'DEBUG'

What is the minimum level of logging that will be processed.

class sophys.common.utils.HDF5PluginWithFileStore(prefix: str, *, override_path: bool = False, write_path_template='/tmp', read_attrs=[], **kwargs)[source]

Bases: HDF5Plugin, FileStoreHDF5IterativeWrite

A simple HDF5 plugin class, with some additional behavior (see Parameters and Attributes below).

Parameters:

  • prefix (str) – Prefix of this plugin (e.g. ‘HDF1:’)

  • override_path (bool, optional) – Whether to use the common override logic of HDF5 ophyd plugins. By default, it is disabled (the currently configured PV values will be used)

  • write_path_template (str, optional) – Used when override_path is True. By default it is /tmp.

warmup_acquire_time

The value to set AcquireTime when doing a warmup. Defaults to 1.0.

Type:

float

warmup_acquire_period

The value to set AcquirePeriod when doing a warmup. Defaults to 1.0.

Type:

float

warmup_signal_timeout

The timeout to pass to each set().wait() call in the warmup. Defaults to 5.0.

Type:

float

make_filename()[source]

Make a filename.

This is a hook so that the read and write paths can either be modified or created on disk prior to configuring the areaDetector plugin.

Returns:

  • filename (str) – The start of the filename

  • read_path (str) – Path that ophyd can read from

  • write_path (str) – Path that the IOC can write to

stage()[source]

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

unstage()[source]

Unstage the device.

This method returns the device to the state it was prior to the last stage call.

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

This method must be idempotent, multiple calls (without a new call to ‘stage’) have no effect.

Returns:

devices – list including self and all child devices unstaged

Return type:

list

warmup()[source]

A convenience method for ‘priming’ the plugin.

The plugin has to ‘see’ one acquisition before it is ready to capture. This sets the array size, etc.

This method is an override of the similar-named method in HDF5Plugin.

class sophys.common.utils.HDF5PluginWithFileStoreV34(prefix: str, *, override_path: bool = False, write_path_template='/tmp', read_attrs=[], **kwargs)[source]

Bases: HDF5PluginWithFileStore

A simple HDF5 plugin class, with some additional behavior (see Parameters and Attributes below).

Parameters:

  • prefix (str) – Prefix of this plugin (e.g. ‘HDF1:’)

  • override_path (bool, optional) – Whether to use the common override logic of HDF5 ophyd plugins. By default, it is disabled (the currently configured PV values will be used)

  • write_path_template (str, optional) – Used when override_path is True. By default it is /tmp.

warmup_acquire_time

The value to set AcquireTime when doing a warmup. Defaults to 1.0.

Type:

float

warmup_acquire_period

The value to set AcquirePeriod when doing a warmup. Defaults to 1.0.

Type:

float

warmup_signal_timeout

The timeout to pass to each set().wait() call in the warmup. Defaults to 5.0.

Type:

float

sophys.common.utils.is_in_queueserver() → bool[source]

Returns whether the current code is running in a queueserver environment (e.g. in the startup sequence), or not.

This is very similar to queueserver’s bluesky_queueserver.is_re_worker_active(), but without a hard dependency on queueserver.

sophys.common.utils.set_debug_mode(run_engine: RunEngine, bluesky_debug: bool | DebugOptions | None = None, ophyd_debug: bool | DebugOptions | None = None, mock_commands: bool = True, print_documents: bool = False, print_messages: bool = False) → dict[source]

Enables / disables debugging facilities for bluesky / ophyd.

Parameters:

  • run_engine (RunEngine) – The RunEngine on which to set the debug options. Note that the bluesky and ophyd debug options are global, so they affect all RunEngines.

  • bluesky_debug (DebugOptions or bool, optional) – Whether to enable / disable debug logging, or options to pass to bluesky’s logging configuration. Passing None will leave the configurations unchanged.

  • ophyd_debug (DebugOptions or bool, optional) – Whether to enable / disable debug logging, options to pass to ophyd’s logging configuration. Passing None will leave the configurations unchanged.

  • mock_commands (bool, optional) –

    Whether to mock all of ‘run_engine’s commands, replacing the default commands for dummy ones that only print to stdout what it would have done. Defaults to True.

    If it is set, the return’s “old_commands” key will be set to a dictionary, in which each command name (key) will be associated with the old command function (value).

  • print_documents (bool, optional) –

    Whether to subscribe ‘run_engine’ to a callback that prints every document generated. Defaults to False.

    If it is set, the return’s “print_sub_id” key will be set to the subscription ID returned by the RunEngine.

  • print_messages (bool, optional) –

    Whether to override ‘run_engine’s ‘msg_hook’ attribute to print every generated Msg object. Defaults to False.

    If it is set, the return’s “old_msg_hook” key will be set to the last function in ‘msg_hook’.