accml.custom.bluesky package

Submodules

accml.custom.bluesky.bluesky_measurement_execution_engine module

Demonstrator of a measurement execution engine

Todo

• implement full functionality

Missing features:

• handling behaviour_on_error

• how much of the accelerator access to wrap?

• hide devices behind a software multiplexer? In its current form quite some information will be stored in the databroker

class accml.custom.bluesky.bluesky_measurement_execution_engine.BlueskyMeasurementExecutionEngine(*, devices, run_engine, info_signals, cache)

Bases: MeasurementExecutionEngine

Demonstrator of a measurement engine as a bluesky runengine

Parameters:

• devices (DevicesFacade)

• run_engine (RunEngine)

• info_signals (Sequence[Signal])

• cache (StateCache)

async execute(commands_collection, detectors, md, **kwargs)

Return type:

str

Returns:

identifier to the data

Parameters:

• commands_collection (Sequence[TransactionCommand])

• detectors (Sequence[ReadCommand])

• md (Dict[str, object])

Measurement engine is responsible to store data as appropriate

async set(cmds)

execute these commands together

Parameters:

cmds (Sequence[Command])

async trigger_read(cmds)

Following ophyd-async / ophyd design

Todo

Is this a good idea?

Return type:

ReadTogether

Parameters:

cmds (Sequence[ReadCommand])

accml.custom.bluesky.plans module

accml.custom.bluesky.plans.simple_command_sequence_execution_plan(*, commands, detectors, actuators, info_signals, md, **kwargs)

Translate simple commands (i.e. only one device at a time) using bluesky plan stubs

Parameters:

• commands (Sequence[Command])

• detectors (Sequence[Device])

• actuators (Dict[str, Device])

• info_signals (Dict[str, Signal])

• md (Dict)

accml.custom.bluesky.plans.transactional_actuator_commands_plan(*, transactional_commands, detectors, actuators, cache, num_readings, wait_before_read=0)

Inner plan for transactional_commands_sequence_execution_plan()

Todo

Implement stop, ignore, rollback etc Device replace by ophyd_async.Settable info_signals as dataclass?

Parameters:

• transactional_commands (Sequence[TransactionCommand])

• detectors (Sequence[Device])

• actuators (Dict[str, Device])

• cache (StateCache)

• num_readings (int)

• wait_before_read (float)

accml.custom.bluesky.plans.transactional_commands_sequence_execution_plan(*, transactional_commands, detectors, actuators, cache, md, **kwargs)

Excute transactional commands using bluesky plan stubs

Parameters:

• transactional_commands (Sequence[TransactionCommand])

• detectors (Sequence[Device])

• actuators (Dict[str, Device])

• cache (StateCache)

• md (Dict)

accml.custom.bluesky.run_correction module

bluesky plans for running a correction in an infinte loop

Todo

remove dependencies on Tune device. Tune Device should return a data model

accml.custom.bluesky.run_correction.corrections_plan(*, oracle, policy, detectors, actuators, set_commands, info_signals, cache, md, **kwargs)

Translate commands to bluesky run-engine messages

Parameters:

• oracle (Oracle)

• policy (PolicyBase)

• detectors (Sequence[Device])

• actuators (Dict[str, Device])

• set_commands (Dict[str, Command])

• info_signals (Dict[str, SignalRW])

• cache (StateCache)

• md (Dict)

accml.custom.bluesky.utils module

accml.custom.bluesky.utils.as_json_compatible_device_id(id_)

Return a JSON-compatible identifier.

• If id_ is a string, it is assumed to already be JSON-compatible and is returned unchanged.

• If id_ is a TangoResourceLocator, its json_compatible() form is returned.

Any other type is considered a programming error.

Return type:

str

Parameters:

id_ (str | TangoResourceLocator)

async accml.custom.bluesky.utils.connect_to_devices(devices, timeout=5.0)

accml.custom.bluesky.utils.extract_device_identifiers(commands_collection)

Return type:

Sequence[str]

Parameters:

commands_collection (Sequence[TransactionCommand])

Module contents