instrument.utils

Add Python modules here that provide any other code not covered by callbacks, configs, core, devices, or plans.

aps_functions	APS utility helper functions
config_loaders	Load configuration files
controls_setup	EPICS & ophyd related setup
helper_functions	Generic utility helper functions
logging_setup	Configure logging for this session.
make_devices_yaml	Make devices from YAML files
metadata	RunEngine Metadata
stored_dict	Storage-backed Dictionary

APS utility helper functions

host_on_aps_subnet()	Detect if this host is on an APS subnet.
aps_dm_setup(dm_setup_file)	APS Data Management setup

instrument.utils.aps_functions.aps_dm_setup(dm_setup_file)

APS Data Management setup

Read the bash shell script file used by DM to setup the environment. Parse any export lines and add their environment variables to this session. This is done by brute force here since the APS DM environment setup requires different Python code than bluesky and the two often clash.

This setup must be done before any of the DM package libraries are called.

instrument.utils.aps_functions.host_on_aps_subnet()

Detect if this host is on an APS subnet.

Load configuration files

Load supported configuration files, such as iconfig.yml.

load_config_yaml([iconfig_yml])	Load iconfig.yml (and other YAML) configuration files.
IConfigFileVersionError	Configuration file version too old.

exception instrument.utils.config_loaders.IConfigFileVersionError

Bases: ValueError

Configuration file version too old.

instrument.utils.config_loaders.load_config_yaml(iconfig_yml=None) → dict

Load iconfig.yml (and other YAML) configuration files.

Parameters

iconfig_yml: str

Name of the YAML file to be loaded. The name can be absolute or relative to the current working directory. Default: INSTRUMENT/configs/iconfig.yml

EPICS & ophyd related setup

oregistry	Registry of all ophyd-style Devices and Signals.
set_control_layer([control_layer])	Communications library between ophyd and EPICS Channel Access.
set_timeouts()	Set default timeout for all EpicsSignal connections & communications.
epics_scan_id_source(_md)	Callback function for RunEngine.
connect_scan_id_pv(RE[, pv])	Define a PV to use for the RunEngine's scan_id.

instrument.utils.controls_setup.connect_scan_id_pv(RE, pv: str = None)

Define a PV to use for the RunEngine’s scan_id.

instrument.utils.controls_setup.epics_scan_id_source(_md)

Callback function for RunEngine. Returns next scan_id to be used.

• Ignore metadata dictionary passed as argument.

• Get current scan_id from PV.

• Apply lower limit of zero.

• Increment (so that scan_id numbering starts from 1).

• Set PV with new value.

• Return new value.

Exception will be raised if PV is not connected when next bps.open_run() is called.

instrument.utils.controls_setup.oregistry

Registry of all ophyd-style Devices and Signals.

instrument.utils.controls_setup.set_control_layer(control_layer: str = 'PyEpics')

Communications library between ophyd and EPICS Channel Access.

Choices are: PyEpics (default) or caproto.

OPHYD_CONTROL_LAYER is an application of “lessons learned.”

Only used in a couple rare cases where PyEpics code was failing. It’s defined here since it was difficult to find how to do this in the ophyd documentation.

instrument.utils.controls_setup.set_timeouts()

Set default timeout for all EpicsSignal connections & communications.

Generic utility helper functions

register_bluesky_magics()	The Bluesky Magick functions are useful with command-lines.
running_in_queueserver()	Detect if running in the bluesky queueserver.
debug_python()
mpl_setup()	Matplotlib setup based on environment (Notebook or non-Notebook).
is_notebook()	Detect if running in a notebook.

instrument.utils.helper_functions.debug_python()

instrument.utils.helper_functions.is_notebook()

Detect if running in a notebook.

see: https://stackoverflow.com/a/39662359/1046449

instrument.utils.helper_functions.mpl_setup()

Matplotlib setup based on environment (Notebook or non-Notebook).

instrument.utils.helper_functions.register_bluesky_magics()

The Bluesky Magick functions are useful with command-lines.

instrument.utils.helper_functions.running_in_queueserver()

Detect if running in the bluesky queueserver.

Configure logging for this session.

Public

configure_logging()

Configure logging as described in file.

Internal

_setup_console_logger(logger, cfg)	Reconfigure the root logger as configured by the user.
_setup_file_logger(logger, cfg)	Record log messages in file(s).
_setup_ipython_logger(logger, cfg)	Internal: Log IPython console session In and Out to a file.
_setup_module_logging(cfg)	Internal: Set logging level for each named module.

See also

https://blueskyproject.io/bluesky/main/debugging.html

instrument.utils.logging_setup._setup_console_logger(logger, cfg)

Reconfigure the root logger as configured by the user.

We can’t apply user configurations in configure_logging() above because the code to read the config file triggers initialization of the logging system.

See also

https://docs.python.org/3/library/logging.html#logging.basicConfig

instrument.utils.logging_setup._setup_file_logger(logger, cfg)

Record log messages in file(s).

instrument.utils.logging_setup._setup_ipython_logger(logger, cfg)

Internal: Log IPython console session In and Out to a file.

See logrotate? int he IPython console for more information.

instrument.utils.logging_setup._setup_module_logging(cfg)

Internal: Set logging level for each named module.

instrument.utils.logging_setup.addLoggingLevel(levelName, levelNum, methodName=None)

Comprehensively adds a new logging level to the logging module and the currently configured logging class.

levelName becomes an attribute of the logging module with the value levelNum. methodName becomes a convenience method for both logging itself and the class returned by logging.getLoggerClass() (usually just logging.Logger). If methodName is not specified, levelName.lower() is used.

To avoid accidental clobberings of existing attributes, this method will raise an AttributeError if the level name is already an attribute of the logging module or if the method name is already present

Example

>>> addLoggingLevel('TRACE', logging.INFO - 5)
>>> logging.getLogger(__name__).setLevel("TEST")
>>> logging.getLogger(__name__).test('that worked')
>>> logging.test('so did this')
>>> logging.TEST
5

instrument.utils.logging_setup.configure_logging()

Configure logging as described in file.

Make devices from YAML files

Construct ophyd-style devices from simple specifications in YAML files.

make_devices	(plan stub) Create the ophyd-style controls for this instrument.
Instrument	Custom YAML loader for guarneri.

class instrument.utils.make_devices_yaml.Instrument(*args: Any, **kwargs: Any)

Bases: Instrument

Custom YAML loader for guarneri.

parse_yaml_file(config_file: Path | str) → list[dict]

Read device configurations from YAML format file.

instrument.utils.make_devices_yaml._loader(yaml_device_file, main=True)

Load our ophyd-style controls as described in a YAML file.

PARAMETERS

yaml_device_filestr or pathlib.Path

YAML file describing ophyd-style controls to be created.

mainbool

If True add these devices to the __main__ namespace.

instrument.utils.make_devices_yaml.make_devices(*, pause: float = 1)

(plan stub) Create the ophyd-style controls for this instrument.

Feel free to modify this plan to suit the needs of your instrument.

EXAMPLE:

RE(make_devices())

PARAMETERS

pausefloat

Wait ‘pause’ seconds (default: 1) for slow objects to connect.

RunEngine Metadata

MD_PATH	Storage path to save RE metadata between sessions.
get_md_path()	Get path for RE metadata.
re_metadata([cat])	Programmatic metadata for the RunEngine.

instrument.utils.metadata.MD_PATH = '.re_md_dict.yml'

Storage path to save RE metadata between sessions.

instrument.utils.metadata.get_md_path()

Get path for RE metadata.

support	path
PersistentDict	Directory where dictionary keys are stored in separate files.
StoredDict	File where dictionary is stored as YAML.

In either case, the ‘path’ can be relative or absolute. Relative paths are with respect to the present working directory when the bluesky session is started.

instrument.utils.metadata.re_metadata(cat=None)

Programmatic metadata for the RunEngine.

Storage-backed Dictionary

A dictionary that writes its contents to YAML file.

Replaces bluesky.utils.PersistentDict.

• Contents must be JSON serializable.

• Contents stored in a single human-readable YAML file.

• Sync to disk shortly after dictionary is updated.

StoredDict(file[, delay, title, serializable])

Dictionary that syncs to storage.

class instrument.utils.stored_dict.StoredDict(file, delay=5, title=None, serializable=True)

Bases: MutableMapping

Dictionary that syncs to storage.

flush()	Force a write of the dictionary to disk
popitem()	Remove and return a (key, value) pair as a 2-tuple.
reload()	Read dictionary from storage.

Static methods

All support for the YAML format is implemented in the static methods.

dump(file, contents[, title])	Write dictionary to YAML file.
load(file)	Read dictionary from YAML file.

---

__init__(file, delay=5, title=None, serializable=True)

StoredDict : Dictionary that syncs to storage

PARAMETERS

filestr or pathlib.Path

Name of file to store dictionary contents.

delaynumber

Time delay (s) since last dictionary update to write to storage. Default: 5 seconds.

titlestr or None

Comment to write at top of file. Default: “Written by StoredDict.”

serializablebool

If True, validate new dictionary entries are JSON serializable.

_abc_impl = <_abc._abc_data object>

_delayed_sync_to_storage()

Sync the metadata to storage.

Start a time-delay thread. New writes to the metadata dictionary will extend the deadline. Sync once the deadline is reached.

static dump(file, contents, title=None)

Write dictionary to YAML file.

flush()

Force a write of the dictionary to disk

static load(file)

Read dictionary from YAML file.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

reload()

Read dictionary from storage.