diff --git a/example/t01-services/synoptic/techui.yaml b/example/t01-services/synoptic/techui.yaml index 341bfaf0..74daf576 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/src/techui_builder/__main__.py b/src/techui_builder/__main__.py index 08b6f046..e7c37f91 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 7b849c65..801ce95b 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 new file mode 100644 index 00000000..9abe66df --- /dev/null +++ b/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() diff --git a/tests/conftest.py b/tests/conftest.py index b0125ef2..6103fd07 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 aee743ed..93b62979 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 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) diff --git a/tests/test_cli.py b/tests/test_cli.py index 1986be71..a1238607 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 00000000..42987dfa --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,94 @@ +from io import StringIO +from pathlib import Path +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" + 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_status_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.status.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()