import logging
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import (
Any,
Callable,
Dict,
Generator,
Generic,
List,
Optional,
Tuple,
TypeVar,
Union,
overload,
)
from ._exchange import Exchange, ExchangeGenerator
from .responses import (
BitMuxFieldInfo,
BitOutFieldInfo,
BlockInfo,
Changes,
EnumFieldInfo,
ExtOutBitsFieldInfo,
ExtOutFieldInfo,
FieldInfo,
PosMuxFieldInfo,
PosOutFieldInfo,
ScalarFieldInfo,
SubtypeTimeFieldInfo,
TableFieldDetails,
TableFieldInfo,
TimeFieldInfo,
UintFieldInfo,
)
# Define the public API of this module
__all__ = [
"Command",
"CommandException",
"Raw",
"Get",
"GetLine",
"GetMultiline",
"Put",
"Append",
"Arm",
"Disarm",
"GetBlockInfo",
"GetFieldInfo",
"GetPcapBitsLabels",
"ChangeGroup",
"GetChanges",
"GetState",
"SetState",
]
T = TypeVar("T")
T2 = TypeVar("T2")
T3 = TypeVar("T3")
T4 = TypeVar("T4")
# Checks whether the server will interpret cmd as a table command: search for
# first of '?', '=', '<', if '<' found first then it's a multiline command.
MULTILINE_COMMAND = re.compile(r"^[^?=]*<")
def is_multiline_command(cmd: str):
return MULTILINE_COMMAND.match(cmd) is not None
[docs]
@dataclass
class Command(Generic[T]):
"""Abstract baseclass for all ControlConnection commands to be inherited from"""
def execute(self) -> ExchangeGenerator[T]:
# A generator that sends lines to the PandA, gets lines back, and returns a
# response
raise NotImplementedError(self)
[docs]
class CommandException(Exception):
"""Raised if a `Command` receives a mal-formed response"""
# `execute_commands()` actually returns a list with length equal to the number
# of tasks passed; however, Tuple is used similar to the annotation for
# zip() because typing does not support variadic type variables. See
# typeshed PR #1550 for discussion.
@overload
def _execute_commands(c1: Command[T]) -> ExchangeGenerator[Tuple[T]]: ...
@overload
def _execute_commands(
c1: Command[T], c2: Command[T2]
) -> ExchangeGenerator[Tuple[T, T2]]: ...
@overload
def _execute_commands(
c1: Command[T], c2: Command[T2], c3: Command[T3]
) -> ExchangeGenerator[Tuple[T, T2, T3]]: ...
@overload
def _execute_commands(
c1: Command[T], c2: Command[T2], c3: Command[T3], c4: Command[T4]
) -> ExchangeGenerator[Tuple[T, T2, T3, T4]]: ...
@overload
def _execute_commands(
*commands: Command[Any],
) -> ExchangeGenerator[Tuple[Any, ...]]: ...
def _execute_commands(*commands):
"""Call the `Command.execute` method on each of the commands to produce
some `Exchange` generators, which are yielded back to the connection,
then zip together the responses to those exchanges into a tuple"""
# If we add type annotations to this function then mypy complains:
# Overloaded function implementation does not accept all possible arguments
# As we want to type check this, we put the logic in _zip_with_return
ret = yield from _zip_with_return([command.execute() for command in commands])
return ret
def _zip_with_return(
generators: List[ExchangeGenerator[Any]],
) -> ExchangeGenerator[Tuple[Any, ...]]:
# Sentinel to show what generators are not yet exhausted
pending = object()
returns = [pending] * len(generators)
while True:
yields: List[Exchange] = []
for i, gen in enumerate(generators):
# If we haven't exhausted the generator
if returns[i] is pending:
try:
# Get the exchanges that it wants to fill in
exchanges = next(gen)
except StopIteration as e:
# Generator is exhausted, store its return value
returns[i] = e.value
else:
# Add the exchanges to the list
if isinstance(exchanges, list):
yields += exchanges
else:
yields.append(exchanges)
if yields:
# There were some Exchanges yielded, so yield them all up
# for the Connection to fill in
yield yields
else:
# All the generators are exhausted, so return the tuple of all
# their return values
return tuple(returns)
[docs]
@dataclass
class Raw(Command[List[str]]):
"""Send a raw command
Args:
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", "."])
"""
inp: List[str]
def execute(self) -> ExchangeGenerator[List[str]]:
ex = Exchange(self.inp)
yield ex
return ex.received
[docs]
@dataclass
class Get(Command[Union[str, List[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.
Args:
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..."
"""
field: str
def execute(self) -> ExchangeGenerator[Union[str, List[str]]]:
ex = Exchange(f"{self.field}?")
yield ex
if ex.is_multiline:
return ex.multiline
else:
# We got OK =value
line = ex.line
assert line.startswith("OK =")
return line[4:]
[docs]
@dataclass
class GetLine(Command[str]):
"""Get the value of a field or star command, when the result is expected to be a
single line.
Args:
field: The field, attribute, or star command to get
For example::
GetLine("PCAP.ACTIVE") -> "1"
GetLine("*IDN") -> "PandA 1.1..."
"""
field: str
def execute(self) -> ExchangeGenerator[str]:
ex = Exchange(f"{self.field}?")
yield ex
# Expect "OK =value"
line = ex.line
assert line.startswith("OK =")
return line[4:]
[docs]
@dataclass
class GetMultiline(Command[List[str]]):
"""Get the value of a field or star command, when the result is expected to be a
multiline response.
Args:
field: The field, attribute, or star command to get
For example::
GetMultiline("SEQ1.TABLE") -> ["1048576", "0", "1000", "1000"]
GetMultiline("*METADATA.*") -> ["LABEL_FILTER1", "APPNAME", ...]
"""
field: str
def execute(self) -> ExchangeGenerator[List[str]]:
ex = Exchange(f"{self.field}?")
yield ex
return ex.multiline
[docs]
@dataclass
class Put(Command[None]):
"""Put the value of a field.
Args:
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")
"""
field: str
value: Union[str, List[str]] = ""
def execute(self) -> ExchangeGenerator[None]:
if isinstance(self.value, list):
# Multiline table with blank line to terminate
ex = Exchange([f"{self.field}<"] + self.value + [""])
else:
ex = Exchange(f"{self.field}={self.value}")
yield ex
assert ex.line == "OK"
[docs]
@dataclass
class Append(Command[None]):
"""Append the value of a table field.
Args:
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"])
"""
field: str
value: List[str]
def execute(self) -> ExchangeGenerator[None]:
# Multiline table with blank line to terminate
ex = Exchange([f"{self.field}<<"] + self.value + [""])
yield ex
assert ex.line == "OK"
[docs]
class Arm(Command[None]):
"""Arm PCAP for an acquisition by sending ``*PCAP.ARM=``"""
def execute(self) -> ExchangeGenerator[None]:
ex = Exchange("*PCAP.ARM=")
yield ex
assert ex.line == "OK"
[docs]
class Disarm(Command[None]):
"""Disarm PCAP, stopping acquisition by sending ``*PCAP.DISARM=``"""
def execute(self) -> ExchangeGenerator[None]:
ex = Exchange("*PCAP.DISARM=")
yield ex
assert ex.line == "OK"
[docs]
@dataclass
class GetBlockInfo(Command[Dict[str, BlockInfo]]):
"""Get the name, number, and description of each block type
in a dictionary, alphabetically ordered
Args:
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"),
...
}
"""
skip_description: bool = False
def execute(self) -> ExchangeGenerator[Dict[str, BlockInfo]]:
ex = Exchange("*BLOCKS?")
yield ex
blocks_list, commands = [], []
for line in ex.multiline:
block, num = line.split()
blocks_list.append((block, int(num)))
commands.append(GetLine(f"*DESC.{block}"))
if self.skip_description:
# Must use tuple() to match type returned by _execute_commands
description_values = tuple(None for _ in commands)
else:
description_values = yield from _execute_commands(*commands)
block_infos = {
block: BlockInfo(number=num, description=desc)
for (block, num), desc in sorted(zip(blocks_list, description_values))
}
return block_infos
# The type of the generators used for creating the Get commands for each field
# and setting the returned data into the FieldInfo structure
_FieldGeneratorType = Generator[
Union[Exchange, List[Exchange]],
None,
Tuple[str, FieldInfo],
]
[docs]
@dataclass
class GetFieldInfo(Command[Dict[str, FieldInfo]]):
"""Get the fields of a block, returning a `FieldInfo` (or appropriate subclass) for
each one, ordered to match the definition order in the PandA
Args:
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', ...]),
...}
"""
block: str
extended_metadata: bool = True
_commands_map: Dict[
Tuple[str, Optional[str]],
Callable[
[str, str, Optional[str]],
_FieldGeneratorType,
],
] = field(init=False, repr=False, default_factory=dict)
def __post_init__(self):
# Map a (type, subtype) to a method that returns the appropriate
# subclasss of FieldInfo, and a list of all the Commands to request.
# Note that fields that do not have additional attributes are not listed.
self._commands_map = {
# Order matches that of PandA server's Field Types docs
("time", None): self._time,
("bit_out", None): self._bit_out,
("pos_out", None): self._pos_out,
("ext_out", "timestamp"): self._ext_out,
("ext_out", "samples"): self._ext_out,
("ext_out", "bits"): self._ext_out_bits,
("bit_mux", None): self._bit_mux,
("pos_mux", None): self._pos_mux,
("table", None): self._table,
("param", "uint"): self._uint,
("read", "uint"): self._uint,
("write", "uint"): self._uint,
("param", "int"): self._no_attributes,
("read", "int"): self._no_attributes,
("write", "int"): self._no_attributes,
("param", "scalar"): self._scalar,
("read", "scalar"): self._scalar,
("write", "scalar"): self._scalar,
("param", "bit"): self._no_attributes,
("read", "bit"): self._no_attributes,
("write", "bit"): self._no_attributes,
("param", "action"): self._no_attributes,
("read", "action"): self._no_attributes,
("write", "action"): self._no_attributes,
("param", "lut"): self._no_attributes,
("read", "lut"): self._no_attributes,
("write", "lut"): self._no_attributes,
("param", "enum"): self._enum,
("read", "enum"): self._enum,
("write", "enum"): self._enum,
("param", "time"): self._subtype_time,
("read", "time"): self._subtype_time,
("write", "time"): self._subtype_time,
}
def _get_desc(self, field_name: str) -> GetLine:
"""Create the Command to retrieve the description"""
return GetLine(f"*DESC.{self.block}.{field_name}")
def _uint(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, maximum = yield from _execute_commands(
self._get_desc(field_name),
GetLine(f"{self.block}1.{field_name}.MAX"),
)
field_info = UintFieldInfo(field_type, field_subtype, desc, int(maximum))
return field_name, field_info
def _scalar(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, units, scale, offset = yield from _execute_commands(
self._get_desc(field_name),
GetLine(f"{self.block}.{field_name}.UNITS"),
GetLine(f"{self.block}.{field_name}.SCALE"),
GetLine(f"{self.block}.{field_name}.OFFSET"),
)
field_info = ScalarFieldInfo(
field_type, field_subtype, desc, units, float(scale), int(offset)
)
return field_name, field_info
def _subtype_time(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, units_labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}.UNITS"),
)
field_info = SubtypeTimeFieldInfo(field_type, field_subtype, desc, units_labels)
return field_name, field_info
def _enum(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}"),
)
field_info = EnumFieldInfo(field_type, field_subtype, desc, labels)
return field_name, field_info
def _time(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, units, min = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}.UNITS"),
GetLine(f"{self.block}1.{field_name}.MIN"),
)
field_info = TimeFieldInfo(field_type, field_subtype, desc, units, float(min))
return field_name, field_info
def _bit_out(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, capture_word, offset = yield from _execute_commands(
self._get_desc(field_name),
GetLine(f"{self.block}1.{field_name}.CAPTURE_WORD"),
GetLine(f"{self.block}1.{field_name}.OFFSET"),
)
field_info = BitOutFieldInfo(
field_type, field_subtype, desc, capture_word, int(offset)
)
return field_name, field_info
def _bit_mux(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, max_delay, labels = yield from _execute_commands(
self._get_desc(field_name),
GetLine(f"{self.block}1.{field_name}.MAX_DELAY"),
GetMultiline(f"*ENUMS.{self.block}.{field_name}"),
)
field_info = BitMuxFieldInfo(
field_type, field_subtype, desc, int(max_delay), labels
)
return field_name, field_info
def _pos_mux(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}"),
)
field_info = PosMuxFieldInfo(field_type, field_subtype, desc, labels)
return field_name, field_info
def _table(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
# Ignore the ROW_WORDS attribute as it's new and won't be present on all PandAs,
# and there's no easy way to try it and catch an error while also running other
# Get commands at the same time
table_desc, max_length, fields = yield from _execute_commands(
self._get_desc(field_name),
GetLine(f"{self.block}1.{field_name}.MAX_LENGTH"),
GetMultiline(f"{self.block}1.{field_name}.FIELDS"),
)
# Keep track of highest bit index
max_bit_offset: int = 0
desc_gets: List[GetLine] = []
enum_field_gets: List[GetMultiline] = []
enum_field_names: List[str] = []
fields_dict: Dict[str, TableFieldDetails] = {}
for field_details in fields:
# Fields are of the form <bit_high>:<bit_low> <name> <subtype>
bit_range, name, subtype = field_details.split()
bit_high_str, bit_low_str = bit_range.split(":")
bit_high = int(bit_high_str)
bit_low = int(bit_low_str)
if bit_high > max_bit_offset:
max_bit_offset = bit_high
if subtype == "enum":
enum_field_gets.append(
GetMultiline(f"*ENUMS.{self.block}1.{field_name}[].{name}")
)
enum_field_names.append(name)
fields_dict[name] = TableFieldDetails(subtype, bit_low, bit_high)
desc_gets.append(GetLine(f"*DESC.{self.block}1.{field_name}[].{name}"))
# Calculate the number of 32 bit words that comprises one table row
row_words = max_bit_offset // 32 + 1
# The first len(enum_field_gets) items are enum labels, type List[str]
# The second part of the list are descriptions, type str
labels_and_descriptions = yield from _execute_commands(
*enum_field_gets, *desc_gets
)
for name, labels in zip(
enum_field_names, labels_and_descriptions[: len(enum_field_gets)]
):
fields_dict[name].labels = labels
for name, desc in zip(
fields_dict.keys(), labels_and_descriptions[len(enum_field_gets) :]
):
fields_dict[name].description = desc
field_info = TableFieldInfo(
field_type,
field_subtype,
table_desc,
int(max_length),
fields_dict,
row_words,
)
return field_name, field_info
def _pos_out(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, capture_labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}.CAPTURE"),
)
field_info = PosOutFieldInfo(field_type, field_subtype, desc, capture_labels)
return field_name, field_info
def _ext_out(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, capture_labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"*ENUMS.{self.block}.{field_name}.CAPTURE"),
)
field_info = ExtOutFieldInfo(field_type, field_subtype, desc, capture_labels)
return field_name, field_info
def _ext_out_bits(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
desc, bits, capture_labels = yield from _execute_commands(
self._get_desc(field_name),
GetMultiline(f"{self.block}.{field_name}.BITS"),
GetMultiline(f"*ENUMS.{self.block}.{field_name}.CAPTURE"),
)
field_info = ExtOutBitsFieldInfo(
field_type, field_subtype, desc, capture_labels, bits
)
return field_name, field_info
def _no_attributes(
self, field_name: str, field_type: str, field_subtype: Optional[str]
) -> _FieldGeneratorType:
"""Calling this method indicates type-subtype pair is known and
has no attributes, so only a description needs to be retrieved"""
desc = yield from self._get_desc(field_name).execute()
return field_name, FieldInfo(field_type, field_subtype, desc)
def execute(self) -> ExchangeGenerator[Dict[str, FieldInfo]]:
ex = Exchange(f"{self.block}.*?")
yield ex
unsorted: Dict[int, Tuple[str, FieldInfo]] = {}
field_generators: List[ExchangeGenerator] = []
for line in ex.multiline:
field_name, index, type_subtype = line.split(maxsplit=2)
field_type: str
subtype: Optional[str]
# Append "None" to list below so there are always at least 2 elements
# so we can always unpack into subtype, even if no split occurs.
field_type, subtype, *_ = [*type_subtype.split(maxsplit=1), None]
# Always create default FieldInfo. If necessary we will replace it later
# with a more type-specific version.
field_info = FieldInfo(field_type, subtype, None)
if self.extended_metadata:
try:
# Construct the list of type-specific generators
field_generators.append(
self._commands_map[(field_type, subtype)](
field_name, field_type, subtype
)
)
except KeyError:
# This exception will be hit if PandA ever defines new types
logging.exception(
f"Unknown type {(field_type, subtype)} detected for "
f"{field_name}, cannot retrieve extended information for it."
)
# We can assume the new field will have a description though
field_generators.append(
self._no_attributes(field_name, field_type, subtype)
)
# Keep track of order of fields as returned by PandA. Important for later
# matching descriptions back to their field.
unsorted[int(index)] = (field_name, field_info)
# Dict keeps insertion order, so insert in the order the server said
fields = {name: field for _, (name, field) in sorted(unsorted.items())}
if self.extended_metadata is False:
# Asked to not perform the requests for extra metadata.
return fields
field_name_info: Tuple[Tuple[str, FieldInfo], ...]
field_name_info = yield from _zip_with_return(field_generators)
fields.update(field_name_info)
return fields
[docs]
class GetPcapBitsLabels(Command):
"""Get the labels for the bit fields in PCAP.
For example::
GetPcapBitsLabels() -> {"BITS0" : ["TTLIN1.VAL", "TTLIN2.VAL", ...], ...}
"""
def execute(self) -> ExchangeGenerator[Dict[str, List[str]]]:
ex = Exchange("PCAP.*?")
yield ex
bits_fields = []
for line in ex.multiline:
split = line.split()
if len(split) == 4:
field_name, _, field_type, field_subtype = split
if field_type == "ext_out" and field_subtype == "bits":
bits_fields.append(f"PCAP.{field_name}")
exchanges = [Exchange(f"{field}.BITS?") for field in bits_fields]
yield exchanges
bits = {field: ex.multiline for field, ex in zip(bits_fields, exchanges)}
return bits
[docs]
class ChangeGroup(Enum):
"""Which group of values to ask for ``*CHANGES`` on:
https://pandablocks-server.readthedocs.io/en/latest/commands.html#system-commands
"""
#: All the groups below
ALL = ""
#: Configuration settings
CONFIG = ".CONFIG"
#: Bits on the system bus
BITS = ".BITS"
#: Positions
POSN = ".POSN"
#: Polled read values
READ = ".READ"
#: Attributes (included capture enable flags)
ATTR = ".ATTR"
#: Table changes
TABLE = ".TABLE"
#: Table changes
METADATA = ".METADATA"
[docs]
@dataclass
class GetChanges(Command[Changes]):
"""Get a `Changes` object showing which fields have changed since the last
time this was called
Args:
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",...]}
)
"""
group: ChangeGroup = ChangeGroup.ALL
get_multiline: bool = False
def execute(self) -> ExchangeGenerator[Changes]:
ex = Exchange(f"*CHANGES{self.group.value}?")
yield ex
changes = Changes({}, [], [], {})
multivalue_get_commands: List[Tuple[str, GetMultiline]] = []
for line in ex.multiline:
if line[-1] == "<":
if self.get_multiline:
field = line[0:-1]
multivalue_get_commands.append((field, GetMultiline(field)))
else:
changes.no_value.append(line[:-1])
elif line.endswith("(error)"):
changes.in_error.append(line.split(" ", 1)[0])
else:
field, value = line.split("=", maxsplit=1)
changes.values[field] = value
if self.get_multiline:
multiline_vals = yield from _execute_commands(
*[item[1] for item in multivalue_get_commands]
)
for field, value in zip(
[item[0] for item in multivalue_get_commands], multiline_vals
):
assert isinstance(value, list)
changes.multiline_values[field] = value
return changes
[docs]
@dataclass
class GetState(Command[List[str]]):
"""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",
]
"""
def execute(self) -> ExchangeGenerator[List[str]]:
# TODO: explain in detail how this works
# See: references/how-it-works
attr, config, table, metadata = yield from _execute_commands(
GetChanges(ChangeGroup.ATTR),
GetChanges(ChangeGroup.CONFIG),
GetChanges(ChangeGroup.TABLE),
GetChanges(ChangeGroup.METADATA),
)
# Add the single line values
line_values = dict(**attr.values, **config.values, **metadata.values)
state = [f"{k}={v}" for k, v in line_values.items()]
# Get the multiline values
multiline_keys, commands = [], []
for field_name in table.no_value:
# Get tables as base64
multiline_keys.append(f"{field_name}<B")
commands.append(GetMultiline(f"{field_name}.B"))
for field_name in metadata.no_value:
# Get metadata as string list
multiline_keys.append(f"{field_name}<")
commands.append(GetMultiline(f"{field_name}"))
multiline_values = yield from _execute_commands(*commands)
for k, v in zip(multiline_keys, multiline_values):
state += [k] + v + [""]
return state
[docs]
@dataclass
class SetState(Command[None]):
"""Set the state of all the fields in a PandA
Args:
state: A list of raw lines as produced by `GetState`
For example::
SetState([
"SEQ1.TABLE<B"
"234fds0SDklnmnr"
""
"PCAP.TRIG=PULSE1.OUT",
])
"""
state: List[str]
def execute(self) -> ExchangeGenerator[None]:
commands: List[Raw] = []
command_lines: List[str] = []
for line in self.state:
command_lines.append(line)
first_line = len(command_lines) == 1
if (first_line and not is_multiline_command(line)) or not line:
# If not a multiline command
# Or blank line at the end of a multiline command
commands.append(Raw(command_lines))
command_lines = []
returns = yield from _execute_commands(*commands)
for command, ret in zip(commands, returns):
if ret != ["OK"]:
logging.warning(f"command {command.inp} failed with {ret}")
