Repository: PandABlocks/PandABlocks-client

Issue:
Add helper function converts Sequencer Table Line into an integer
When we use `PandABlocks-client`to update/append sequencer table lines, we have to convert each sequencer table line from __Name__ into __Bit Field__ according to the [sequencer table line composition](http://localhost:8008/fpga_docs//build/seq_doc.html).

We created a function to convert and hope it can be added to the upstream.

Discussion:

Patch:
diff --git a/docs/reference/api.rst b/docs/reference/api.rst
index d804c0b..b9071f8 100644
--- a/docs/reference/api.rst
+++ b/docs/reference/api.rst
@@ -12,6 +12,7 @@ from code:
 - `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
 
 
 .. automodule:: pandablocks.commands
@@ -82,3 +83,10 @@ from code:
     gives multi-CPU benefits without hitting the limit of the GIL.
 
     .. seealso:: `library-hdf`, `performance`
+
+.. automodule:: pandablocks.utils    
+    
+    Utilities
+    ---------
+
+    This package contains general methods for working with pandablocks.
diff --git a/pandablocks/utils.py b/pandablocks/utils.py
new file mode 100644
index 0000000..7970731
--- /dev/null
+++ b/pandablocks/utils.py
@@ -0,0 +1,125 @@
+from typing import Dict, Iterable, List, Sequence, Union, cast
+
+import numpy as np
+import numpy.typing as npt
+
+from pandablocks.responses import TableFieldInfo
+
+UnpackedArray = Union[
+    npt.NDArray[np.int32],
+    npt.NDArray[np.uint32],
+    Sequence[str],
+]
+
+
+def words_to_table(
+    words: Iterable[str], table_field_info: TableFieldInfo
+) -> Dict[str, UnpackedArray]:
+    """Unpacks the given `packed` data based on the fields provided.
+    Returns the unpacked data in {column_name: column_data} column-indexed format
+
+    Args:
+        words: An iterable of data for this table, from PandA. Each item is
+            expected to be the string representation of a uint32.
+        table_fields_info: The info for tables, containing the number of words per row,
+            and the bit information for fields.
+    Returns:
+        unpacked: A dict containing record information, where keys are field names
+            and values are numpy arrays or a sequence of strings of record values
+            in that column.
+    """
+
+    row_words = table_field_info.row_words
+    data = np.array(words, dtype=np.uint32)
+    # Convert 1-D array into 2-D, one row element per row in the PandA table
+    data = data.reshape(len(data) // row_words, row_words)
+    packed = data.T
+
+    unpacked: Dict[str, UnpackedArray] = {}
+
+    for field_name, field_info in table_field_info.fields.items():
+        offset = field_info.bit_low
+        bit_length = field_info.bit_high - field_info.bit_low + 1
+
+        # The word offset indicates which column this field is in
+        # (column is exactly one 32-bit word)
+        word_offset = offset // 32
+
+        # bit offset is location of field inside the word
+        bit_offset = offset & 0x1F
+
+        # Mask to remove every bit that isn't in the range we want
+        mask = (1 << bit_length) - 1
+
+        value = (packed[word_offset] >> bit_offset) & mask
+
+        if field_info.subtype == "int":
+            # First convert from 2's complement to offset, then add in offset.
+            packing_value = (value ^ (1 << (bit_length - 1))) + (-1 << (bit_length - 1))
+        elif field_info.labels:
+            packing_value = [field_info.labels[x] for x in value]
+        else:
+            packing_value = value
+
+        unpacked.update({field_name: packing_value})
+
+    return unpacked
+
+
+def table_to_words(
+    table: Dict[str, Iterable], table_field_info: TableFieldInfo
+) -> List[str]:
+    """Convert records based on the field definitions into the format PandA expects
+    for table writes.
+
+    Args:
+        table: A dict containing record information, where keys are field names
+            and values are iterables of record values in that column.
+        table_field_info: The info for tables, containing the dict `fields` for
+            information on each field, and the number of words per row.
+    Returns:
+        List[str]: The list of data ready to be sent to PandA
+    """
+    row_words = table_field_info.row_words
+
+    # Iterate over the zipped fields and their associated records to construct the
+    # packed array.
+    packed = None
+
+    for column_name, column in table.items():
+        field_details = table_field_info.fields[column_name]
+        if field_details.labels:
+            # Must convert the list of ints into strings
+            column = [field_details.labels.index(x) for x in column]
+
+        # PandA always handles tables in uint32 format
+        column_value = np.array(column, dtype=np.uint32)
+
+        if packed is None:
+            # Create 1-D array sufficiently long to exactly hold the entire table, cast
+            # to prevent type error, this will still work if column is another iterable
+            # e.g numpy array
+            column = cast(List, column)
+            packed = np.zeros((len(column), row_words), dtype=np.uint32)
+        else:
+            assert len(packed) == len(column), (
+                f"Table record {column_name} has mismatched length "
+                "compared to other records, cannot pack data"
+            )
+
+        offset = field_details.bit_low
+
+        # The word offset indicates which column this field is in
+        # (each column is one 32-bit word)
+        word_offset = offset // 32
+        # bit offset is location of field inside the word
+        bit_offset = offset & 0x1F
+
+        # Slice to get the column to apply the values to.
+        # bit shift the value to the relevant bits of the word
+        packed[:, word_offset] |= column_value << bit_offset
+
+    assert isinstance(packed, np.ndarray), "Table has no columns"  # Squash mypy warning
+
+    # 2-D array -> 1-D array -> list[int] -> list[str]
+    return [str(x) for x in packed.flatten().tolist()]
diff --git a/pyproject.toml b/pyproject.toml
index e6e5a15..64253b6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,4 +1,4 @@
 [build-system]
 # Pin versions compatible with dls-python3 for reproducible wheels
 requires = ["setuptools==44.1.1", "wheel==0.33.1"]
-build-backend = "setuptools.build_meta"
\ No newline at end of file
+build-backend = "setuptools.build_meta"


Test patch:
diff --git a/tests/test_utils.py b/tests/test_utils.py
new file mode 100644
index 0000000..6057dd6
--- /dev/null
+++ b/tests/test_utils.py
@@ -0,0 +1,357 @@
+from typing import Dict, Iterable, List, OrderedDict
+
+import numpy as np
+import pytest
+
+from pandablocks.responses import TableFieldDetails, TableFieldInfo
+from pandablocks.utils import UnpackedArray, table_to_words, words_to_table
+
+
+@pytest.fixture
+def table_fields() -> Dict[str, TableFieldDetails]:
+    """Table field definitions, taken from a SEQ.TABLE instance.
+    Associated with table_data and table_field_info fixtures"""
+    return {
+        "REPEATS": TableFieldDetails(
+            subtype="uint",
+            bit_low=0,
+            bit_high=15,
+            description="Number of times the line will repeat",
+            labels=None,
+        ),
+        "TRIGGER": TableFieldDetails(
+            subtype="enum",
+            bit_low=16,
+            bit_high=19,
+            description="The trigger condition to start the phases",
+            labels=[
+                "Immediate",
+                "BITA=0",
+                "BITA=1",
+                "BITB=0",
+                "BITB=1",
+                "BITC=0",
+                "BITC=1",
+                "POSA>=POSITION",
+                "POSA<=POSITION",
+                "POSB>=POSITION",
+                "POSB<=POSITION",
+                "POSC>=POSITION",
+                "POSC<=POSITION",
+            ],
+        ),
+        "POSITION": TableFieldDetails(
+            subtype="int",
+            bit_low=32,
+            bit_high=63,
+            description="The position that can be used in trigger condition",
+            labels=None,
+        ),
+        "TIME1": TableFieldDetails(
+            subtype="uint",
+            bit_low=64,
+            bit_high=95,
+            description="The time the optional phase 1 should take",
+            labels=None,
+        ),
+        "OUTA1": TableFieldDetails(
+            subtype="uint",
+            bit_low=20,
+            bit_high=20,
+            description="Output A value during phase 1",
+            labels=None,
+        ),
+        "OUTB1": TableFieldDetails(
+            subtype="uint",
+            bit_low=21,
+            bit_high=21,
+            description="Output B value during phase 1",
+            labels=None,
+        ),
+        "OUTC1": TableFieldDetails(
+            subtype="uint",
+            bit_low=22,
+            bit_high=22,
+            description="Output C value during phase 1",
+            labels=None,
+        ),
+        "OUTD1": TableFieldDetails(
+            subtype="uint",
+            bit_low=23,
+            bit_high=23,
+            description="Output D value during phase 1",
+            labels=None,
+        ),
+        "OUTE1": TableFieldDetails(
+            subtype="uint",
+            bit_low=24,
+            bit_high=24,
+            description="Output E value during phase 1",
+            labels=None,
+        ),
+        "OUTF1": TableFieldDetails(
+            subtype="uint",
+            bit_low=25,
+            bit_high=25,
+            description="Output F value during phase 1",
+            labels=None,
+        ),
+        "TIME2": TableFieldDetails(
+            subtype="uint",
+            bit_low=96,
+            bit_high=127,
+            description="The time the mandatory phase 2 should take",
+            labels=None,
+        ),
+        "OUTA2": TableFieldDetails(
+            subtype="uint",
+            bit_low=26,
+            bit_high=26,
+            description="Output A value during phase 2",
+            labels=None,
+        ),
+        "OUTB2": TableFieldDetails(
+            subtype="uint",
+            bit_low=27,
+            bit_high=27,
+            description="Output B value during phase 2",
+            labels=None,
+        ),
+        "OUTC2": TableFieldDetails(
+            subtype="uint",
+            bit_low=28,
+            bit_high=28,
+            description="Output C value during phase 2",
+            labels=None,
+        ),
+        "OUTD2": TableFieldDetails(
+            subtype="uint",
+            bit_low=29,
+            bit_high=29,
+            description="Output D value during phase 2",
+            labels=None,
+        ),
+        "OUTE2": TableFieldDetails(
+            subtype="uint",
+            bit_low=30,
+            bit_high=30,
+            description="Output E value during phase 2",
+            labels=None,
+        ),
+        "OUTF2": TableFieldDetails(
+            subtype="uint",
+            bit_low=31,
+            bit_high=31,
+            description="Output F value during phase 2",
+            labels=None,
+        ),
+    }
+
+
+@pytest.fixture
+def table_field_info(table_fields) -> TableFieldInfo:
+    """Table data associated with table_fields and table_data fixtures"""
+    return TableFieldInfo(
+        "table", None, "Sequencer table of lines", 16384, table_fields, 4
+    )
+
+
+@pytest.fixture
+def table_1() -> OrderedDict[str, Iterable]:
+    return OrderedDict(
+        {
+            "REPEATS": [5, 0, 50000],
+            "TRIGGER": ["Immediate", "BITC=1", "Immediate"],
+            "POSITION": [-5, 678, 0],
+            "TIME1": [100, 0, 9],
+            "OUTA1": [0, 1, 1],
+            "OUTB1": [0, 0, 1],
+            "OUTC1": [0, 0, 1],
+            "OUTD1": [1, 0, 1],
+            "OUTE1": [0, 0, 1],
+            "OUTF1": [1, 0, 1],
+            "TIME2": [0, 55, 9999],
+            "OUTA2": [0, 0, 1],
+            "OUTB2": [0, 0, 1],
+            "OUTC2": [1, 1, 1],
+            "OUTD2": [0, 0, 1],
+            "OUTE2": [0, 0, 1],
+            "OUTF2": [1, 0, 1],
+        }
+    )
+
+
+@pytest.fixture
+def table_1_np_arrays() -> OrderedDict[str, Iterable]:
+    # Intentionally not in panda order. Whatever types the np arrays are,
+    # the outputs from words_to_table will be uint32 or int32.
+    return OrderedDict(
+        {
+            "POSITION": np.array([-5, 678, 0], dtype=np.int32),
+            "TIME1": np.array([100, 0, 9], dtype=np.uint32),
+            "OUTA1": np.array([0, 1, 1], dtype=np.uint8),
+            "OUTB1": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTD1": np.array([1, 0, 1], dtype=np.uint8),
+            "OUTE1": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTC1": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTF1": np.array([1, 0, 1], dtype=np.uint8),
+            "TIME2": np.array([0, 55, 9999], dtype=np.uint32),
+            "OUTA2": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTB2": np.array([0, 0, 1], dtype=np.uint8),
+            "REPEATS": np.array([5, 0, 50000], dtype=np.uint32),
+            "OUTC2": np.array([1, 1, 1], dtype=np.uint8),
+            "OUTD2": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTE2": np.array([0, 0, 1], dtype=np.uint8),
+            "OUTF2": np.array([1, 0, 1], dtype=np.uint8),
+            "TRIGGER": np.array(["Immediate", "BITC=1", "Immediate"], dtype="<U9"),
+        }
+    )
+
+
+@pytest.fixture
+def table_1_not_in_panda_order() -> OrderedDict[str, Iterable]:
+    return OrderedDict(
+        {
+            "REPEATS": [5, 0, 50000],
+            "TRIGGER": ["Immediate", "BITC=1", "Immediate"],
+            "POSITION": [-5, 678, 0],
+            "TIME1": [100, 0, 9],
+            "OUTA1": [0, 1, 1],
+            "OUTB1": [0, 0, 1],
+            "OUTC1": [0, 0, 1],
+            "OUTD1": [1, 0, 1],
+            "OUTF1": [1, 0, 1],
+            "OUTE1": [0, 0, 1],
+            "TIME2": [0, 55, 9999],
+            "OUTA2": [0, 0, 1],
+            "OUTC2": [1, 1, 1],
+            "OUTB2": [0, 0, 1],
+            "OUTD2": [0, 0, 1],
+            "OUTE2": [0, 0, 1],
+            "OUTF2": [1, 0, 1],
+        }
+    )
+
+
+@pytest.fixture
+def table_data_1() -> List[str]:
+    return [
+        "2457862149",
+        "4294967291",
+        "100",
+        "0",
+        "269877248",
+        "678",
+        "0",
+        "55",
+        "4293968720",
+        "0",
+        "9",
+        "9999",
+    ]
+
+
+@pytest.fixture
+def table_2() -> Dict[str, Iterable]:
+    table: Dict[str, Iterable] = dict(
+        REPEATS=[1, 0],
+        TRIGGER=["Immediate", "Immediate"],
+        POSITION=[-20, 2**31 - 1],
+        TIME1=[12, 2**32 - 1],
+        TIME2=[32, 1],
+    )
+
+    table["OUTA1"] = [False, True]
+    table["OUTA2"] = [True, False]
+    for key in "BCDEF":
+        table[f"OUT{key}1"] = table[f"OUT{key}2"] = [False, False]
+
+    return table
+
+
+@pytest.fixture
+def table_data_2() -> List[str]:
+    return [
+        "67108865",
+        "4294967276",
+        "12",
+        "32",
+        "1048576",
+        "2147483647",
+        "4294967295",
+        "1",
+    ]
+
+
+def test_table_packing_pack_length_mismatched(
+    table_1: OrderedDict[str, Iterable],
+    table_field_info: TableFieldInfo,
+):
+    assert table_field_info.row_words
+
+    # Adjust one of the record lengths so it mismatches
+    field_info = table_field_info.fields[("OUTC1")]
+    assert field_info
+    table_1["OUTC1"] = np.array([1, 2, 3, 4, 5, 6, 7, 8])
+
+    with pytest.raises(AssertionError):
+        table_to_words(table_1, table_field_info)
+
+
+@pytest.mark.parametrize(
+    "table_fixture_name,table_data_fixture_name",
+    [
+        ("table_1_not_in_panda_order", "table_data_1"),
+        ("table_2", "table_data_2"),
+        ("table_1_np_arrays", "table_data_1"),
+    ],
+)
+def test_table_to_words_and_words_to_table(
+    table_fixture_name: str,
+    table_data_fixture_name: str,
+    table_field_info: TableFieldInfo,
+    request,
+):
+    table: Dict[str, Iterable] = request.getfixturevalue(table_fixture_name)
+    table_data: List[str] = request.getfixturevalue(table_data_fixture_name)
+
+    output_data = table_to_words(table, table_field_info)
+    assert output_data == table_data
+    output_table = words_to_table(output_data, table_field_info)
+
+    # Test the correct keys are outputted
+    assert output_table.keys() == table.keys()
+
+    # Check the items have been inserted in panda order
+    sorted_output_table = OrderedDict({key: table[key] for key in output_table.keys()})
+    assert sorted_output_table != OrderedDict(table)
+
+    # Check the values are the same
+    for output_key in output_table.keys():
+        np.testing.assert_equal(output_table[output_key], table[output_key])
+
+
+def test_table_packing_unpack(
+    table_1_np_arrays: OrderedDict[str, np.ndarray],
+    table_field_info: TableFieldInfo,
+    table_data_1: List[str],
+):
+    assert table_field_info.row_words
+    output_table = words_to_table(table_data_1, table_field_info)
+
+    actual: UnpackedArray
+    for field_name, actual in output_table.items():
+        expected = table_1_np_arrays[str(field_name)]
+        np.testing.assert_array_equal(actual, expected)
+
+
+def test_table_packing_pack(
+    table_1: Dict[str, Iterable],
+    table_field_info: TableFieldInfo,
+    table_data_1: List[str],
+):
+    assert table_field_info.row_words
+    unpacked = table_to_words(table_1, table_field_info)
+
+    for actual, expected in zip(unpacked, table_data_1):
+        assert actual == expected