ophyd_async.fastcs.panda#

Package Contents#

Classes#

CommonPandaBlocks

Pandablocks device with blocks which are common and required on introspection.

DataBlock

Data block for the PandA. Used for writing data through the IOC.

PandaBitMux

Bit input with configurable delay in the PandA.

PandaCaptureMode

Capture mode for the DataBlock on the PandA.

PcapBlock

Position capture block in the PandA.

PcompBlock

Position compare block in the PandA.

PandaPcompDirection

Direction options for position compare in the PandA.

PulseBlock

Used for configuring pulses in the PandA.

SeqBlock

Sequencer block in the PandA.

PandaTimeUnits

Options for units of time in the PandA.

HDFPanda

PandA with common blocks for standard HDF writing.

PandaHDFWriter

For writing for PandA data from the DataBlock.

PandaPcapController

For controlling a PCAP capture on the PandA.

DatasetTable

An abstraction of a Table where each field is a column.

PandaHdf5DatasetType

Dataset options for HDF capture.

SeqTable

Data type for the panda seq table.

SeqTrigger

Trigger options for the SeqTable.

PcompInfo

Info for the PandA PcompBlock for flyscanning.

SeqTableInfo

Info for the PandA SeqTable for flyscanning.

StaticPcompTriggerLogic

For controlling the PandA PcompBlock when flyscanning.

StaticSeqTableTriggerLogic

For controlling the PandA SeqTable when flyscanning.

API#

class ophyd_async.fastcs.panda.CommonPandaBlocks(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Pandablocks device with blocks which are common and required on introspection.

pulse: DeviceVector[PulseBlock]#

None

seq: DeviceVector[SeqBlock]#

None

pcomp: DeviceVector[PcompBlock]#

None

pcap: PcapBlock#

None

data: DataBlock#

None

class ophyd_async.fastcs.panda.DataBlock(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Data block for the PandA. Used for writing data through the IOC.

hdf_directory: SignalRW[str]#

None

hdf_file_name: SignalRW[str]#

None

num_capture: SignalRW[int]#

None

num_captured: SignalR[int]#

None

create_directory: SignalRW[int]#

None

directory_exists: SignalR[bool]#

None

capture_mode: SignalRW[PandaCaptureMode]#

None

capture: SignalRW[bool]#

None

flush_period: SignalRW[float]#

None

datasets: SignalR[DatasetTable]#

None

class ophyd_async.fastcs.panda.PandaBitMux[source]#

Bases: ophyd_async.core.SubsetEnum

Bit input with configurable delay in the PandA.

ZERO#

‘ZERO’

ONE#

‘ONE’

class ophyd_async.fastcs.panda.PandaCaptureMode[source]#

Bases: ophyd_async.core.StrictEnum

Capture mode for the DataBlock on the PandA.

FIRST_N#

‘FIRST_N’

LAST_N#

‘LAST_N’

FOREVER#

‘FOREVER’

class ophyd_async.fastcs.panda.PcapBlock(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Position capture block in the PandA.

active: SignalR[bool]#

None

arm: SignalRW[bool]#

None

class ophyd_async.fastcs.panda.PcompBlock(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Position compare block in the PandA.

active: SignalR[bool]#

None

dir: SignalRW[PandaPcompDirection]#

None

enable: SignalRW[PandaBitMux]#

None

pulses: SignalRW[int]#

None

start: SignalRW[int]#

None

step: SignalRW[int]#

None

width: SignalRW[int]#

None

class ophyd_async.fastcs.panda.PandaPcompDirection[source]#

Bases: ophyd_async.core.StrictEnum

Direction options for position compare in the PandA.

POSITIVE#

‘Positive’

NEGATIVE#

‘Negative’

EITHER#

‘Either’

class ophyd_async.fastcs.panda.PulseBlock(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Used for configuring pulses in the PandA.

delay: SignalRW[float]#

None

width: SignalRW[float]#

None

class ophyd_async.fastcs.panda.SeqBlock(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Sequencer block in the PandA.

table: SignalRW[SeqTable]#

None

active: SignalR[bool]#

None

repeats: SignalRW[int]#

None

prescale: SignalRW[float]#

None

prescale_units: SignalRW[PandaTimeUnits]#

None

enable: SignalRW[PandaBitMux]#

None

class ophyd_async.fastcs.panda.PandaTimeUnits[source]#

Bases: ophyd_async.core.StrictEnum

Options for units of time in the PandA.

MIN#

‘min’

S#

‘s’

MS#

‘ms’

US#

‘us’

class ophyd_async.fastcs.panda.HDFPanda(prefix: str, path_provider: PathProvider, config_sigs: Sequence[SignalR] = (), name: str = '')[source]#

Bases: ophyd_async.fastcs.panda._block.CommonPandaBlocks, ophyd_async.core.StandardDetector[ophyd_async.fastcs.panda._control.PandaPcapController, ophyd_async.fastcs.panda._writer.PandaHDFWriter]

PandA with common blocks for standard HDF writing.

class ophyd_async.fastcs.panda.PandaHDFWriter(path_provider: PathProvider, name_provider: NameProvider, panda_data_block: DataBlock)[source]#

Bases: ophyd_async.core.DetectorWriter

For writing for PandA data from the DataBlock.

async open(multiplier: int = 1) dict[str, DataKey][source]#

Retrieve and get descriptor of all PandA signals marked for capture.

async wait_for_index(index: int, timeout: float | None = DEFAULT_TIMEOUT)[source]#
async get_indices_written() int[source]#

Get the number of indices written.

async observe_indices_written(timeout=DEFAULT_TIMEOUT) AsyncGenerator[int, None][source]#

Wait until a specific index is ready to be collected.

async collect_stream_docs(indices_written: int) AsyncIterator[StreamAsset][source]#
async close()[source]#

Close writer, blocks until I/O is complete.

class ophyd_async.fastcs.panda.PandaPcapController(pcap: PcapBlock)[source]#

Bases: ophyd_async.core.DetectorController

For controlling a PCAP capture on the PandA.

get_deadtime(exposure: float | None) float[source]#

For a given exposure, how long should the time between exposures be.

async prepare(trigger_info: TriggerInfo)[source]#

Do all necessary steps to prepare the detector for triggers.

Parameters:

trigger_info – The sort of triggers to expect.

async arm()[source]#

Arm the detector.

async wait_for_idle()[source]#

Wait on the internal _arm_status and wait for it to get disarmed/idle.

async disarm()[source]#

Disarm the detector, return detector to an idle state.

class ophyd_async.fastcs.panda.DatasetTable(**kwargs)[source]#

Bases: ophyd_async.core.Table

An abstraction of a Table where each field is a column.

For example:

>>> from ophyd_async.core import Table, Array1D
>>> import numpy as np
>>> from collections.abc import Sequence
>>> class MyTable(Table):
...     a: Array1D[np.int8]
...     b: Sequence[str]
...
>>> t = MyTable(a=[1, 2], b=["x", "y"])
>>> len(t)  # the length is the number of rows
2
>>> t2 = t + t  # adding tables together concatenates them
>>> t2.a
array([1, 2, 1, 2], dtype=int8)
>>> t2.b
['x', 'y', 'x', 'y']
>>> t2[1]  # slice a row
array([(2, b'y')], dtype=[('a', 'i1'), ('b', 'S40')])

name: Sequence[str]#

None

dtype: Sequence[PandaHdf5DatasetType]#

None

class ophyd_async.fastcs.panda.PandaHdf5DatasetType[source]#

Bases: ophyd_async.core.StrictEnum

Dataset options for HDF capture.

FLOAT_64#

‘float64’

UINT_32#

‘uint32’

class ophyd_async.fastcs.panda.SeqTable(**kwargs)[source]#

Bases: ophyd_async.core.Table

Data type for the panda seq table.

repeats: Array1D[uint16]#

None

trigger: Sequence[SeqTrigger]#

None

position: Array1D[int32]#

None

time1: Array1D[uint32]#

None

outa1: Array1D[bool_]#

None

outb1: Array1D[bool_]#

None

outc1: Array1D[bool_]#

None

outd1: Array1D[bool_]#

None

oute1: Array1D[bool_]#

None

outf1: Array1D[bool_]#

None

time2: Array1D[uint32]#

None

outa2: Array1D[bool_]#

None

outb2: Array1D[bool_]#

None

outc2: Array1D[bool_]#

None

outd2: Array1D[bool_]#

None

oute2: Array1D[bool_]#

None

outf2: Array1D[bool_]#

None

static row(*, repeats: int = 1, trigger: str = SeqTrigger.IMMEDIATE, position: int = 0, time1: int = 0, outa1: bool = False, outb1: bool = False, outc1: bool = False, outd1: bool = False, oute1: bool = False, outf1: bool = False, time2: int = 0, outa2: bool = False, outb2: bool = False, outc2: bool = False, outd2: bool = False, oute2: bool = False, outf2: bool = False) SeqTable[source]#
class ophyd_async.fastcs.panda.SeqTrigger[source]#

Bases: ophyd_async.core.StrictEnum

Trigger options for the SeqTable.

IMMEDIATE#

‘Immediate’

BITA_0#

‘BITA=0’

BITA_1#

‘BITA=1’

BITB_0#

‘BITB=0’

BITB_1#

‘BITB=1’

BITC_0#

‘BITC=0’

BITC_1#

‘BITC=1’

POSA_GT#

‘POSA>=POSITION’

POSA_LT#

‘POSA<=POSITION’

POSB_GT#

‘POSB>=POSITION’

POSB_LT#

‘POSB<=POSITION’

POSC_GT#

‘POSC>=POSITION’

POSC_LT#

‘POSC<=POSITION’

class ophyd_async.fastcs.panda.PcompInfo(/, **data: ~typing.Any)[source]#

Bases: pydantic.BaseModel

Info for the PandA PcompBlock for flyscanning.

start_postion: int#

‘Field(…)’

pulse_width: int#

‘Field(…)’

rising_edge_step: int#

‘Field(…)’

number_of_pulses: int#

‘Field(…)’

direction: PandaPcompDirection#

‘Field(…)’

class ophyd_async.fastcs.panda.SeqTableInfo(/, **data: ~typing.Any)[source]#

Bases: pydantic.BaseModel

Info for the PandA SeqTable for flyscanning.

sequence_table: SeqTable#

‘Field(…)’

repeats: int#

‘Field(…)’

prescale_as_us: float#

‘Field(…)’

class ophyd_async.fastcs.panda.StaticPcompTriggerLogic(pcomp: PcompBlock)[source]#

Bases: ophyd_async.core.FlyerController[ophyd_async.fastcs.panda._trigger.PcompInfo]

For controlling the PandA PcompBlock when flyscanning.

async prepare(value: PcompInfo)[source]#

Move to the start of the flyscan.

async kickoff() None[source]#

Start the flyscan.

async complete(timeout: float | None = None) None[source]#

Block until the flyscan is done.

async stop()[source]#

Stop flying and wait everything to be stopped.

class ophyd_async.fastcs.panda.StaticSeqTableTriggerLogic(seq: SeqBlock)[source]#

Bases: ophyd_async.core.FlyerController[ophyd_async.fastcs.panda._trigger.SeqTableInfo]

For controlling the PandA SeqTable when flyscanning.

async prepare(value: SeqTableInfo)[source]#

Move to the start of the flyscan.

async kickoff() None[source]#

Start the flyscan.

async complete() None[source]#

Block until the flyscan is done.

async stop()[source]#

Stop flying and wait everything to be stopped.