Separated PV Status Generation from Builder.py

example/t01-services/synoptic/techui.yaml

@@ -9,6 +9,8 @@ components:
   fshtr:
     label: Fast Shutter
     prefix: BL01T-EA-FSHTR-01
+    status:
+      - BL01T-EA-FSHTR-01:FSHTR1.STAT
 
   d1:
     label: Diode 1

src/techui_builder/__main__.py

@@ -11,6 +11,7 @@
 from techui_builder.autofill import Autofiller
 from techui_builder.builder import Builder
 from techui_builder.schema_generator import schema_generator
+from techui_builder.status import status_run
 
 logger_ = logging.getLogger(__name__)
 
@@ -115,7 +116,9 @@ def main(
     filename: Annotated[Path, typer.Argument(help="The path to techui.yaml")],
     bobfile: Annotated[
         Path | None,
-        typer.Argument(help="Override for template bob file location."),
+        typer.Option(
+            "--bob-file", "-bb", help="Override for template bob file location."
+        ),
     ] = None,
     version: Annotated[
         bool | None, typer.Option("--version", callback=version_callback)
@@ -138,6 +141,13 @@ def main(
             callback=schema_callback,
         ),
     ] = None,
+    status: Annotated[
+        bool | None,
+        typer.Option(
+            "--status",
+            help="Generate status PVs for components with a status field",
+        ),
+    ] = None,
 ) -> None:
     """Default function called from cmd line tool."""
 
@@ -162,7 +172,9 @@ def main(
 
     gui.setup()
     gui.create_screens()
-    gui.write_status_pvs()
+    if status:
+        status_run(filename)  # Generate status PVs if required
+        logger_.info(f"Status PVs generated for {gui.conf.beamline.location}.")
 
     logger_.info(f"Screens generated for {gui.conf.beamline.location}.")
 

src/techui_builder/builder.py

@@ -7,10 +7,8 @@
 from typing import Any
 
 import yaml
-from epicsdbbuilder.recordbase import Record
 from lxml import etree, objectify
 from lxml.objectify import ObjectifiedElement
-from softioc.builder import records
 
 from techui_builder.generate import Generator
 from techui_builder.models import Entity, TechUi
@@ -48,7 +46,6 @@ class Builder:
     entities: defaultdict[str, list[Entity]] = field(
         default_factory=lambda: defaultdict(list), init=False
     )
-    status_pvs: dict[str, Record] = field(default_factory=dict, init=False)
     _services_dir: Path = field(init=False, repr=False)
     _gui_map: dict = field(init=False, repr=False)
     _write_directory: Path = field(default=Path("opis"), init=False, repr=False)
@@ -100,54 +97,6 @@ def clean_files(self):
             logger_.debug(f"Removing generated file: {file_.name}")
             os.remove(file_)
 
-    def _create_status_pv(self, prefix: str, inputs: list[str]):
-        # Extract all input PVs, provided a default "" if not provided
-        values = [(inputs[i] if i < len(inputs) else "") for i in range(12)]
-        inpa, inpb, inpc, inpd, inpe, inpf, inpg, inph, inpi, inpj, inpk, inpl = values
-
-        status_pv = records.calc(  # pyright: ignore[reportAttributeAccessIssue]
-            f"{prefix}:STA",
-            CALC="(A|B|C|D|E|F|G|H|I|J|K|L)>0?1:0",
-            SCAN="1 second",
-            ACKT="NO",
-            INPA=inpa,
-            INPB=inpb,
-            INPC=inpc,
-            INPD=inpd,
-            INPE=inpe,
-            INPF=inpf,
-            INPG=inpg,
-            INPH=inph,
-            INPI=inpi,
-            INPJ=inpj,
-            INPK=inpk,
-            INPL=inpl,
-        )
-
-        self.status_pvs[prefix] = status_pv
-
-    def write_status_pvs(self):
-        conf_dir = self._write_directory.joinpath("config")
-
-        # Create the config/ dir if it doesn't exist
-        if not conf_dir.exists():
-            os.mkdir(conf_dir)
-
-        with open(conf_dir.joinpath("status.db"), "w") as f:
-            # Add a header explaining the file is autogenerated
-            f.write("#" * 51 + "\n")
-            f.write(
-                "#" * 2
-                + " THIS FILE HAS BEEN AUTOGENERATED; DO NOT EDIT "
-                + "#" * 2
-                + "\n"
-            )
-            f.write("#" * 51 + "\n")
-
-            # Write the status PVs
-            for dpv in self.status_pvs.values():
-                dpv.Print(f)
-
     def _extract_services(self):
         """
         Finds the services folders in the services directory
@@ -215,9 +164,6 @@ def create_screens(self):
         for component_name, component in self.conf.components.items():
             screen_entities: list[Entity] = []
 
-            if component.status is not None:
-                self._create_status_pv(component.prefix, component.status)
-
             # ONLY IF there is a matching component and entity, generate a screen
             if component.prefix in self.entities.keys():
                 # Populate child labels for any entities

src/techui_builder/status.py

@@ -0,0 +1,97 @@
+import logging
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Annotated
+
+import typer
+import yaml
+from epicsdbbuilder.recordbase import Record
+from softioc.builder import records
+from typer.cli import app
+
+from techui_builder.models import TechUi
+
+logger_ = logging.getLogger(__name__)
+
+
+@dataclass
+class GenerateStatusPvs:
+    techui_path: Path = field(repr=False)
+    status_pvs: dict[str, Record] = field(default_factory=dict, init=False)
+
+    def __post_init__(self):
+        self._write_directory = self.techui_path.parent
+
+        try:
+            self.techui_yaml: TechUi = TechUi.model_validate(
+                yaml.safe_load(self.techui_path.read_text(encoding="utf-8"))
+            )
+        except Exception as e:
+            logger_.error(f"Error loading techui.yaml: {e}")
+
+            raise
+
+    def create_status_pv(self, prefix: str, inputs: list[str]):
+        # Extract all input PVs, provided a default "" if not provided
+        values = [(inputs[i] if i < len(inputs) else "") for i in range(12)]
+        inpa, inpb, inpc, inpd, inpe, inpf, inpg, inph, inpi, inpj, inpk, inpl = values
+        status_pv = records.calc(  # pyright: ignore[reportAttributeAccessIssue]
+            f"{prefix}:STA",
+            CALC="(A|B|C|D|E|F|G|H|I|J|K|L)>0?1:0",
+            SCAN="1 second",
+            ACKT="NO",
+            INPA=inpa,
+            INPB=inpb,
+            INPC=inpc,
+            INPD=inpd,
+            INPE=inpe,
+            INPF=inpf,
+            INPG=inpg,
+            INPH=inph,
+            INPI=inpi,
+            INPJ=inpj,
+            INPK=inpk,
+            INPL=inpl,
+        )
+
+        self.status_pvs[prefix] = status_pv
+
+    def write_status_pvs(self):
+        conf_dir = self._write_directory.joinpath("config")
+
+        # Create the config/ dir if it doesn't exist
+        if not conf_dir.exists():
+            os.mkdir(conf_dir)
+
+        with open(conf_dir.joinpath("status.db"), "w") as f:
+            # Add a header explaining the file is autogenerated
+            f.write("#" * 51 + "\n")
+            f.write(
+                "#" * 2
+                + " THIS FILE HAS BEEN AUTOGENERATED; DO NOT EDIT "
+                + "#" * 2
+                + "\n"
+            )
+            f.write("#" * 51 + "\n")
+
+            # Write the status PVs
+            for dpv in self.status_pvs.values():
+                dpv.Print(f)
+
+
+@app.callback(invoke_without_command=True)
+def status_run(
+    techui: Annotated[Path, typer.Argument(help="The path to techui.yaml")],
+):
+    status_gen = GenerateStatusPvs(techui)
+    for component in status_gen.techui_yaml.components.values():
+        if component.status is not None:
+            # if a status field is provided, generate a status PV for the component
+            status_gen.create_status_pv(component.prefix, component.status)
+    # write the generated PVs to a file
+    status_gen.write_status_pvs()
+
+
+if __name__ == "__main__":
+    app()

tests/conftest.py

@@ -9,6 +9,7 @@
 from techui_builder.builder import Builder, JsonMap
 from techui_builder.generate import Generator
 from techui_builder.models import Component
+from techui_builder.status import GenerateStatusPvs
 from techui_builder.validator import Validator
 
 
@@ -44,6 +45,11 @@ def components(builder_with_test_files: Builder):
     return builder_with_test_files.conf.components
 
 
+@pytest.fixture
+def status_gen():
+    return GenerateStatusPvs(Path("tests/t01-services/synoptic/techui.yaml").absolute())
+
+
 @pytest.fixture
 def test_files():
     screen_path = Path("tests/test_files/test_bob.bob").absolute()

tests/test_builder.py

@@ -1,13 +1,11 @@
 import logging
 import os
-from io import StringIO
 from pathlib import Path
-from unittest.mock import MagicMock, Mock, mock_open, patch
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 from lxml import objectify
 from phoebusgen.widget import ActionButton, Group
-from softioc.builder import ClearRecords, records
 
 from techui_builder.builder import (
     JsonMap,
@@ -73,78 +71,6 @@ def test_component_attributes(
         assert component.extras == extras
 
 
-def test_builder_create_status_pv(builder):
-    p = "BL01T-MO-MOTOR-01"
-    inpa = "BL01T-MO-MOTOR-01:MOTOR1.MOVN"
-    builder._create_status_pv(prefix=p, inputs=[inpa])
-
-    status_pv = """
-record(calc, "BL01T-MO-MOTOR-01:STA")
-{
-    field(ACKT, "NO")
-    field(CALC, "(A|B|C|D|E|F|G|H|I|J|K|L)>0?1:0")
-    field(INPA, "BL01T-MO-MOTOR-01:MOTOR1.MOVN")
-    field(INPB, "")
-    field(INPC, "")
-    field(INPD, "")
-    field(INPE, "")
-    field(INPF, "")
-    field(INPG, "")
-    field(INPH, "")
-    field(INPI, "")
-    field(INPJ, "")
-    field(INPK, "")
-    field(INPL, "")
-    field(SCAN, "1 second")
-}
-"""
-
-    assert builder.status_pvs != {}
-
-    # Fake file-like object to "print" the record to
-    auto_status_pv = StringIO()
-    # Get the string representation of the record
-    builder.status_pvs[p].Print(auto_status_pv)
-
-    assert auto_status_pv.getvalue() == status_pv
-
-    # Make sure the record is deleted
-    ClearRecords()
-
-
-def test_builder_write_status_pvs(builder):
-    # To mock the open() function used in _write_status_pvs
-    m = mock_open()
-
-    p = "BL01T-MO-MOTOR-01"
-    inpa = "BL01T-MO-MOTOR-01:MOTOR1.MOVN"
-    status_pv = records.calc(  # pyright: ignore[reportAttributeAccessIssue]
-        f"{p}:STA",
-        CALC="(A|B|C|D|E|F|G|H|I|J|K|L)>0?1:0",
-        SCAN="1 second",
-        ACKT="NO",
-        INPA=inpa,
-    )
-    builder.status_pvs[p] = status_pv
-
-    # Mock the Print() function so we don't actually write a file
-    with (
-        patch("builtins.open", m),
-        patch("techui_builder.builder.Record.Print") as mock_print,
-    ):
-        builder.write_status_pvs()
-
-        # Check open() was called with the correct args
-        m.assert_called_once_with(
-            Path(builder._write_directory.joinpath("config/status.db")),
-            "w",
-        )
-        mock_print.assert_called_once()
-
-    # Make sure the record is deleted
-    ClearRecords()
-
-
 def test_missing_service(builder, caplog):
     builder._extract_entities = Mock(side_effect=OSError())
     builder._extract_services()
@@ -213,8 +139,6 @@ def test_builder_validate_screen(builder_with_setup):
 
 
 def test_create_screens(builder_with_setup):
-    # We don't want to make a status PV in this test
-    builder_with_setup._create_status_pv = Mock()
     # We don't want to access Generator in this test
     builder_with_setup._generate_screen = Mock()
     builder_with_setup._validate_screen = Mock()
@@ -226,9 +150,6 @@ def test_create_screens(builder_with_setup):
 
 
 def test_create_screens_no_entities(builder, caplog):
-    # We don't want to make a status PV in this test
-    builder._create_status_pv = Mock()
-
     builder.entities = []
 
     # We only wan't to capture CRITICAL output in this test
@@ -244,8 +165,6 @@ def test_create_screens_no_entities(builder, caplog):
 
 
 def test_create_screens_extra_p_does_not_exist(builder_with_setup, caplog):
-    # We don't want to make a status PV in this test
-    builder_with_setup._create_status_pv = Mock()
     # We don't want to actually generate a screen
     builder_with_setup._generate_screen = Mock(side_effect=None)