started the structure and async iterator parsing the panda

pyproject.toml

@@ -12,7 +12,13 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 description = "A softioc to control a PandABlocks-FPGA."
-dependencies = [] # Add project dependencies here, e.g. ["click", "numpy"]
+dependencies = [
+    "fastcs~=0.6.0",
+    "pandablocks~=0.10.0",
+    "numpy<2",    # until https://github.com/mdavidsaver/p4p/issues/145 is fixed
+    "pydantic>2",
+    "h5py",
+] 
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"
@@ -37,7 +43,7 @@ dev = [
 ]
 
 [project.scripts]
-fastcs-PandABlocks = "fastcs_pandablocks.__main__:main"
+fastcs-pandablocks = "fastcs_pandablocks.__main__:main"
 
 [project.urls]
 GitHub = "https://github.com/PandABlocks-ioc/fastcs-PandABlocks"

src/fastcs_pandablocks/__main__.py

@@ -1,23 +1,60 @@
 """Interface for ``python -m fastcs_pandablocks``."""
 
-from argparse import ArgumentParser
-from collections.abc import Sequence
+import argparse
+import logging
+
+from fastcs_pandablocks.fastcs import ioc
 
 from . import __version__
 
 __all__ = ["main"]
 
 
-def main(args: Sequence[str] | None = None) -> None:
+def main():
     """Argument parser for the CLI."""
-    parser = ArgumentParser()
+    parser = argparse.ArgumentParser(
+        description="Connect to the given HOST and create an IOC with the given PREFIX."
+    )
+    parser.add_argument("host", type=str, help="The host to connect to.")
+    parser.add_argument("prefix", type=str, help="The prefix for the IOC.")
+    parser.add_argument(
+        "--screens-dir",
+        type=str,
+        help=(
+            "Provide an existing directory to export generated bobfiles to, if no "
+            "directory is provided then bobfiles will not be generated."
+        ),
+    )
+    parser.add_argument(
+        "--clear-bobfiles",
+        action="store_true",
+        help="Clear bobfiles from the given `screens-dir` before generating new ones.",
+    )
     parser.add_argument(
         "-v",
         "--version",
         action="version",
         version=__version__,
     )
-    parser.parse_args(args)
+    parser.add_argument(
+        "--log-level",
+        default="INFO",
+        choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"],
+        help="Set the logging level.",
+    )
+
+    parsed_args = parser.parse_args()
+
+    # Set the logging level
+    level = getattr(logging, parsed_args.log_level.upper(), None)
+    logging.basicConfig(format="%(levelname)s:%(message)s", level=level)
+
+    ioc(
+        parsed_args.host,
+        parsed_args.prefix,
+        parsed_args.screens_dir,
+        parsed_args.clear_bobfiles,
+    )
 
 
 if __name__ == "__main__":

src/fastcs_pandablocks/fastcs/__init__.py

@@ -0,0 +1,34 @@
+"""Contains logic relevant to fastcs. Will use `fastcs_pandablocks.panda`."""
+
+
+from pathlib import Path
+
+from fastcs.backends.epics.backend import EpicsBackend
+
+from .gui import PandaGUIOptions
+from .controller import PandaController
+from fastcs_pandablocks.types import EpicsName
+
+
+def ioc(
+    panda_hostname: str,
+    pv_prefix: EpicsName,
+    screens_directory: Path | None,
+    clear_bobfiles: bool = False,
+):
+    controller = PandaController(panda_hostname)
+    backend = EpicsBackend(controller, pv_prefix=str(pv_prefix))
+
+    if clear_bobfiles and not screens_directory:
+        raise ValueError("`clear_bobfiles` is True with no `screens_directory`")
+
+    if screens_directory:
+        if not screens_directory.is_dir():
+            raise ValueError(
+                f"`screens_directory` {screens_directory} is not a directory"
+            )
+        backend.create_gui(
+            PandaGUIOptions()
+        )
+
+    backend.run()

src/fastcs_pandablocks/fastcs/controller.py

@@ -0,0 +1,14 @@
+# TODO: tackle after I have a MVP of the panda part.
+from fastcs.controller import Controller
+from fastcs.datatypes import Bool, Float, Int, String
+
+
+class PandaController(Controller):
+    def __init__(self, hostname: str) -> None:
+        super().__init__()
+
+    async def initialise(self) -> None:
+        pass
+
+    async def connect(self) -> None:
+        pass

src/fastcs_pandablocks/fastcs/gui.py

@@ -0,0 +1,4 @@
+from fastcs.backends.epics.gui import EpicsGUIOptions
+
+class PandaGUIOptions(EpicsGUIOptions):
+    ...

src/fastcs_pandablocks/panda/__init__.py

@@ -0,0 +1 @@
+"""Contains the logic relevant to the Panda's operation."""

src/fastcs_pandablocks/panda/client_wrapper.py

@@ -0,0 +1,99 @@
+"""
+Over the years we've had to add little adjustments on top of the `BlockInfo`, `BlockAndFieldInfo`, etc.
+
+This method has a `RawPanda` which handles all the io with the client.
+"""
+
+import asyncio
+from pandablocks.asyncio import AsyncioClient
+from pandablocks.commands import (
+    ChangeGroup,
+    Changes,
+    GetBlockInfo,
+    GetChanges,
+    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, 
+]
+
+class RawPanda:
+    _blocks: dict[str, BlockInfo] | None = None
+    _metadata: tuple[Changes] | None = None
+
+    _responses: list[dict[str, ResponseType]] | None = None
+    _changes: Changes | None = None
+
+    def __init__(self, host: str):
+        self._client = AsyncioClient(host)
+
+    async def connect(self): await self._client.connect()
+
+    async def disconnect(self): await self._client.close()
+
+    async def introspect(self):
+        self._blocks = await self._client.send(GetBlockInfo())
+        self._responses = await asyncio.gather(
+            *[self._client.send(GetFieldInfo(block)) for block in self._blocks],
+        )
+        self._metadata = await self._client.send(GetChanges(ChangeGroup.ALL, True)),
+
+    async def get_changes(self):
+        self._changes = await self._client.send(GetChanges(ChangeGroup.ALL, False))
+
+
+    async def _sync_with_panda(self):
+        if not self._client.is_connected():
+            await self.connect()
+        await self.introspect()
+
+    async def _ensure_connected(self):
+        if not self._blocks:
+            await self._sync_with_panda()
+
+    async def __aenter__(self):
+        await self._sync_with_panda()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self._ensure_connected()
+        await self.disconnect()
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        await self._ensure_connected()
+        return await self.get_changes()
+

src/fastcs_pandablocks/types.py

@@ -0,0 +1,134 @@
+from dataclasses import dataclass
+import re
+from pydantic import BaseModel
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+# Dataclasses for names
+
+@dataclass
+class _Name:
+    _name: str
+
+    def __str__(self):
+        return str(self._name)
+
+class PandaName(_Name):
+    def to_epics_name(self):
+        return EpicsName(self._name.replace(".", ":"))
+
+class EpicsName(_Name):
+    def to_panda_name(self):
+        return PandaName(self._name.replace(":", "."))
+
+    def to_pvi_name(self):
+        relevant_section = self._name.split(":")[-1]
+        words = relevant_section.replace("-", "_").split("_")
+        capitalised_word = "".join(word.capitalize() for word in words)
+
+        # We don't want to allow any non-alphanumeric characters.
+        formatted_word = re.search(r"[A-Za-z0-9]+", capitalised_word)
+        assert formatted_word
+
+        return PviName(formatted_word.group())
+
+class PviName(_Name):
+    ...
+
+
+
+Field_T = Literal[
+    "time",
+    "bit_out",
+    "pos_out",
+    "ext_out",
+    "bit_mux",
+    "pos_mux",
+    "param",
+    "read",
+    "write",
+]
+
+FieldSubtype_T = Literal[
+    "timestamp",
+    "samples",
+    "bits",
+    "uint",
+    "int",
+    "scalar",
+    "bit",
+    "action",
+    "lut",
+    "enum",
+    "time",
+]
+
+
+class PandaField(BaseModel, frozen=True):
+    """Validates fields from the client."""
+
+    field_type: Field_T
+    field_subtype: FieldSubtype_T | None
+
+
+TIME_FIELDS = {
+    PandaField(field_type="time", field_subtype=None),
+}
+
+BIT_OUT_FIELDS = {
+    PandaField(field_type="bit_out", field_subtype=None),
+}
+
+POS_OUT_FIELDS = {
+    PandaField(field_type="pos_out", field_subtype=None),
+}
+
+EXT_OUT_FIELDS = {
+    PandaField(field_type="ext_out", field_subtype="timestamp"),
+    PandaField(field_type="ext_out", field_subtype="samples"),
+}
+
+EXT_OUT_BITS_FIELDS = {
+    PandaField(field_type="ext_out", field_subtype="bits"),
+}
+
+BIT_MUX_FIELDS = {
+    PandaField(field_type="bit_mux", field_subtype=None),
+}
+
+POS_MUX_FIELDS = {
+    PandaField(field_type="pos_mux", field_subtype=None),
+}
+
+UINT_FIELDS = {
+    PandaField(field_type="param", field_subtype="uint"),
+    PandaField(field_type="read", field_subtype="uint"),
+    PandaField(field_type="write", field_subtype="uint"),
+}
+
+INT_FIELDS = {
+    PandaField(field_type="param", field_subtype="int"),
+    PandaField(field_type="read", field_subtype="int"),
+    PandaField(field_type="write", field_subtype="int"),
+}
+
+SCALAR_FIELDS = {
+    PandaField(field_type="param", field_subtype="scalar"),
+    PandaField(field_type="read", field_subtype="scalar"),
+    PandaField(field_type="write", field_subtype="scalar"),
+}
+
+BIT_FIELDS = {
+    PandaField(field_type="param", field_subtype="bit"),
+    PandaField(field_type="read", field_subtype="bit"),
+    PandaField(field_type="write", field_subtype="bit"),
+}
+
+ACTION_FIELDS = {
+    PandaField(field_type="param", field_subtype="action"),
+    PandaField(field_type="read", field_subtype="action"),
+    PandaField(field_type="write", field_subtype="action"),
+}
+