Experiment

class liso.experiment.DoocsInterface(facility_name: str, config: Optional[dict] = None)

Bases: liso.experiment.machine_interface.MachineInterface

Interface for machines which uses DOOCS control system.

add_control_channel(kls: Type[liso.experiment.doocs_channels.DoocsChannel], read_address: str, write_address: Optional[str] = None, *, non_event: bool = False, **kwargs) → None

Add a DOOCS channel for control data.

Parameters

  • kls – A concrete DoocsChannel class.

  • read_address – DOOCS read address.

  • write_address – DOOCS write address. It will be set to the same as the read address if not given.

  • non_event – True for a non-event-based channel (slow collector).

  • kwargs – Keyword arguments which will be passed to the constructor of kls after address.

Examples

>>> from liso import doocs_channels as dc
>>> from liso import EuXFELInterface

>>> m = EuXFELInterface()
>>> m.add_control_channel(
>>>     dc.FLOAT32,
>>>     'XFEL.RF/LLRF.CONTROLLER/VS.GUN.I1/PHASE.SAMPLE',
>>>     'XFEL.RF/LLRF.CONTROLLER/CTRL.GUN.I1/SP.PHASE')
add_diagnostic_channel(kls: Type[liso.experiment.doocs_channels.DoocsChannel], address: str, *, non_event: bool = False, **kwargs) → None

Add a DOOCS channel to diagnostic data.

Parameters

  • kls – A concrete DoocsChannel class.

  • address – DOOCS address.

  • non_event – True for a non-event-based channel.

  • kwargs – Keyword arguments which will be passed to the constructor of kls after address.

Examples

>>> from liso import doocs_channels as dc
>>> from liso import EuXFELInterface

>>> m = EuXFELInterface()
>>> m.add_diagnostic_channel(
>>>     dc.IMAGE, 'XFEL.DIAG/CAMERA/OTRC.64.I1D/IMAGE_EXT_ZMQ',
>>>     shape=(1750, 2330), dtype='uint16')
write(mapping: dict[str, Any], *, loop: Optional[asyncio.AbstractEventLoop] = None, executor: Optional[ThreadPoolExecutor] = None) → None

Write new value(s) to the given control channel(s).

Parameters

  • mapping – A mapping between DOOCS channel(s) and value(s).

  • loop – The event loop.

  • executor – ThreadPoolExecutor instance.

Raises

  • ModuleNotFoundError – If PyDOOCS cannot be imported.

  • LisoRuntimeError – If there is error when writing any channels.

read(loop: Optional[asyncio.events.AbstractEventLoop] = None, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None, correlate: bool = True, validate: bool = True) → dict

Return readout value(s) of the diagnostics channel(s).

Parameters

  • loop – The event loop.

  • executor – ThreadPoolExecutor instance.

  • correlate – True for returning the latest group of data with the same train ID.

  • validate – True for validate the readout values.

Raises

  • ModuleNotFoundError – If PyDOOCS cannot be imported.

  • LisoRuntimeError – If validation fails.

The returned data from each channel contains the following keys:

data, macropulse, timestamp, type, miscellaneous

monitor(executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None, correlate: bool = False, validate: bool = True) → None

Continuously monitoring the diagnostic channels.

Parameters

  • executor – ThreadPoolExecutor instance.

  • correlate – True for correlating all channel data.

  • validate – True for validating the readout data.

class liso.experiment.EuXFELInterface(config: Optional[dict] = None)

Bases: liso.experiment.doocs_interface.DoocsInterface

__init__(config: Optional[dict] = None)

Initialization.

Parameters

config – Config parameters for the facility.

class liso.experiment.FLASHInterface(config: Optional[dict] = None)

Bases: liso.experiment.doocs_interface.DoocsInterface

__init__(config: Optional[dict] = None)

Initialization.

Parameters

config – Config parameters for the facility.