added types for the field elements

Also tried to make an `EpicsString` type capable of being used as a hash.

src/fastcs_pandablocks/__main__.py

@@ -2,8 +2,11 @@
 
 import argparse
 import logging
+import asyncio
+
+#from fastcs_pandablocks.fastcs import ioc
+from fastcs_pandablocks.panda.panda import Panda
 
-from fastcs_pandablocks.fastcs import ioc
 
 from . import __version__
 
@@ -49,12 +52,19 @@ def main():
     level = getattr(logging, parsed_args.log_level.upper(), None)
     logging.basicConfig(format="%(levelname)s:%(message)s", level=level)
 
+    async def meh():
+        await Panda(parsed_args.host).connect()
+    asyncio.run(meh())
+
+
+    """
     ioc(
         parsed_args.host,
         parsed_args.prefix,
         parsed_args.screens_dir,
         parsed_args.clear_bobfiles,
     )
+    """
 
 
 if __name__ == "__main__":

src/fastcs_pandablocks/panda/blocks.py

@@ -0,0 +1,227 @@
+import itertools
+from pprint import pprint
+from typing import Type
+from fastcs_pandablocks.types import EpicsName, ResponseType
+
+class Field:
+        def __init__(self, name: EpicsName, field_info: ResponseType):
+                self.name = name
+                self.field_info = field_info
+
+def change_value(self, new_field_value):
+        print("setting value", new_field_value)
+        self.value = new_field_value
+
+class TableField(Field):
+        ...
+
+class TimeField(Field):
+        ...
+
+class BitOutField(Field):
+        ...
+
+class PosOutField(Field):
+        ...
+
+class ExtOutField(Field):
+        ...
+
+class ExtOutBitsField(ExtOutField):
+        ...
+
+class BitMuxField(Field):
+        ...
+
+class PosMuxField(Field):
+        ...
+
+class UintParamField(Field):
+        ...
+
+class UintReadField(Field):
+        ...
+
+class UintWriteField(Field):
+        ...
+
+class IntParamField(Field):
+        ...
+
+class IntReadField(Field):
+        ...
+
+class IntWriteField(Field):
+        ...
+
+class ScalarParamField(Field):
+        ...
+
+class ScalarReadField(Field):
+        ...
+
+class ScalarWriteField(Field):
+        ...
+
+class BitParamField(Field):
+        ...
+
+class BitWriteField(Field):
+        ...
+
+class BitReadField(Field):
+        ...
+
+class ActionWriteField(Field):
+        ...
+
+class ActionReadField(Field):
+        ...
+
+class LutParamField(Field):
+        ...
+
+class LutWriteField(Field):
+        ...
+
+class LutReadField(Field):
+        ...
+
+class EnumParamField(Field):
+        ...
+
+class EnumWriteField(Field):
+        ...
+
+class EnumReadField(Field):
+        ...
+
+class TimeSubTypeParamField(Field):
+        ...
+
+class TimeSubTypeReadField(Field):
+        ...
+
+class TimeSubTypeWriteField(Field):
+        ...
+
+FieldType = (
+    TableField
+    | BitParamField
+    | BitWriteField
+    | BitReadField
+    | ActionWriteField
+    | ActionReadField
+    | LutParamField
+    | LutWriteField
+    | LutReadField
+    | EnumParamField
+    | EnumWriteField
+    | EnumReadField
+    | TimeSubTypeParamField
+    | TimeSubTypeReadField
+    | TimeSubTypeWriteField
+    | TimeField
+    | BitOutField
+    | PosOutField
+    | ExtOutField
+    | ExtOutBitsField
+    | BitMuxField
+    | PosMuxField
+    | UintParamField
+    | UintReadField
+    | UintWriteField
+    | IntParamField
+    | IntReadField
+    | IntWriteField
+    | ScalarParamField
+    | ScalarReadField
+    | ScalarWriteField
+)
+
+FIELD_TYPE_TO_FASTCS_TYPE: dict[str, dict[str | None, Type[FieldType]]] = {
+    "table": {
+        None: TableField
+    },
+    "time": {
+        None: TimeField,
+        "param": TimeSubTypeParamField,
+        "read": TimeSubTypeReadField,
+        "write": TimeSubTypeWriteField,
+    },
+    "bit_out": {
+        None: BitOutField,
+    },
+    "pos_out": {
+        None: PosOutField,
+    },
+    "ext_out": {
+        "timestamp": ExtOutField,
+        "samples": ExtOutField,
+        "bits": ExtOutBitsField,
+    },
+    "bit_mux": {
+        None: BitMuxField,
+    },
+    "pos_mux": {
+        None: PosMuxField,
+    },
+    "param": {
+        "uint": UintParamField,
+        "int": IntParamField,
+        "scalar": ScalarParamField,
+        "bit": BitParamField,
+        "action": ActionReadField,
+        "lut": LutParamField,
+        "enum": EnumParamField,
+        "time": TimeSubTypeParamField,
+    },
+    "read": {
+        "uint": UintReadField,
+        "int": IntReadField,
+        "scalar": ScalarReadField,
+        "bit": BitReadField,
+        "action": ActionReadField,
+        "lut": LutReadField,
+        "enum": EnumReadField,
+        "time": TimeSubTypeReadField,
+    },
+    "write": {
+        "uint": UintWriteField,
+        "int": IntWriteField,
+        "scalar": ScalarWriteField,
+        "bit": BitWriteField,
+        "action": ActionWriteField,
+        "lut": LutWriteField,
+        "enum": EnumWriteField,
+        "time": TimeSubTypeWriteField,
+    },
+}
+
+
+class Block:
+    _sub_blocks: dict[int, dict[EpicsName, FieldType]]
+
+    def __init__(self, name: EpicsName, number: int, description: str | None, raw_fields: dict[str, ResponseType]):
+        self.name = name
+        self.number = number
+        self.description = description
+        self._sub_blocks = {}
+
+        for number in range(1, number + 1):
+            numbered_block = name + EpicsName(str(number))
+            single_block = self._sub_blocks[number] = {}
+
+            for field_suffix, field_info in (
+                raw_fields.items()
+            ):
+                field_name = (
+                       numbered_block + EpicsName(field_suffix)
+                )
+                single_block[EpicsName(field_suffix)] = (
+                    FIELD_TYPE_TO_FASTCS_TYPE[field_info.type][field_info.subtype](
+                        field_name, field_info
+                    )
+                )
+    def change_value(self, new_field_value, block_number, field_name):
+        self._sub_blocks[block_number][field_name].change_value(new_field_value)

src/fastcs_pandablocks/panda/client_wrapper.py

@@ -14,39 +14,12 @@
     GetFieldInfo,
 )
 from pandablocks.responses import (
-    BitMuxFieldInfo,
-    BitOutFieldInfo,
     BlockInfo,
     Changes,
-    EnumFieldInfo,
-    ExtOutBitsFieldInfo,
-    ExtOutFieldInfo,
-    FieldInfo,
-    PosMuxFieldInfo,
-    PosOutFieldInfo,
-    ScalarFieldInfo,
-    SubtypeTimeFieldInfo,
-    TableFieldInfo,
-    TimeFieldInfo,
-    UintFieldInfo,
 )
-from typing import Union
 
-ResponseType = Union[
-    BitMuxFieldInfo,
-    BitOutFieldInfo,
-    EnumFieldInfo,
-    ExtOutBitsFieldInfo,
-    ExtOutFieldInfo,
-    FieldInfo,
-    PosMuxFieldInfo,
-    PosOutFieldInfo,
-    ScalarFieldInfo,
-    SubtypeTimeFieldInfo,
-    TableFieldInfo,
-    TimeFieldInfo,
-    UintFieldInfo, 
-]
+from fastcs_pandablocks.types import ResponseType
+
 
 class RawPanda:
     _blocks: dict[str, BlockInfo] | None = None
@@ -96,4 +69,3 @@ def __aiter__(self):
     async def __anext__(self):
         await self._ensure_connected()
         return await self.get_changes()
-

src/fastcs_pandablocks/panda/panda.py

@@ -0,0 +1,51 @@
+from pprint import pprint
+from typing import Callable
+from dataclasses import dataclass
+from .client_wrapper import RawPanda
+from .blocks import Block
+from fastcs_pandablocks.types import EpicsName, PandaName
+from pandablocks.responses import Changes
+
+
+
+class Panda:
+    _raw_panda: RawPanda
+    _blocks: dict[EpicsName, Block]
+
+    def __init__(self, host: str):
+        self._raw_panda = RawPanda(host)
+        self._blocks = {}
+
+    async def connect(self):
+        await self._raw_panda._sync_with_panda()
+        self._parse_introspected_data()
+
+    def _parse_introspected_data(self):
+        self._blocks = {}
+        if (
+            self._raw_panda._blocks is None or self._raw_panda._responses is None
+        ):
+            raise ValueError("Panda not introspected.")
+
+        for (block_name, block_info), raw_fields in zip(
+            self._raw_panda._blocks.items(), self._raw_panda._responses
+        ):
+            self._blocks[EpicsName(block=block_name)] = Block(
+                EpicsName(block_name),
+                block_info.number,
+                block_info.description,
+                raw_fields
+            )
+
+
+    def _parse_values(self, changes: Changes):
+        for panda_name, field_value in changes.values.items():
+            epics_name = PandaName(panda_name).to_epics_name()
+            self._blocks[epics_name.block].change_value(
+                field_value, epics_name.block_number, epics_name.field
+            )
+
+
+
+
+    async def disconnect(self): await self._raw_panda.disconnect()