From e7afb50c59b9c91257310ab95f18885e6bfc3b47 Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Tue, 26 May 2026 10:52:14 +0000 Subject: [PATCH 1/6] Added status.py --- src/techui_builder/status.py | 63 ++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 src/techui_builder/status.py diff --git a/src/techui_builder/status.py b/src/techui_builder/status.py new file mode 100644 index 0000000..924294f --- /dev/null +++ b/src/techui_builder/status.py @@ -0,0 +1,63 @@ +import os +from dataclasses import dataclass, field +from pathlib import Path + +from epicsdbbuilder.recordbase import Record +from softioc.builder import records + + +@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.status_pvs = {} + + 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) From 0eac1e7f643b7b668c8becb1921ab4ca71a2d8ed Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Tue, 26 May 2026 10:58:09 +0000 Subject: [PATCH 2/6] create write directory --- src/techui_builder/status.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/techui_builder/status.py b/src/techui_builder/status.py index 924294f..27c1b8a 100644 --- a/src/techui_builder/status.py +++ b/src/techui_builder/status.py @@ -13,6 +13,7 @@ class GenerateStatusPvs: def __post_init__(self): self.status_pvs = {} + self._write_directory = self.techui_path.parent def _create_status_pv(self, prefix: str, inputs: list[str]): # Extract all input PVs, provided a default "" if not provided From 8aab371b767c5fdcafa8ed41c7ce05f7acedee7d Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Tue, 26 May 2026 13:55:46 +0000 Subject: [PATCH 3/6] Separated status pv generation from builder.py --- src/techui_builder/__main__.py | 16 ++++++- src/techui_builder/builder.py | 54 ---------------------- src/techui_builder/status.py | 39 ++++++++++++++-- tests/conftest.py | 6 +++ tests/test_builder.py | 83 +--------------------------------- tests/test_cli.py | 15 +++++- tests/test_status.py | 77 +++++++++++++++++++++++++++++++ 7 files changed, 148 insertions(+), 142 deletions(-) create mode 100644 tests/test_status.py diff --git a/src/techui_builder/__main__.py b/src/techui_builder/__main__.py index 08b6f04..e7c37f9 100644 --- a/src/techui_builder/__main__.py +++ b/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}.") diff --git a/src/techui_builder/builder.py b/src/techui_builder/builder.py index 7b849c6..801ce95 100644 --- a/src/techui_builder/builder.py +++ b/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 diff --git a/src/techui_builder/status.py b/src/techui_builder/status.py index 27c1b8a..9abe66d 100644 --- a/src/techui_builder/status.py +++ b/src/techui_builder/status.py @@ -1,9 +1,18 @@ +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 @@ -12,14 +21,21 @@ class GenerateStatusPvs: status_pvs: dict[str, Record] = field(default_factory=dict, init=False) def __post_init__(self): - self.status_pvs = {} self._write_directory = self.techui_path.parent - def _create_status_pv(self, prefix: str, inputs: list[str]): + 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", @@ -62,3 +78,20 @@ def write_status_pvs(self): # 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() diff --git a/tests/conftest.py b/tests/conftest.py index b0125ef..6103fd0 100644 --- a/tests/conftest.py +++ b/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() diff --git a/tests/test_builder.py b/tests/test_builder.py index aee743e..782fbdc 100644 --- a/tests/test_builder.py +++ b/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 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) diff --git a/tests/test_cli.py b/tests/test_cli.py index 1986be7..a123860 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -36,6 +36,15 @@ def test_app_version(): # assert result.exit_code == 0 +@patch("techui_builder.__main__.status_run") +def test_status_run(mock_status_run, caplog): + mock_status_run.return_value = Mock() + result = runner.invoke( + app, ["--status", "example/t01-services/synoptic/techui.yaml"] + ) + assert result.exit_code == 0 + + @patch("techui_builder.__main__.schema_generator") def test_schema_callback(mock_schema_generator): with pytest.raises(typer.Exit): @@ -133,7 +142,11 @@ def test_find_bob_no_bob_file_found(caplog): @patch("techui_builder.__main__.find_dirs") @patch("techui_builder.__main__.Autofiller") @patch("techui_builder.__main__.Builder") -def test_main(mock_builder, mock_autofiller, mock_find_dirs, mock_find_bob): +@patch("techui_builder.__main__.status_run") +def test_main( + mock_status, mock_builder, mock_autofiller, mock_find_dirs, mock_find_bob +): + mock_status.return_value = Mock() mock_find_dirs.return_value = Mock(), Mock() mock_path = Mock(spec=Path) main(mock_path) diff --git a/tests/test_status.py b/tests/test_status.py new file mode 100644 index 0000000..ad31a26 --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,77 @@ +from io import StringIO +from pathlib import Path +from unittest.mock import mock_open, patch + +from softioc.builder import ClearRecords, records + + +def test_builder_create_status_pv(status_gen): + p = "BL01T-MO-MOTOR-01" + inpa = "BL01T-MO-MOTOR-01:MOTOR1.MOVN" + status_gen.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 status_gen.status_pvs != {} + + # Fake file-like object to "print" the record to + auto_status_pv = StringIO() + # Get the string representation of the record + status_gen.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(status_gen): + # 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, + ) + status_gen.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, + ): + status_gen.write_status_pvs() + + # Check open() was called with the correct args + m.assert_called_once_with( + Path(status_gen._write_directory.joinpath("config/status.db")), + "w", + ) + mock_print.assert_called_once() + + # Make sure the record is deleted + ClearRecords() From 7c6279b10c56fe810c9e64a5cf2aae54fc75389f Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Tue, 26 May 2026 13:58:13 +0000 Subject: [PATCH 4/6] Correcting typos in tests --- tests/test_status.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_status.py b/tests/test_status.py index ad31a26..480cf9f 100644 --- a/tests/test_status.py +++ b/tests/test_status.py @@ -5,7 +5,7 @@ from softioc.builder import ClearRecords, records -def test_builder_create_status_pv(status_gen): +def test_status_create_status_pv(status_gen): p = "BL01T-MO-MOTOR-01" inpa = "BL01T-MO-MOTOR-01:MOTOR1.MOVN" status_gen.create_status_pv(prefix=p, inputs=[inpa]) @@ -44,7 +44,7 @@ def test_builder_create_status_pv(status_gen): ClearRecords() -def test_builder_write_status_pvs(status_gen): +def test_status_write_status_pvs(status_gen): # To mock the open() function used in _write_status_pvs m = mock_open() @@ -62,7 +62,7 @@ def test_builder_write_status_pvs(status_gen): # 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, + patch("techui_builder.status.Record.Print") as mock_print, ): status_gen.write_status_pvs() From 978207b8fda3916fcbe27f2a318ed78ccf7934ff Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Thu, 28 May 2026 08:03:59 +0000 Subject: [PATCH 5/6] added missing function for test --- tests/test_builder.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_builder.py b/tests/test_builder.py index 782fbdc..93b6297 100644 --- a/tests/test_builder.py +++ b/tests/test_builder.py @@ -1,7 +1,7 @@ import logging import os from pathlib import Path -from unittest.mock import Mock, patch +from unittest.mock import MagicMock, Mock, patch import pytest from lxml import objectify From 2f5976aca343b5f93d6d9e71d5a1405da0b20f83 Mon Sep 17 00:00:00 2001 From: "Sode, Adedamola (DLSLtd,RAL,LSCI)" Date: Thu, 28 May 2026 08:44:44 +0000 Subject: [PATCH 6/6] Added test for status_run and reading techui.yaml --- example/t01-services/synoptic/techui.yaml | 2 ++ tests/test_status.py | 19 ++++++++++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/example/t01-services/synoptic/techui.yaml b/example/t01-services/synoptic/techui.yaml index 341bfaf..74daf57 100644 --- a/example/t01-services/synoptic/techui.yaml +++ b/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 diff --git a/tests/test_status.py b/tests/test_status.py index 480cf9f..42987df 100644 --- a/tests/test_status.py +++ b/tests/test_status.py @@ -1,9 +1,26 @@ from io import StringIO from pathlib import Path -from unittest.mock import mock_open, patch +from unittest.mock import Mock, mock_open, patch +import pytest from softioc.builder import ClearRecords, records +from techui_builder.status import status_run + + +@patch("techui_builder.status.GenerateStatusPvs.write_status_pvs") +@patch("techui_builder.status.GenerateStatusPvs.create_status_pv") +def test_status_run(mock_create, mock_write): + mock_create.return_value = Mock() + mock_write.return_value = Mock() + status_run(Path("tests/t01-services/synoptic/techui.yaml").absolute()) + + +def test_status_run_invalid_yaml(caplog): + with pytest.raises(Exception): # noqa: B017 + status_run(Path("tests/invalid_techui.yaml").absolute()) + assert "Error loading techui.yaml" in caplog.text + def test_status_create_status_pv(status_gen): p = "BL01T-MO-MOTOR-01"