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(address: str, kls: Type[liso.experiment.doocs_channels.DoocsChannel] = <class 'liso.experiment.doocs_channels.AnyDoocsChannel'>, *, write_address: Optional[str] = None, non_event: bool = False, **kwargs) → None

Add a DOOCS channel for control data.

Parameters

  • address – DOOCS read address.

  • kls – A concrete DoocsChannel class.

  • 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(
>>>     'XFEL.RF/LLRF.CONTROLLER/VS.GUN.I1/PHASE.SAMPLE',
>>>     dc.FLOAT32,
>>>     write_address='XFEL.RF/LLRF.CONTROLLER/CTRL.GUN.I1/SP.PHASE')
add_diagnostic_channel(address: str, kls: Type[liso.experiment.doocs_channels.DoocsChannel] = <class 'liso.experiment.doocs_channels.AnyDoocsChannel'>, *, non_event: bool = False, **kwargs) → None

Add a DOOCS channel to diagnostic data.

Parameters

  • address – DOOCS address.

  • kls – A concrete DoocsChannel class.

  • 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(
>>>     'XFEL.DIAG/CAMERA/OTRC.64.I1D/IMAGE_EXT_ZMQ', dc.ARRAY,
>>>     shape=(1750, 2330), dtype='uint16')
async awrite(mapping: Dict[str, Any], *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → None

Asynchronously write new value(s) to the given control channel(s).

Parameters

  • mapping – A mapping between DOOCS channel(s) and value(s). These channels must be among the existing control channels and the new value will actually be written into the corresponding DOOCS write address. See add_control_channel()

  • executor – ThreadPoolExecutor instance.

Raises

LisoRuntimeError – If there is error when writing any channels.

aread(n: Optional[int] = None, *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → AsyncIterable[Tuple[int, Dict[str, dict]]]

Asynchronously read correlated data from all the channels.

Parameters

  • n – Number of correlated pulses to return. None for generating data infinitely.

  • executor – ThreadPoolExecutor instance.

Raises

LisoRuntimeError – If there is no event channel.

write(mapping: Dict[str, Any], *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → None

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

See awrite().

read(n: int, *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → List[Tuple[int, dict]]

Read correlated data from all the channels.

See aread().

query(addresses: Optional[Iterable[str]] = None, *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → Dict[str, dict]

Read data from channels once without correlating them.

Parameters

  • addresses – DOOCS addresses. None for reading all the existing channels.

  • executor – ThreadPoolExecutor instance.

acquire(output_dir: Union[str, pathlib.Path] = './', *, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None, chmod: bool = True, group: int = 1) → None

Acquiring correlated data and saving it to HDF5 files.

Parameters

  • output_dir – Directory in which data for each run is stored in in its own sub-directory.

  • executor – ThreadPoolExecutor instance.

  • chmod – True for changing the permission to 400 after finishing writing.

  • group – Writer group.

Raises

LisoRuntimeError – If there is no event channel.

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

Continuously monitoring the diagnostic channels.

Parameters

  • executor – ThreadPoolExecutor instance.

  • correlate – True for correlating all channel data.

  • verbose – True for printing out the full DOOCS message. Otherwise, only ‘data’ is printed out.

  • app – True for used in app with all channels belong to diagnostic.

Raises

LisoRuntimeError – If correlate is True and there is no event channel.

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.