Update numpy requirement from <2 to <3 in the dev-dependencies group across 1 directory

.github/workflows/ci.yml

@@ -21,7 +21,7 @@ jobs:
     strategy:
       matrix:
         runs-on: ["ubuntu-latest"] # can add windows-latest, macos-latest
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.11", "3.12"]
         include:
           # Include one that runs in the dev environment
           - runs-on: "ubuntu-latest"

docs/tutorials/installation.md

@@ -2,7 +2,7 @@
 
 ## Check your version of python
 
-You will need python 3.10 or later. You can check your version of python by
+You will need python 3.11 or later. You can check your version of python by
 typing into a terminal:
 
 ```

pyproject.toml

@@ -7,16 +7,21 @@ name = "fastcs-pandablocks"
 classifiers = [
     "Development Status :: 3 - Alpha",
     "License :: OSI Approved :: Apache Software License",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
 ]
 description = "A softioc to control a PandABlocks-FPGA."
-dependencies = [] # Add project dependencies here, e.g. ["click", "numpy"]
+dependencies = [
+    "fastcs@git+https://github.com/DiamondLightSource/FastCS@panda-conversion-improvements",
+    "pandablocks~=0.10.0",
+    "numpy<3",    # until https://github.com/mdavidsaver/p4p/issues/145 is fixed
+    "pydantic>2",
+    "h5py",
+] 
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 
 [project.optional-dependencies]
 dev = [
@@ -37,7 +42,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"
@@ -114,3 +119,6 @@ lint.select = [
 # See https://github.com/DiamondLightSource/python-copier-template/issues/154
 # Remove this line to forbid private member access in tests
 "tests/**/*" = ["SLF001"]
+# Ruff wants us to use `(|)` instead of `Union[,]` in types, but pyright warns against
+# it. For now we'll use `Union` and turn the ruff error off.
+"**" = ["UP007"]

src/fastcs_pandablocks/__init__.py

@@ -1,11 +1,60 @@
-"""Top level API.
+"""Contains logic relevant to fastcs. Will use `fastcs_pandablocks.panda`."""
 
-.. data:: __version__
-    :type: str
+from pathlib import Path
 
-    Version number as calculated by https://github.com/pypa/setuptools_scm
-"""
+from fastcs.backends.epics.backend import EpicsBackend
+from fastcs.backends.epics.gui import EpicsGUIFormat
+from fastcs.backends.epics.ioc import EpicsIOCOptions
+from fastcs.backends.epics.util import EpicsNameOptions, PvNamingConvention
 
 from ._version import __version__
+from .gui import PandaGUIOptions
+from .panda.controller import PandaController
+from .types import EpicsName
 
-__all__ = ["__version__"]
+DEFAULT_POLL_PERIOD = 0.1
+
+
+def ioc(
+    epics_prefix: EpicsName,
+    hostname: str,
+    screens_directory: Path | None = None,
+    clear_bobfiles: bool = False,
+    poll_period: float = DEFAULT_POLL_PERIOD,
+    naming_convention: PvNamingConvention = PvNamingConvention.CAPITALIZED,
+    pv_separator: str = ":",
+):
+    name_options = EpicsNameOptions(
+        pv_naming_convention=naming_convention, pv_separator=pv_separator
+    )
+    epics_ioc_options = EpicsIOCOptions(terminal=True, name_options=name_options)
+
+    controller = PandaController(hostname, poll_period)
+    backend = EpicsBackend(
+        controller, pv_prefix=str(epics_prefix), ioc_options=epics_ioc_options
+    )
+
+    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"
+            )
+        if not clear_bobfiles:
+            if list(screens_directory.iterdir()):
+                raise RuntimeError("`screens_directory` is not empty.")
+
+        backend.create_gui(
+            PandaGUIOptions(
+                output_path=screens_directory / "output.bob",
+                file_format=EpicsGUIFormat.bob,
+                title="PandA",
+            )
+        )
+
+    backend.run()
+
+
+__all__ = ["__version__", "ioc", "DEFAULT_POLL_PERIOD"]

src/fastcs_pandablocks/__main__.py

@@ -1,23 +1,96 @@
 """Interface for ``python -m fastcs_pandablocks``."""
 
-from argparse import ArgumentParser
-from collections.abc import Sequence
+import argparse
+import logging
+from pathlib import Path
+
+from fastcs.backends.epics.util import PvNamingConvention
+
+from fastcs_pandablocks import DEFAULT_POLL_PERIOD, ioc
+from fastcs_pandablocks.types import EpicsName
 
 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(
         "-v",
         "--version",
         action="version",
         version=__version__,
     )
-    parser.parse_args(args)
+
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    run_parser = subparsers.add_parser(
+        "run", help="Run the IOC with the given HOST and PREFIX."
+    )
+    run_parser.add_argument("hostname", type=str, help="The host to connect to.")
+    run_parser.add_argument("prefix", type=str, help="The prefix for the IOC.")
+    run_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."
+        ),
+    )
+    run_parser.add_argument(
+        "--clear-bobfiles",
+        action="store_true",
+        help=(
+            "Overwrite existing bobfiles from the given `screens-dir` "
+            "before generating new ones."
+        ),
+    )
+
+    run_parser.add_argument(
+        "--log-level",
+        default="INFO",
+        choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"],
+        help="Set the logging level.",
+    )
+    run_parser.add_argument(
+        "--poll-period",
+        default=DEFAULT_POLL_PERIOD,
+        type=float,
+        help="Period in seconds with which to poll the panda.",
+    )
+    run_parser.add_argument(
+        "--pv-naming-convention",
+        default=PvNamingConvention.CAPITALIZED.name,
+        choices=[choice.name for choice in PvNamingConvention],
+        help="Naming convention of the EPICS PVs.",
+    )
+    run_parser.add_argument(
+        "--pv-separator",
+        default=":",
+        type=str,
+        help="Separator to use between EPICS PV sections.",
+    )
+
+    parsed_args = parser.parse_args()
+    if parsed_args.command != "run":
+        return
+
+    # Set the logging level
+    level = getattr(logging, parsed_args.log_level.upper(), None)
+    logging.basicConfig(format="%(levelname)s:%(message)s", level=level)
+
+    ioc(
+        EpicsName(prefix=parsed_args.prefix),
+        parsed_args.hostname,
+        screens_directory=Path(parsed_args.screens_dir),
+        clear_bobfiles=parsed_args.clear_bobfiles,
+        poll_period=parsed_args.poll_period,
+        naming_convention=PvNamingConvention(parsed_args.pv_naming_convention),
+        pv_separator=parsed_args.pv_separator,
+    )
 
 
 if __name__ == "__main__":

src/fastcs_pandablocks/gui.py

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

src/fastcs_pandablocks/handlers.py

@@ -0,0 +1,23 @@
+from typing import Any
+
+from fastcs.attributes import Attribute, AttrW, Sender
+
+from fastcs_pandablocks.types import PandaName
+
+
+class DefaultFieldSender(Sender):
+    def __init__(self, panda_name: PandaName):
+        self.panda_name = panda_name
+
+    async def put(self, controller: Any, attr: AttrW, value: str) -> None:
+        await controller.put_value_to_panda(self.panda_name, value)
+
+
+class UpdateEguSender(Sender):
+    def __init__(self, attr_to_update: Attribute):
+        """Update the attr"""
+        self.attr_to_update = attr_to_update
+
+    async def put(self, controller: Any, attr: AttrW, value: str) -> None:
+        # TODO find out how to update attr_to_update's EGU with the value
+        ...

src/fastcs_pandablocks/panda/blocks.py

@@ -0,0 +1,116 @@
+from collections.abc import Generator
+
+from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
+from fastcs.controller import SubController
+from pandablocks.responses import BlockInfo
+
+from fastcs_pandablocks.types import EpicsName, PandaName, ResponseType
+
+from .fields import FieldControllerType, get_field_controller_from_field_info
+
+
+class BlockController(SubController):
+    fields: dict[str, FieldControllerType]
+
+    def __init__(
+        self,
+        panda_name: PandaName,
+        number: int | None,
+        description: str | None | None,
+        raw_fields: dict[str, ResponseType],
+    ):
+        self._additional_attributes: dict[str, Attribute] = {}
+        self.panda_name = panda_name
+        self.number = number
+        self.description = description
+        self.fields = {}
+
+        for field_raw_name, field_info in raw_fields.items():
+            field_panda_name = PandaName(field=field_raw_name)
+            field = get_field_controller_from_field_info(field_info)
+            self.fields[field_panda_name.attribute_name] = field
+
+        super().__init__()
+
+    def initialise(self):
+        for field_name, field in self.fields.items():
+            if field.additional_attributes:
+                self.register_sub_controller(field_name, sub_controller=field)
+            if field.top_level_attribute:
+                self._additional_attributes[field_name] = field.top_level_attribute
+
+            field.initialise()
+
+    @property
+    def additional_attributes(self) -> dict[str, Attribute]:
+        return self._additional_attributes
+
+
+class Blocks:
+    _blocks: dict[str, dict[int | None, BlockController]]
+    epics_prefix: EpicsName
+
+    def __init__(self):
+        self._blocks = {}
+
+    def parse_introspected_data(
+        self, blocks: dict[str, BlockInfo], fields: list[dict[str, ResponseType]]
+    ):
+        self._blocks = {}
+
+        for (block_name, block_info), raw_fields in zip(
+            blocks.items(), fields, strict=True
+        ):
+            iterator = (
+                range(1, block_info.number + 1)
+                if block_info.number > 1
+                else iter(
+                    [
+                        None,
+                    ]
+                )
+            )
+            self._blocks[block_name] = {
+                number: BlockController(
+                    PandaName(block=block_name, block_number=number),
+                    block_info.number,
+                    block_info.description,
+                    raw_fields,
+                )
+                for number in iterator
+            }
+
+    async def update_field_value(self, panda_name: PandaName, value: str):
+        attribute = self[panda_name]
+
+        if isinstance(attribute, AttrW):
+            await attribute.process(value)
+        elif isinstance(attribute, (AttrRW | AttrR)):
+            await attribute.set(value)
+        else:
+            raise RuntimeError(f"Couldn't find panda field for {panda_name}.")
+
+    def flattened_attribute_tree(
+        self,
+    ) -> Generator[tuple[str, BlockController], None, None]:
+        for blocks in self._blocks.values():
+            for block in blocks.values():
+                yield (block.panda_name.attribute_name, block)
+
+    def __getitem__(
+        self, name: PandaName
+    ) -> dict[int | None, BlockController] | BlockController | Attribute:
+        if name.block is None:
+            raise ValueError(f"Cannot find block for name {name}.")
+        blocks = self._blocks[name.block]
+        if name.block_number is None:
+            return blocks
+        block = blocks[name.block_number]
+        if name.field is None:
+            return block
+        field = block.fields[name.field]
+        if not name.sub_field:
+            assert field.top_level_attribute
+            return field.top_level_attribute
+
+        return field.additional_attributes[name.sub_field]