API

The top level pandablocks module contains a number of packages that can be used from code:

• pandablocks.commands: The control commands that can be sent to a PandA

• pandablocks.responses: The control and data responses that will be received

• pandablocks.connections: Control and data connections that implements the parsing logic

• pandablocks.asyncio: An asyncio client that uses the control and data connections

• pandablocks.blocking: A blocking client that uses the control and data connections

• pandablocks.hdf: Some helpers for efficiently writing data responses to disk

• pandablocks.utils: General utility methods for use with pandablocks

Commands

There is a Command subclass for every sort of command that can be sent to the ControlConnection of a PandA. Many common actions can be accomplished with a simple Get or Put, but some convenience commands like GetBlockInfo, GetFieldInfo, etc. are provided that parse output into specific classes.

class pandablocks.commands.Command

Abstract baseclass for all ControlConnection commands to be inherited from

exception pandablocks.commands.CommandException

Raised if a Command receives a mal-formed response

class pandablocks.commands.Raw(inp: List[str])

Send a raw command

Parameters:

inp – The input lines to send

For example:

Raw(["PCAP.ACTIVE?"]) -> ["OK =1"]
Raw(["SEQ1.TABLE>", "1", "1", "0", "0", ""]) -> ["OK"]
Raw(["SEQ1.TABLE?"]) -> ["!1", "!1", "!0", "!0", "."])

class pandablocks.commands.Get(field: str)

Get the value of a field or star command.

If the form of the expected return is known, consider using GetLine or GetMultiline instead.

Parameters:

field – The field, attribute, or star command to get

For example:

Get("PCAP.ACTIVE") -> "1"
Get("SEQ1.TABLE") -> ["1048576", "0", "1000", "1000"]
Get("*IDN") -> "PandA 1.1..."

class pandablocks.commands.GetLine(field: str)

Get the value of a field or star command, when the result is expected to be a single line.

Parameters:

field – The field, attribute, or star command to get

For example:

GetLine("PCAP.ACTIVE") -> "1"
GetLine("*IDN") -> "PandA 1.1..."

class pandablocks.commands.GetMultiline(field: str)

Get the value of a field or star command, when the result is expected to be a multiline response.

Parameters:

field – The field, attribute, or star command to get

For example:

GetMultiline("SEQ1.TABLE") -> ["1048576", "0", "1000", "1000"]
GetMultiline("*METADATA.*") -> ["LABEL_FILTER1", "APPNAME", ...]

class pandablocks.commands.Put(field: str, value: str | List[str] = '')

Put the value of a field.

Parameters:

• field – The field, attribute, or star command to put

• value – The value, either string or list of strings or empty, to put

For example:

Put("PCAP.TRIG", "PULSE1.OUT")
Put("SEQ1.TABLE", ["1048576", "0", "1000", "1000"])
Put("SFP3_SYNC_IN1.SYNC_RESET")

class pandablocks.commands.Append(field: str, value: List[str])

Append the value of a table field.

Parameters:

• field – The field, attribute, or star command to append

• value – The value, list of strings, to append

For example:

Append("SEQ1.TABLE", ["1048576", "0", "1000", "1000"])

class pandablocks.commands.Arm

Arm PCAP for an acquisition by sending *PCAP.ARM=

class pandablocks.commands.Disarm

Disarm PCAP, stopping acquisition by sending *PCAP.DISARM=

class pandablocks.commands.GetBlockInfo(skip_description: bool = False)

Get the name, number, and description of each block type in a dictionary, alphabetically ordered

Parameters:

skip_description – If True, prevents retrieving the description for each Block. This will reduce network calls.

For example:

GetBlockInfo() ->
    {
        "LUT": BlockInfo(number=8, description="Lookup table"),
        "PCAP": BlockInfo(number=1, description="Position capture control"),
        ...
    }

class pandablocks.commands.GetFieldInfo(block: str, extended_metadata: bool = True)

Get the fields of a block, returning a FieldInfo (or appropriate subclass) for each one, ordered to match the definition order in the PandA

Parameters:

• block – The name of the block type

• extended_metadata – If True, retrieves detailed metadata about a field and all of its attributes. This will cause an additional network round trip. If False only the field names and types will be returned. Default True.

For example:

GetFieldInfo("LUT") -> {
    "INPA":
        BitMuxFieldInfo(type='bit_mux',
                        subtype=None,
                        description='Input A',
                        max_delay=5
                        label=['TTLIN1.VAL', 'TTLIN2.VAL', ...]),
    ...}

class pandablocks.commands.GetPcapBitsLabels

Get the labels for the bit fields in PCAP.

For example:

GetPcapBitsLabels() -> {"BITS0" : ["TTLIN1.VAL", "TTLIN2.VAL", ...], ...}

class pandablocks.commands.ChangeGroup(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Which group of values to ask for *CHANGES on: https://pandablocks-server.readthedocs.io/en/latest/commands.html#system-commands

ALL = ''

All the groups below

CONFIG = '.CONFIG'

Configuration settings

BITS = '.BITS'

Bits on the system bus

POSN = '.POSN'

Positions

READ = '.READ'

Polled read values

ATTR = '.ATTR'

Attributes (included capture enable flags)

TABLE = '.TABLE'

Table changes

METADATA = '.METADATA'

Table changes

class pandablocks.commands.GetChanges(group: ChangeGroup = ChangeGroup.ALL, get_multiline: bool = False)

Get a Changes object showing which fields have changed since the last time this was called

Parameters:

• group – Restrict to a particular ChangeGroup

• get_multiline – If True, return values of multiline fields in the multiline_values attribute. Note that this will invoke additional network requests. If False these fields will instead be returned in the no_value attribute. Default value is False.

For example:

GetChanges() -> Changes(
    value={"PCAP.TRIG": "PULSE1.OUT"},
    no_value=["SEQ1.TABLE"],
    in_error=["BAD.ENUM"],
    multiline_values={}
)

GetChanges(ChangeGroup.ALL, True) -> Changes(
    values={"PCAP.TRIG": "PULSE1.OUT"},
    no_value=[],
    in_error=["BAD.ENUM"],
    multiline_values={"SEQ1.TABLE" : ["1", "2", "3",...]}
)

class pandablocks.commands.GetState

Get the state of all the fields in a PandA that should be saved as a list of raw lines that could be sent with SetState.

NOTE: GetState may behave unexpectedly if GetChanges has previously been called using the same client. The caller should use separate clients to avoid potential issues.

For example:

GetState() -> [
    "SEQ1.TABLE<B"
    "234fds0SDklnmnr"
    ""
    "PCAP.TRIG=PULSE1.OUT",
]

class pandablocks.commands.SetState(state: List[str])

Set the state of all the fields in a PandA

Parameters:

state – A list of raw lines as produced by GetState

For example:

SetState([
    "SEQ1.TABLE<B"
    "234fds0SDklnmnr"
    ""
    "PCAP.TRIG=PULSE1.OUT",
])

Responses

Classes used in responses from both the ControlConnection and DataConnection of a PandA live in this package.

class pandablocks.responses.BlockInfo(number: int = 0, description: str | None = None)

Block number and description as exposed by the TCP server

number

The index of this block

Type:

int

description

The description for this block

Type:

str | None

class pandablocks.responses.FieldInfo(type: str, subtype: str | None, description: str | None)

Field type, subtype, description and labels as exposed by TCP server: https://pandablocks-server.readthedocs.io/en/latest/fields.html#field-types

Note that many fields will use a more specialised subclass of FieldInfo for their additional attributes.

type

Field type, like “param”, “bit_out”, “pos_mux”, etc.

Type:

str

subtype

Some types have subtype, like “uint”, “scalar”, “lut”, etc.

Type:

str | None

description

A description of the field

Type:

str | None

labels

A list of the valid values for the field when there is a defined list of valid values, e.g. those with sub-type “enum”

class pandablocks.responses.UintFieldInfo(type: str, subtype: str | None, description: str | None, max_val: int)

Extended FieldInfo for fields with type “param”,”read”, or “write” and subtype “uint

class pandablocks.responses.ScalarFieldInfo(type: str, subtype: str | None, description: str | None, units: str, scale: float, offset: float)

Extended FieldInfo for fields with type “param”,”read”, or “write” and subtype “scalar

class pandablocks.responses.TimeFieldInfo(type: str, subtype: str | None, description: str | None, units_labels: List[str], min_val: float)

Extended FieldInfo for fields with type “time

class pandablocks.responses.SubtypeTimeFieldInfo(type: str, subtype: str | None, description: str | None, units_labels: List[str])

Extended FieldInfo for fields with type “param”,”read”, or “write” and subtype “time

class pandablocks.responses.EnumFieldInfo(type: str, subtype: str | None, description: str | None, labels: List[str])

Extended FieldInfo for fields with type “param”,”read”, or “write” and subtype “enum

class pandablocks.responses.BitOutFieldInfo(type: str, subtype: str | None, description: str | None, capture_word: str, offset: int)

Extended FieldInfo for fields with type “bit_out

class pandablocks.responses.BitMuxFieldInfo(type: str, subtype: str | None, description: str | None, max_delay: int, labels: List[str])

Extended FieldInfo for fields with type “bit_mux

class pandablocks.responses.PosMuxFieldInfo(type: str, subtype: str | None, description: str | None, labels: List[str])

Extended FieldInfo for fields with type “pos_mux

class pandablocks.responses.TableFieldDetails(subtype: str, bit_low: int, bit_high: int, description: str | None = None, labels: List[str] | None = None)

Info for each field in a table

class pandablocks.responses.TableFieldInfo(type: str, subtype: str | None, description: str | None, max_length: int, fields: Dict[str, TableFieldDetails], row_words: int)

Extended FieldInfo for fields with type “table

class pandablocks.responses.PosOutFieldInfo(type: str, subtype: str | None, description: str | None, capture_labels: List[str])

Extended FieldInfo for fields with type “pos_out

class pandablocks.responses.ExtOutFieldInfo(type: str, subtype: str | None, description: str | None, capture_labels: List[str])

Extended FieldInfo for fields with type “ext_out” and subtypes “timestamp” or “samples

class pandablocks.responses.ExtOutBitsFieldInfo(type: str, subtype: str | None, description: str | None, capture_labels: List[str], bits: List[str])

Extended ExtOutFieldInfo for fields with type “ext_out” and subtype “bits

class pandablocks.responses.Changes(values: Dict[str, str], no_value: List[str], in_error: List[str], multiline_values: Dict[str, List[str]])

The changes returned from a *CHANGES command

values: Dict[str, str]

Map field -> value for single-line values that were returned

no_value: List[str]

The fields that were present but without value

in_error: List[str]

The fields that were in error

multiline_values: Dict[str, List[str]]

Map field -> value for multi-line values that were returned

class pandablocks.responses.EndReason(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

The reason that a PCAP acquisition completed

OK = 'Ok'

Experiment completed by falling edge of PCAP.ENABLE`

EARLY_DISCONNECT = 'Early disconnect'

Client disconnect detected

DATA_OVERRUN = 'Data overrun'

Client not taking data quickly or network congestion, internal buffer overflow. In raw unscaled mode (i.e., no server-side scaling), the most recent FrameData is likely corrupted.

FRAMING_ERROR = 'Framing error'

Triggers too fast for configured data capture

DRIVER_DATA_OVERRUN = 'Driver data overrun'

Probable CPU overload on PandA, should not occur

DMA_DATA_ERROR = 'DMA data error'

Data capture too fast for memory bandwidth

UNKNOWN_EXCEPTION = 'Unknown exception'

An unknown exception occurred during HDF5 file processing

START_DATA_MISMATCH = 'Start Data mismatched'

StartData packets did not match when trying to continue printing to a file

MANUALLY_STOPPED = 'Manually stopped'

Experiment manually completed by DATA:CAPTURE

DISARMED = 'Disarmed'

Experiment manually completed by *PCAP.DISARM= command

class pandablocks.responses.FieldCapture(name: str, type: dtype, capture: str, scale: float = 1.0, offset: float = 0.0, units: str = '')

Information about a field that is being captured

name

Name of captured field

Type:

str

type

Numpy data type of the field as transmitted

Type:

numpy.dtype

capture

Value of CAPTURE field used to enable this field

Type:

str

scale

Scaling factor, default 1.0

Type:

float

offset

Offset, default 0.0

Type:

float

units

Units string, default “”

Type:

str

class pandablocks.responses.Data

Baseclass for all responses yielded by a DataConnection

class pandablocks.responses.ReadyData

Yielded once when the connection is established and ready to take data

class pandablocks.responses.StartData(fields: List[FieldCapture], missed: int, process: str, format: str, sample_bytes: int)

Yielded when a new PCAP acquisition starts.

fields

Information about each captured field as a FieldCapture object

Type:

List[pandablocks.responses.FieldCapture]

missed

Number of samples missed by late data port connection

Type:

int

process

Data processing option, only “Scaled” or “Raw” are requested

Type:

str

format

Data delivery formatting, only “Framed” is requested

Type:

str

sample_bytes

Number of bytes in one sample

Type:

int

class pandablocks.responses.FrameData(data: ndarray)

Yielded when a new data frame is flushed.

data

A numpy Structured Array

Type:

numpy.ndarray

Data is structured into complete columns. Each column name is <name>.<capture> from the corresponding FieldInfo. Data can be accessed with these column names. For example:

# Table view with 2 captured fields
>>> import numpy
>>> data = numpy.array([(0, 10),
...       (1, 11),
...       (2, 12)],
...      dtype=[('COUNTER1.OUT.Value', '<f8'), ('COUNTER2.OUT.Value', '<f8')])
>>> fdata = FrameData(data)
>>> fdata.data[0] # Row view
(0., 10.)
>>> fdata.column_names # Column names
('COUNTER1.OUT.Value', 'COUNTER2.OUT.Value')
>>> fdata.data['COUNTER1.OUT.Value'] # Column view
array([0., 1., 2.])

property column_names: Tuple[str, ...]

Return all the column names

class pandablocks.responses.EndData(samples: int, reason: EndReason)

Yielded when a PCAP acquisition ends.

samples

The total number of samples (rows) that were yielded

Type:

int

reason

The EndReason for the end of acquisition

Type:

pandablocks.responses.EndReason

Connections

Sans-IO connections for both the Control and Data ports of PandA TCP server.

exception pandablocks.connections.NeedMoreData

Raised if the Buffer isn’t full enough to return the requested bytes

exception pandablocks.connections.NoContextAvailable

Raised if there were no contexts available for this connection. This may result from calling ControlConnection.receive_bytes() without calling ControlConnection.send(), or if there were unmatched sends/receives

class pandablocks.connections.Buffer

Byte storage that provides line reader and bytes reader interfaces. For example:

buf = Buffer()
buf += bytes_from_server
line = buf.read_line()  # raises NeedMoreData if no line
for line in buf:
    pass
bytes = buf.read_bytes(50)  # raises NeedMoreData if not enough bytes

read_bytes(num: int) → bytearray

Read and pop num bytes from the beginning of the buffer, raising NeedMoreData if the buffer isn’t full enough to do so

peek_bytes(num: int) → bytearray

Read but do not pop num bytes from the beginning of the buffer, raising NeedMoreData if the buffer isn’t full enough to do so

read_line()

Read and pop a newline terminated line (without terminator) from the beginning of the buffer, raising NeedMoreData if the buffer isn’t full enough to do so

class pandablocks.connections.ControlConnection

Sans-IO connection to control port of PandA TCP server, supporting a Command based interface. For example:

cc = ControlConnection()
# Connection says what bytes should be sent to execute command
to_send = cc.send(command)
socket.sendall(to_send)
while True:
    # Repeatedly process bytes from the PandA
    received = socket.recv()
    # Sending any subsequent bytes to be sent back to the PandA
    to_send = cc.receive_bytes(received)
    socket.sendall(to_send)
    # And processing the produced responses
    for command, response in cc.responses()
        do_something_with(response)

receive_bytes(received: bytes) → bytes

Tell the connection that you have received some bytes off the network. Parse these into high level responses which are yielded back by responses. Return any bytes to send back

responses() → Iterator[Tuple[Command, Any]]

Get the (command, response) tuples generated as part of the last receive_bytes

send(command: Command) → bytes

Tell the connection you want to send an event, and it will return some bytes to send down the network

class pandablocks.connections.DataConnection

Sans-IO connection to data port of PandA TCP server, supporting an flushable iterator interface. For example:

dc = DataConnection()
# Single connection string to send
to_send = dc.connect()
socket.sendall(to_send)
while True:
    # Repeatedly process bytes from the PandA looking for data
    received = socket.recv()
    for data in dc.receive_bytes(received):
        do_something_with(data)

receive_bytes(received: bytes, flush_every_frame=True) → Iterator[Data]

Tell the connection that you have received some bytes off the network. Parse these into Data structures and yield them back.

Parameters:

• received – the bytes you received from the socket

• flush_every_frame – Whether to flush FrameData as soon as received. If False then they will only be sent if flush is called or end of acquisition reached

flush() → Iterator[FrameData]

If there is a partial data frame, pop and yield it

connect(scaled: bool) → bytes

Return the bytes that need to be sent on connection

Asyncio Client

This is an asyncio wrapper to the ControlConnection and DataConnection objects, allowing async calls to send(command) and iterate over data().

class pandablocks.asyncio.AsyncioClient(host: str)

Asyncio implementation of a PandABlocks client. For example:

async with AsyncioClient("hostname-or-ip") as client:
    # Control port is now connected
    resp1, resp2 = await asyncio.gather(client.send(cmd1), client.send(cmd2))
    resp3 = await client.send(cmd3)
    async for data in client.data():
        handle(data)
# Control and data ports are now disconnected

async connect()

Connect to the control port, and be ready to handle commands

is_connected()

True if there is a currently active connection. NOTE: This does not indicate if the remote end is still connected.

async close()

Close the control connection, and wait for completion

async send(command: Command[T], timeout: float | None = None) → T

Send a command to control port of the PandA, returning its response.

Parameters:

command – The Command to send

async data(scaled: bool = True, flush_period: float | None = None, frame_timeout: float | None = None) → AsyncGenerator[Data, None]

Connect to data port and yield data frames

Parameters:

• scaled – Whether to scale and average data frames, reduces throughput

• flush_period – How often to flush partial data frames, None is on every chunk of data from the server

• frame_timeout – If no data is received for this amount of time, raise asyncio.TimeoutError

Blocking Client

This is a blocking wrapper to the ControlConnection and DataConnection objects, allowing blocking calls to send(commands) and iterate over data().

class pandablocks.blocking.BlockingClient(host: str)

Blocking implementation of a PandABlocks client. For example:

with BlockingClient("hostname-or-ip") as client:
    # Control port is now connected
    resp1, resp2 = client.send([cmd1, cmd2])
    resp3 = client.send(cmd3)
    for data in client.data():
        handle(data)
# Control and data ports are now disconnected

connect()

Connect to the control port, and be ready to handle commands

close()

Close the control connection, and wait for completion

send(commands: Command[T], timeout: int | None = None) → T

send(commands: Iterable[Command], timeout: int | None = None) → List

Send a command to control port of the PandA, returning its response.

Parameters:

• commands – If single Command, return its response. If a list of commands return a list of reponses

• timeout – If no reponse in this time, raise socket.timeout

data(scaled: bool = True, frame_timeout: int | None = None) → Iterator[Data]

Connect to data port and yield data frames

Parameters:

• scaled – Whether to scale and average data frames, reduces throughput

• frame_timeout – If no data is received for this amount of time, raise socket.timeout

HDF Writing

This package contains components needed to write PCAP data to and HDF file in the most efficient way. The oneshot write_hdf_files is exposed in the commandline interface. It assembles a short Pipeline of:

AsyncioClient -> FrameProcessor -> HDFWriter

The FrameProcessor and HDFWriter run in their own threads as most of the heavy lifting is done by numpy and h5py, so running in their own threads gives multi-CPU benefits without hitting the limit of the GIL.

See also

How to use the library to capture HDF files, How fast can we write HDF files?

class pandablocks.hdf.Pipeline

Helper class that runs a pipeline consumer process in its own thread

what_to_do: Dict[Type, Callable]

Subclasses should create this dictionary with handlers for each data type, returning transformed data that should be passed downstream

stop()

Stop the processing after the current queue has been emptied

class pandablocks.hdf.HDFWriter(file_names: Iterator[str])

Write an HDF file per data collection. Each field will be written in a 1D dataset /<field.name>.<field.capture>.

Parameters:

file_names – Iterator of file names. Must be full file paths. Will be called once per file created.

class pandablocks.hdf.FrameProcessor

Scale field data according to the information in the StartData

exception pandablocks.hdf.HDFDataOverrunException

Raised if DATA_OVERRUN occurs while receiving data for HDF file

pandablocks.hdf.create_pipeline(*elements: Pipeline) → List[Pipeline]

Create a pipeline of elements, wiring them and starting them before returning them

pandablocks.hdf.create_default_pipeline(file_names: Iterator[str], *additional_downstream_pipelines: Pipeline) → List[Pipeline]

Create the default processing pipeline consisting of one FrameProcessor and one HDFWriter. See create_pipeline for more details.

Parameters:

• file_names – Iterator of file names. Must be full file paths. Will be called once per file created. As required by HDFWriter.

• additional_downstream_pipelines – Any number of additional pipelines to add downstream.

pandablocks.hdf.stop_pipeline(pipeline: List[Pipeline])

Stop and join each element of the pipeline

async pandablocks.hdf.write_hdf_files(client: AsyncioClient, file_names: Iterator[str], num: int = 1, arm: bool = False, flush_period: float = 1)

Connect to host PandA data port, and write num acquisitions to HDF file according to scheme

Parameters:

• client – The AsyncioClient to use for communications

• file_names – Iterator of file names. Must be full file paths. Will be called once per file created.

• num – The number of acquisitions to store in separate files. 0 = Infinite capture

• arm – Whether to arm PCAP at the start, and after each successful acquisition

Raises:

HDFDataOverrunException – if there is a data overrun.

Utilities

This package contains general methods for working with pandablocks.