From 151a2cb811d002ab22091eedca8ef81d71fba7e4 Mon Sep 17 00:00:00 2001 From: Matt King Date: Wed, 8 Nov 2023 18:38:59 +0000 Subject: [PATCH] Convert magnet to use pydantic model (#104) * magnets and screen conversion from csv to dictionary structures working. Can extract by area * add functionality for extracting multiple areas at once * add PVs to yaml magnets extraction * removing unneeded file * working on adding image pvs to yaml files * working on image device code * flake8 and black formatting * make construction function self, add error handling for MEME timeout * formatting * make area a list in screens * add pv_info to screen construction * restrict wildcard search in meme * added models for magnet and associated fields * start to move magnet functions over to pydantic model * magnet pydantic model working with reader now * updated yaml test files to reflect new yaml for devices * magnet and reader tests up to date with pydantic changes, minimal functional changes to tests * make MagnetPVSet have frozen attributes.# * make MagnetPVSet have frozen attributes. * frozen for controls_information base class * missed return for area property! * add setter for bdes * working on being able to set bdes values for several mags * find file by area in reader * use area over location for scope * change yaml path. * remove path * first attempt at function to scan mags * magnet and device now using pydantic model * tests reflect model changes, need to check coverage * add check for function in scan * remove custom exception, pydantic handles mandatory information now * coverage up to 100% for each new class now * formatting * add pydantic to requirements * remove superceeded yaml * add comments for units in tolerance/length * add comment explaining adding name for device here --------- Co-authored-by: Neveu Co-authored-by: Neveu Co-authored-by: matt --- lcls_tools/common/devices/device.py | 72 +-- lcls_tools/common/devices/magnet/magnet.py | 220 +++++---- lcls_tools/common/devices/magnet/reader.py | 44 +- lcls_tools/common/devices/yaml/generate.py | 6 +- .../device/config/magnet/gun.yaml | 52 --- requirements.txt | 1 + .../datasets/devices/config/base_device.yaml | 25 +- .../devices/config/magnet/bad_magnet.yaml | 72 +-- .../devices/config/magnet/typical_magnet.yaml | 424 ++++++++++++++++-- .../common/devices/magnet/test_magnet.py | 131 ++++-- .../common/devices/magnet/test_reader.py | 47 +- .../lcls_tools/common/devices/test_device.py | 31 +- 12 files changed, 792 insertions(+), 333 deletions(-) delete mode 100644 lcls_tools/normalconducting/device/config/magnet/gun.yaml diff --git a/lcls_tools/common/devices/device.py b/lcls_tools/common/devices/device.py index 06065e1e..a29d6a78 100644 --- a/lcls_tools/common/devices/device.py +++ b/lcls_tools/common/devices/device.py @@ -1,10 +1,35 @@ -from typing import Callable, Union +from pydantic import BaseModel, SerializeAsAny, ConfigDict +from typing import List, Union, Callable, Optional from epics import PV -class MandatoryFieldNotFoundInYAMLError(Exception): - def __init__(self, message: str): - super().__init__(message) +class PVSet(BaseModel): + model_config = ConfigDict( + arbitrary_types_allowed=True, + extra="forbid", + frozen=True, + ) + ... + + +class ControlInformation(BaseModel): + model_config = ConfigDict( + frozen=True, + ) + control_name: str + PVs: PVSet + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + +class Metadata(BaseModel): + area: str + beam_path: List[str] + sum_l_meters: float + + def __init__(self, **kwargs): + super().__init__(**kwargs) class ApplyDeviceCallbackError(Exception): @@ -17,30 +42,25 @@ def __init__(self, message: str) -> None: super().__init__(message) -class Device: - _name: str - control_information: dict = None - metadata: dict = None - _mandatory_fields: list = [ - "control_information", - "metadata", - ] - - def __init__(self, name: str = None, config: dict = None): - self._name = name - try: - # extract and create metadata and control_information attributes from **kwargs. - for field in self._mandatory_fields: - setattr(self, field, config[field]) - except KeyError as ke: - missing_field = ke.args[0] - raise MandatoryFieldNotFoundInYAMLError( - f"Missing {missing_field} for device {self.name}, please check yaml file." - ) +class Device(BaseModel): + name: Optional[str] = None + controls_information: SerializeAsAny[ControlInformation] + metadata: SerializeAsAny[Metadata] + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @property + def area(self): + return self.metadata.area + + @property + def sum_l_meters(self): + return self.metadata.sum_l_meters @property - def name(self): - return self._name + def beam_path(self): + return self.metadata.beam_path def get_callbacks(self, pv: str) -> Union[None, dict]: pv_obj = self._get_pv_object_from_str(pv) diff --git a/lcls_tools/common/devices/magnet/magnet.py b/lcls_tools/common/devices/magnet/magnet.py index d516d75c..e9c5b675 100644 --- a/lcls_tools/common/devices/magnet/magnet.py +++ b/lcls_tools/common/devices/magnet/magnet.py @@ -1,47 +1,73 @@ -#!/usr/local/lcls/package/python/current/bin/python +from pydantic import ( + BaseModel, + PositiveFloat, + SerializeAsAny, + field_validator, +) +from typing import ( + Dict, + List, + Optional, + Union, +) +from lcls_tools.common.devices.device import ( + Device, + ControlInformation, + Metadata, + PVSet, +) from epics import PV -from typing import Union -from lcls_tools.common.devices.device import Device + + +class MagnetPVSet(PVSet): + bctrl: PV + bact: PV + bdes: PV + bcon: PV + ctrl: PV + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @field_validator("*", mode="before") + def validate_pv_fields(cls, v: str): + return PV(v) + + +class MagnetControlInformation(ControlInformation): + PVs: SerializeAsAny[MagnetPVSet] + ctrl_options: SerializeAsAny[Optional[Dict[str, int]]] = { + "READY": 0, + "TRIM": 1, + "PERTURB": 2, + "BCON_TO_BDES": 3, + "SAVE_BDES": 4, + "LOAD_BDES": 5, + "UNDO_BDES": 6, + "DAC_ZERO": 7, + "CALIB": 8, + "STDZ": 9, + "RESET": 10, + } + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + +class MagnetMetadata(Metadata): + length: Optional[PositiveFloat] = None + b_tolerance: Optional[PositiveFloat] = None + + def __init__(self, **kwargs): + super().__init__(**kwargs) class Magnet(Device): - _bctrl: PV - _bact: PV - _bdes: PV - _bcon: PV - _ctrl: PV - _length: float = None - _b_tolerance: float = None - _ctrl_options: dict = None - _mandatory_pvs: list = [ - "bctrl", - "bact", - "bdes", - "bcon", - "ctrl", - ] - - def __init__(self, name=None, **kwargs): - super().__init__(name=name, config=kwargs) - self._make_pv_attributes() - self._set_control_options() - self._make_metadata_attributes() - - def _set_control_options(self): - self._ctrl_options = self.control_information["ctrl_options"] - - def _make_pv_attributes(self): - # setup PVs and private attributes - for alias in self._mandatory_pvs: - if alias in self.control_information["PVs"]: - pv = self.control_information["PVs"][alias] - setattr(self, "_" + alias, PV(pv)) - else: - raise AttributeError(f"PV for {alias} not defined in .yaml") - - def _make_metadata_attributes(self): - # setup metadata and private attributes - [setattr(self, "_" + k, v) for k, v in self.metadata.items()] + controls_information: SerializeAsAny[MagnetControlInformation] + metadata: SerializeAsAny[MagnetMetadata] + + def __init__(self, **kwargs): + super().__init__(**kwargs) """ Decorators """ @@ -56,9 +82,35 @@ def decorated(self, *args, **kwargs): return decorated + @property + def ctrl_options(self): + return self.controls_information.ctrl_options + + @property + def b_tolerance(self): + """Returns the field tolerance in kG or kGm""" + return self.metadata.b_tolerance + + @b_tolerance.setter + def b_tolerance(self, value): + if not isinstance(value, float): + return + self.metadata.b_tolerance = value + + @property + def length(self): + """Returns the effective length in meters""" + return self.metadata.length + + @length.setter + def length(self, value): + if not isinstance(value, float): + return + self.metadata.length = value + @property def bctrl(self) -> Union[float, int]: - return self._bctrl.get() + return self.controls_information.PVs.bact.get() @bctrl.setter @check_state @@ -67,91 +119,99 @@ def bctrl(self, val: Union[float, int]) -> None: if not (isinstance(val, float) or isinstance(val, int)): print("you need to provide an int or float") return - self._bctrl.put(val) + self.controls_information.PVs.bctrl.put(value=val) @property def bact(self) -> float: """Get the BACT value""" - return self._bact.get() + return self.controls_information.PVs.bact.get() @property def bdes(self) -> float: """Get BDES value""" - return self._bdes.get() + return self.controls_information.PVs.bdes.get() + + @bdes.setter + def bdes(self, bval) -> None: + self.controls_information.PVs.bdes.put(value=bval) @property def ctrl(self) -> str: """Get the current action on magnet""" - return self._ctrl.get(as_string=True) + return self.controls_information.PVs.ctrl.get(as_string=True) @property - def length(self) -> float: - """Magnetic Length, should be from model""" - return self._length - - @length.setter - def length(self, length: float) -> None: - """Set the magnetic length for a magnet""" - if not isinstance(length, float): - print("You must provide a float for magnet length") - return - - self._length = length - - @property - def b_tolerance(self) -> float: - return self._b_tolerance - - @b_tolerance.setter - def b_tolerance(self, tol: float) -> None: - """Set the magnetic length for a magnet""" - if not isinstance(tol, float): - print("You must provide a float for magnet tol") - return False - - self._b_tolerance = tol + def bcon(self) -> float: + """Get the configuration strength of magnet""" + return self.controls_information.PVs.bcon.get() @check_state def trim(self) -> None: """Issue trim command""" - self._ctrl.put(self._ctrl_options["TRIM"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["TRIM"]) @check_state def perturb(self) -> None: """Issue perturb command""" - self._ctrl.put(self._ctrl_options["PERTURB"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["PERTURB"]) def con_to_des(self) -> None: """Issue con to des commands""" - self._ctrl.put(self._ctrl_options["BCON_TO_BDES"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["BCON_TO_BDES"]) def save_bdes(self) -> None: """Save BDES""" - self._ctrl.put(self._ctrl_options["SAVE_BDES"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["SAVE_BDES"]) def load_bdes(self) -> None: """Load BtolDES""" - self._ctrl.put(self._ctrl_options["LOAD_BDES"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["LOAD_BDES"]) def undo_bdes(self) -> None: """Save BDES""" - self._ctrl.put(self._ctrl_options["UNDO_BDES"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["UNDO_BDES"]) @check_state def dac_zero(self) -> None: """DAC zero magnet""" - self._ctrl.put(self._ctrl_options["DAC_ZERO"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["DAC_ZERO"]) @check_state def calibrate(self) -> None: """Calibrate magnet""" - self._ctrl.put(self._ctrl_options["CALIB"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["CALIB"]) @check_state def standardize(self) -> None: """Standardize magnet""" - self._ctrl.put(self._ctrl_options["STDZ"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["STDZ"]) def reset(self) -> None: """Reset magnet""" - self._ctrl.put(self._ctrl_options["RESET"]) + self.controls_information.PVs.ctrl.put(self.ctrl_options["RESET"]) + + +class MagnetCollection(BaseModel): + magnets: Dict[str, SerializeAsAny[Magnet]] + + def set_bdes(self, magnet_dict: Dict[str, float]): + if not magnet_dict: + return + + for magnet, bval in magnet_dict.items(): + try: + self.magnets[magnet].bdes = bval + self.magnets[magnet].trim() + # TODO: settle time, and check bact is equal to bdes + except KeyError: + print( + "You tried to set a magnet that does not exist.", + f"{magnet} was not set to {bval}.", + ) + + def scan( + self, scan_settings: List[Dict[str, float]], function: Optional[callable] = None + ): + for setting in scan_settings: + self.set_bdes(setting) + function() if function else None diff --git a/lcls_tools/common/devices/magnet/reader.py b/lcls_tools/common/devices/magnet/reader.py index 39cb98c5..2bfdb3a9 100644 --- a/lcls_tools/common/devices/magnet/reader.py +++ b/lcls_tools/common/devices/magnet/reader.py @@ -1,45 +1,49 @@ import os import yaml from typing import Union -from lcls_tools.common.devices.device import MandatoryFieldNotFoundInYAMLError -from lcls_tools.common.devices.magnet.magnet import Magnet +from pydantic import ValidationError +from lcls_tools.common.devices.magnet.magnet import Magnet, MagnetCollection +DEFAULT_YAML_LOCATION = "./lcls_tools/common/devices/yaml/" -def _find_yaml_file(yaml_filename: str) -> str: - if os.path.isfile(yaml_filename): - return os.path.abspath(yaml_filename) + +def _find_yaml_file(area: str) -> str: + filename = area + ".yaml" + path = os.path.join(DEFAULT_YAML_LOCATION, filename) + if os.path.isfile(path): + return os.path.abspath(path) else: raise FileNotFoundError( - f"No such file {yaml_filename}", + f"No such file {path}, please choose another area.", ) def create_magnet( - yaml_filename: str = None, name: str = None -) -> Union[None, dict, Magnet]: - if yaml_filename: + area: str = None, name: str = None +) -> Union[None, Magnet, MagnetCollection]: + if area: try: location = _find_yaml_file( - yaml_filename=yaml_filename, + area=area, ) with open(location, "r") as device_file: + config_data = yaml.safe_load(device_file) if name: - config_data = yaml.safe_load(device_file)[name] - return Magnet(name=name, **config_data) + magnet_data = config_data["magnets"][name] + # this data is not available from YAML directly in this form, so we add it here. + magnet_data.update({"name": name}) + return Magnet(**magnet_data) else: - return { - name: Magnet(name=name, **config_data) - for name, config_data in yaml.safe_load(device_file).items() - } + return MagnetCollection(**config_data) except FileNotFoundError: - print(f"Could not find yaml file: {yaml_filename}") + print(f"Could not find yaml file for area: {area}") return None except KeyError: - print(f"Could not find name {name} in {yaml_filename}") + print(f"Could not find name {name} in file for area: {area}") return None - except MandatoryFieldNotFoundInYAMLError as field_error: + except ValidationError as field_error: print(field_error) return None else: - print("Please provide a yaml file location to create a magnet.") + print("Please provide a machine area to create a magnet from.") return None diff --git a/lcls_tools/common/devices/yaml/generate.py b/lcls_tools/common/devices/yaml/generate.py index 004aa372..a331c0df 100644 --- a/lcls_tools/common/devices/yaml/generate.py +++ b/lcls_tools/common/devices/yaml/generate.py @@ -30,6 +30,7 @@ def __init__( def _is_required_field(pair: tuple): key, _ = pair return key in self._required_fields + # only store the required fields from lcls_elements, there are lots more! self.elements = [ dict(filter(_is_required_field, element.items())) @@ -178,11 +179,6 @@ def extract_magnets(self, area: Union[str, List[str]] = "GUNB") -> dict: pv_search_list=magnet_pv_search_list, ) - def camera_type(): - # Determine type of camera and PVs - # Is there a way to get the camera type w/o a list? - pass - def extract_screens(self, area: Union[str, List[str]] = ["HTR"]): required_screen_types = ["PROF"] possible_screen_pvs = [ diff --git a/lcls_tools/normalconducting/device/config/magnet/gun.yaml b/lcls_tools/normalconducting/device/config/magnet/gun.yaml deleted file mode 100644 index da0425cd..00000000 --- a/lcls_tools/normalconducting/device/config/magnet/gun.yaml +++ /dev/null @@ -1,52 +0,0 @@ -SOL2B: - control_information: - control_name: 'SOLN:GUNB:823' - PVs: - bctrl : 'SOLN:GUNB:823:BCTRL' - bact : 'SOLN:GUNB:823:BACT' - bdes : 'SOLN:GUNB:823:BDES' - bcon : 'SOLN:GUNB:823:BCON' - ctrl : 'SOLN:GUNB:823:CTRL' - ctrl_options: - READY : 0 - TRIM : 1 - PERTURB : 2 - BCON_TO_BDES : 3 - SAVE_BDES : 4 - LOAD_BDES : 5 - UNDO_BDES : 6 - DAC_ZERO : 7 - CALIB : 8 - STDZ : 9 - RESET : 10 - metadata: - b_tolerance : 0.002 # in Tesla? - length : 0.135 # in meters? - area: 'GUN' - - -SOL1B: - control_information: - control_name: 'SOLN:GUNB:212' - PVs: - bctrl : 'SOLN:GUNB:212:BCTRL' - bact : 'SOLN:GUNB:212:BACT' - bdes : 'SOLN:GUNB:212:BDES' - bcon : 'SOLN:GUNB:212:BCON' - ctrl : 'SOLN:GUNB:212:CTRL' - ctrl_options: - READY : 0 - TRIM : 1 - PERTURB : 2 - BCON_TO_BDES : 3 - SAVE_BDES : 4 - LOAD_BDES : 5 - UNDO_BDES : 6 - DAC_ZERO : 7 - CALIB : 8 - STDZ : 9 - RESET : 10 - metadata: - b_tolerance : 0.002 # in Tesla? - length : 0.1342 # in meters? - area: 'GUN' \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 7d4bce85..e9ae6103 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,3 +5,4 @@ matplotlib pyepics pyyaml requests +pydantic \ No newline at end of file diff --git a/tests/datasets/devices/config/base_device.yaml b/tests/datasets/devices/config/base_device.yaml index b7f4424b..5cbebd3d 100644 --- a/tests/datasets/devices/config/base_device.yaml +++ b/tests/datasets/devices/config/base_device.yaml @@ -1,23 +1,36 @@ DEVICE_1: - control_information: + controls_information: control_name: 'DEV:001' - PVs: - ctrl: 'DEV:001:CTRL' - set: 'DEV:001:SET' - read: 'DEV:001:RBV' + PVs: {} metadata: read_tolerance: 0.0002 safe_level: 0.15 area: L1B + beam_path: + - CU_ALINE + - CU_HTXI + - CU_HXR + - CU_HXTES + - CU_SFTH + - CU_SXR + sum_l_meters: 0.0 DEVICE_2: metadata: read_tolerance: 0.023 safe_level: 1000 area: L2B + beam_path: + - CU_ALINE + - CU_HTXI + - CU_HXR + - CU_HXTES + - CU_SFTH + - CU_SXR + sum_l_meters: 0.0 DEVICE_3: - control_information: + controls_information: control_name: 'DEV:003' PVs: ctrl: 'DEV:003:CTRL' diff --git a/tests/datasets/devices/config/magnet/bad_magnet.yaml b/tests/datasets/devices/config/magnet/bad_magnet.yaml index 3dc234fb..eee521f5 100644 --- a/tests/datasets/devices/config/magnet/bad_magnet.yaml +++ b/tests/datasets/devices/config/magnet/bad_magnet.yaml @@ -1,27 +1,49 @@ -SOL2B: - metadata: - b_tolerance : 0.002 # in Tesla? - length : 0.135 # in meters? +magnets: + CQ01B: + controls_information: + PVs: + bact: QUAD:GUNB:212:1:BACT + bcon: QUAD:GUNB:212:1:BCON + bctrl: QUAD:GUNB:212:1:BCTRL + bdes: QUAD:GUNB:212:1:BDES + ctrl: QUAD:GUNB:212:1:CTRL + control_name: QUAD:GUNB:212:1 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.247 + CQ02B: + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.57 + SOL1B: + controls_information: + PVs: + bact: SOLN:GUNB:212:BACT + bcon: SOLN:GUNB:212:BCON + bctrl: SOLN:GUNB:212:BCTRL + bdes: SOLN:GUNB:212:BDES + ctrl: SOLN:GUNB:212:CTRL + control_name: SOLN:GUNB:212 + -SOL1B: - control_information: - control_name: 'SOLN:GUNB:212' - PVs: - bctrl : 'QUAD:LI22:201:BCTRL' - bact : 'QUAD:LI22:201:BACT' - bdes : 'QUAD:LI22:201:BDES' - bcon : 'QUAD:LI22:201:BCON' - ctrl : 'QUAD:LI22:201:CTRL' - ctrl_options: - READY : 0 - TRIM : 1 - PERTURB : 2 - BCON_TO_BDES : 3 - SAVE_BDES : 4 - LOAD_BDES : 5 - UNDO_BDES : 6 - DAC_ZERO : 7 - CALIB : 8 - STDZ : 9 - RESET : 10 diff --git a/tests/datasets/devices/config/magnet/typical_magnet.yaml b/tests/datasets/devices/config/magnet/typical_magnet.yaml index a999f877..43caf3ac 100644 --- a/tests/datasets/devices/config/magnet/typical_magnet.yaml +++ b/tests/datasets/devices/config/magnet/typical_magnet.yaml @@ -1,50 +1,376 @@ -SOL2B: - control_information: - control_name: 'SOLN:GUNB:823' - PVs: - bctrl : 'SOLN:GUNB:823:BCTRL' - bact : 'SOLN:GUNB:823:BACT' - bdes : 'SOLN:GUNB:823:BDES' - bcon : 'SOLN:GUNB:823:BCON' - ctrl : 'SOLN:GUNB:823:CTRL' - ctrl_options: - READY : 0 - TRIM : 1 - PERTURB : 2 - BCON_TO_BDES : 3 - SAVE_BDES : 4 - LOAD_BDES : 5 - UNDO_BDES : 6 - DAC_ZERO : 7 - CALIB : 8 - STDZ : 9 - RESET : 10 - metadata: - b_tolerance : 0.002 # in Tesla? - length : 0.135 # in meters? +magnets: + CQ01B: + controls_information: + PVs: + bact: QUAD:GUNB:212:1:BACT + bcon: QUAD:GUNB:212:1:BCON + bctrl: QUAD:GUNB:212:1:BCTRL + bdes: QUAD:GUNB:212:1:BDES + ctrl: QUAD:GUNB:212:1:CTRL + control_name: QUAD:GUNB:212:1 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.247 + CQ02B: + controls_information: + PVs: + bact: QUAD:GUNB:823:1:BACT + bcon: QUAD:GUNB:823:1:BCON + bctrl: QUAD:GUNB:823:1:BCTRL + bdes: QUAD:GUNB:823:1:BDES + ctrl: QUAD:GUNB:823:1:CTRL + control_name: QUAD:GUNB:823:1 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.57 + SOL1B: + controls_information: + PVs: + bact: SOLN:GUNB:212:BACT + bcon: SOLN:GUNB:212:BCON + bctrl: SOLN:GUNB:212:BCTRL + bdes: SOLN:GUNB:212:BDES + ctrl: SOLN:GUNB:212:CTRL + control_name: SOLN:GUNB:212 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.247 + SOL1BKB: + controls_information: + PVs: + bact: SOLN:GUNB:100:BACT + bcon: SOLN:GUNB:100:BCON + bctrl: SOLN:GUNB:100:BCTRL + bdes: SOLN:GUNB:100:BDES + ctrl: SOLN:GUNB:100:CTRL + control_name: SOLN:GUNB:100 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: -0.072 + SOL2B: + controls_information: + PVs: + bact: SOLN:GUNB:823:BACT + bcon: SOLN:GUNB:823:BCON + bctrl: SOLN:GUNB:823:BCTRL + bdes: SOLN:GUNB:823:BDES + ctrl: SOLN:GUNB:823:CTRL + control_name: SOLN:GUNB:823 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.646 + SQ01B: + controls_information: + PVs: + bact: QUAD:GUNB:212:2:BACT + bcon: QUAD:GUNB:212:2:BCON + bctrl: QUAD:GUNB:212:2:BCTRL + bdes: QUAD:GUNB:212:2:BDES + ctrl: QUAD:GUNB:212:2:CTRL + control_name: QUAD:GUNB:212:2 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.247 + SQ02B: + controls_information: + PVs: + bact: QUAD:GUNB:823:2:BACT + bcon: QUAD:GUNB:823:2:BCON + bctrl: QUAD:GUNB:823:2:BCTRL + bdes: QUAD:GUNB:823:2:BDES + ctrl: QUAD:GUNB:823:2:CTRL + control_name: QUAD:GUNB:823:2 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.646 + XC01B: + controls_information: + PVs: + bact: XCOR:GUNB:293:BACT + bcon: XCOR:GUNB:293:BCON + bctrl: XCOR:GUNB:293:BCTRL + bdes: XCOR:GUNB:293:BDES + ctrl: XCOR:GUNB:293:CTRL + control_name: XCOR:GUNB:293 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.484 + XC02B: + controls_information: + PVs: + bact: XCOR:GUNB:388:BACT + bcon: XCOR:GUNB:388:BCON + bctrl: XCOR:GUNB:388:BCTRL + bdes: XCOR:GUNB:388:BDES + ctrl: XCOR:GUNB:388:CTRL + control_name: XCOR:GUNB:388 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.67 + XC03B: + controls_information: + PVs: + bact: XCOR:GUNB:513:BACT + bcon: XCOR:GUNB:513:BCON + bctrl: XCOR:GUNB:513:BCTRL + bdes: XCOR:GUNB:513:BDES + ctrl: XCOR:GUNB:513:CTRL + control_name: XCOR:GUNB:513 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.964 + XC04B: + controls_information: + PVs: + bact: XCOR:GUNB:713:BACT + bcon: XCOR:GUNB:713:BCON + bctrl: XCOR:GUNB:713:BCTRL + bdes: XCOR:GUNB:713:BDES + ctrl: XCOR:GUNB:713:CTRL + control_name: XCOR:GUNB:713 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.369 + XC05B: + controls_information: + PVs: + bact: XCOR:GUNB:927:BACT + bcon: XCOR:GUNB:927:BCON + bctrl: XCOR:GUNB:927:BCTRL + bdes: XCOR:GUNB:927:BDES + ctrl: XCOR:GUNB:927:CTRL + control_name: XCOR:GUNB:927 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.889 + YC01B: + controls_information: + PVs: + bact: YCOR:GUNB:293:BACT + bcon: YCOR:GUNB:293:BCON + bctrl: YCOR:GUNB:293:BCTRL + bdes: YCOR:GUNB:293:BDES + ctrl: YCOR:GUNB:293:CTRL + control_name: YCOR:GUNB:293 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.484 + YC02B: + controls_information: + PVs: + bact: YCOR:GUNB:388:BACT + bcon: YCOR:GUNB:388:BCON + bctrl: YCOR:GUNB:388:BCTRL + bdes: YCOR:GUNB:388:BDES + ctrl: YCOR:GUNB:388:CTRL + control_name: YCOR:GUNB:388 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.67 + YC03B: + controls_information: + PVs: + bact: YCOR:GUNB:513:BACT + bcon: YCOR:GUNB:513:BCON + bctrl: YCOR:GUNB:513:BCTRL + bdes: YCOR:GUNB:513:BDES + ctrl: YCOR:GUNB:513:CTRL + control_name: YCOR:GUNB:513 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 0.964 + YC04B: + controls_information: + PVs: + bact: YCOR:GUNB:713:BACT + bcon: YCOR:GUNB:713:BCON + bctrl: YCOR:GUNB:713:BCTRL + bdes: YCOR:GUNB:713:BDES + ctrl: YCOR:GUNB:713:CTRL + control_name: YCOR:GUNB:713 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.369 + YC05B: + controls_information: + PVs: + bact: YCOR:GUNB:927:BACT + bcon: YCOR:GUNB:927:BCON + bctrl: YCOR:GUNB:927:BCTRL + bdes: YCOR:GUNB:927:BDES + ctrl: YCOR:GUNB:927:CTRL + control_name: YCOR:GUNB:927 + metadata: + area: GUNB + beam_path: + - SC_BSYD + - SC_DASEL + - SC_DIAG0 + - SC_HXR + - SC_S2_X + - SC_SFTS + - SC_STMO + - SC_STXI + - SC_SXR + sum_l_meters: 1.889 - -SOL1B: - control_information: - control_name: 'SOLN:GUNB:212' - PVs: - bctrl : 'QUAD:LI22:201:BCTRL' - bact : 'QUAD:LI22:201:BACT' - bdes : 'QUAD:LI22:201:BDES' - bcon : 'QUAD:LI22:201:BCON' - ctrl : 'QUAD:LI22:201:CTRL' - ctrl_options: - READY : 0 - TRIM : 1 - PERTURB : 2 - BCON_TO_BDES : 3 - SAVE_BDES : 4 - LOAD_BDES : 5 - UNDO_BDES : 6 - DAC_ZERO : 7 - CALIB : 8 - STDZ : 9 - RESET : 10 - metadata: - b_tolerance : 0.002 # in Tesla? - length : 0.1342 # in meters? \ No newline at end of file diff --git a/tests/unit_tests/lcls_tools/common/devices/magnet/test_magnet.py b/tests/unit_tests/lcls_tools/common/devices/magnet/test_magnet.py index 16e42e66..6a675c02 100644 --- a/tests/unit_tests/lcls_tools/common/devices/magnet/test_magnet.py +++ b/tests/unit_tests/lcls_tools/common/devices/magnet/test_magnet.py @@ -7,6 +7,7 @@ # Local imports from lcls_tools.common.devices.magnet.reader import create_magnet +from lcls_tools.common.devices.magnet.magnet import MagnetCollection class MagnetTest(TestCase): @@ -16,29 +17,23 @@ class MagnetTest(TestCase): def setUp(self) -> None: self.magnet = create_magnet( - "./tests/datasets/devices/config/magnet/typical_magnet.yaml", + area="GUNB", + # "./tests/datasets/devices/config/magnet/typical_magnet.yaml", name="SOL1B", ) return super().setUp() - def test_missing_mandatory_pv_raises_attribute_error(self): - with self.assertRaises(AttributeError): - self.bad_magnet = create_magnet( - "./tests/datasets/devices/config/magnet/missing_pv_magnet.yaml", - name="SOL2B", - ) - def test_properties_exist(self): """Test that all the properties we expect exist""" # Assert that magnet has all auto-generated private attributes - for item in self.magnet._mandatory_pvs: + for handle, _ in self.magnet.controls_information.PVs: self.assertTrue( - hasattr(self.magnet, "_" + item), - msg=f"expected magnet to have attribute {item}", + hasattr(self.magnet, handle), + msg=f"expected magnet to have attribute {handle}", ) - for item in self.magnet.metadata: + for item, _ in self.magnet.metadata: self.assertTrue( - hasattr(self.magnet, "_" + item), + hasattr(self.magnet, item), msg=f"expected magnet to have attribute {item}", ) # Assert that magnet has public properties @@ -54,12 +49,6 @@ def test_properties_exist(self): msg=f"expected magnet to have attribute {item}", ) - for item in self.magnet._mandatory_fields: - self.assertTrue( - hasattr(self.magnet, item), - msg=f"expected magnet to have attribute {item}", - ) - def test_methods(self): """Test that all the methods we expect exist""" self.assertEqual(inspect.ismethod(self.magnet.trim), True) @@ -81,21 +70,21 @@ def test_name(self): def test_tol(self): """Test tol float validation""" - self.assertEqual(self.magnet.b_tolerance, 0.002) + self.assertIsNone(self.magnet.b_tolerance) self.magnet.b_tolerance = "a" - self.assertEqual(self.magnet.b_tolerance, 0.002) + self.assertIsNone(self.magnet.b_tolerance) self.magnet.b_tolerance = 1 - self.assertEqual(self.magnet.b_tolerance, 0.002) + self.assertIsNone(self.magnet.b_tolerance) self.magnet.b_tolerance = 0.1 self.assertEqual(self.magnet.b_tolerance, 0.1) def test_length(self): """Test length float validation""" - self.assertEqual(self.magnet.length, 0.1342) + self.assertIsNone(self.magnet.length) self.magnet.length = "a" - self.assertEqual(self.magnet.length, 0.1342) + self.assertIsNone(self.magnet.length) self.magnet.length = 1 - self.assertEqual(self.magnet.length, 0.1342) + self.assertIsNone(self.magnet.length) self.magnet.length = 0.05 self.assertEqual(self.magnet.length, 0.05) @@ -111,6 +100,12 @@ def test_bdes(self, mock_pv_get): self.assertEqual(self.magnet.bdes, 0.5) mock_pv_get.assert_called_once() + @patch("epics.PV.put", new_callable=Mock) + def test_set_bdes(self, mock_pv_put): + mock_pv_put.return_value = None + self.magnet.bdes = 0.1 + mock_pv_put.assert_called_once_with(value=0.1) + @patch("epics.PV.get", new_callable=Mock) def test_get_bctrl(self, mock_pv_get): mock_pv_get.return_value = 0.5 @@ -125,7 +120,7 @@ def test_get_bctrl(self, mock_pv_get): def test_set_bctrl_with_int_and_ready(self, mock_ctrl_option, mock_pv_put): mock_ctrl_option.return_value = "Ready" self.magnet.bctrl = 3 - mock_pv_put.assert_called_once_with(3) + mock_pv_put.assert_called_once_with(value=3) @patch("epics.PV.put", new_callable=Mock) @patch( @@ -177,7 +172,7 @@ def test_control_functions_call_pv_put_if_ready( } for option, func in options_and_getter_function.items(): func() - pv_put_mock.assert_called_once_with(self.magnet._ctrl_options[option]) + pv_put_mock.assert_called_once_with(self.magnet.ctrl_options[option]) pv_put_mock.reset_mock() @patch("epics.PV.put", new_callable=Mock) @@ -211,5 +206,85 @@ def test_trim_does_nothing_if_not_ready(self, mock_ctrl_option, pv_put_mock): if option in options_requiring_state_check: pv_put_mock.assert_not_called() else: - pv_put_mock.assert_called_once_with(self.magnet._ctrl_options[option]) + pv_put_mock.assert_called_once_with(self.magnet.ctrl_options[option]) pv_put_mock.reset_mock() + + +class MagnetCollectionTest(TestCase): + def setUp(self) -> None: + self.magnet_collection = create_magnet(area="GUNB") + return super().setUp() + + def test_magnet_collection_creation(self): + self.assertIsInstance(self.magnet_collection, MagnetCollection) + + @patch( + "lcls_tools.common.devices.magnet.magnet.Magnet.bdes", new_callable=PropertyMock + ) + @patch("lcls_tools.common.devices.magnet.magnet.Magnet.trim", new_callable=Mock) + def test_set_bdes_with_no_args(self, mock_trim, mock_bdes): + with self.assertRaises(TypeError): + self.magnet_collection.set_bdes() + self.magnet_collection.set_bdes(magnet_dict={}) + mock_bdes.assert_not_called() + mock_trim.assert_not_called() + + @patch("epics.PV.put", new_callable=Mock) + @patch("lcls_tools.common.devices.magnet.magnet.Magnet.trim", new_callable=Mock) + def test_set_bdes_with_args(self, mock_trim, mock_bdes_put): + bdes_settings = { + "SOL1B": 0.1, + } + self.magnet_collection.set_bdes(magnet_dict=bdes_settings) + mock_bdes_put.assert_called_once_with(value=0.1) + mock_trim.assert_called_once() + + @patch("epics.PV.put", new_callable=Mock) + @patch("lcls_tools.common.devices.magnet.magnet.Magnet.trim", new_callable=Mock) + def test_set_bdes_with_bad_magnet_name(self, mock_trim, mock_bdes_put): + bdes_settings = { + "BAD-MAG": 0.3, + } + self.magnet_collection.set_bdes(magnet_dict=bdes_settings) + mock_bdes_put.assert_not_called() + mock_trim.assert_not_called() + + @patch( + "lcls_tools.common.devices.magnet.magnet.MagnetCollection.set_bdes", + new_callable=Mock, + ) + def test_scan_with_no_callable(self, mock_set_bdes): + settings = [ + { + "SOL1B": 0.1, + }, + { + "SOL1B": 0.15, + }, + { + "SOL1B": 0.2, + }, + ] + self.magnet_collection.scan(scan_settings=settings) + self.assertEqual(mock_set_bdes.call_count, 3) + + @patch( + "lcls_tools.common.devices.magnet.magnet.MagnetCollection.set_bdes", + new_callable=Mock, + ) + def test_scan_with_callable(self, mock_set_bdes): + mock_daq_function = Mock() + settings = [ + { + "SOL1B": 0.1, + }, + { + "SOL1B": 0.15, + }, + { + "SOL1B": 0.2, + }, + ] + self.magnet_collection.scan(scan_settings=settings, function=mock_daq_function) + self.assertEqual(mock_set_bdes.call_count, 3) + self.assertEqual(mock_daq_function.call_count, 3) diff --git a/tests/unit_tests/lcls_tools/common/devices/magnet/test_reader.py b/tests/unit_tests/lcls_tools/common/devices/magnet/test_reader.py index 3bbdc787..917520cb 100644 --- a/tests/unit_tests/lcls_tools/common/devices/magnet/test_reader.py +++ b/tests/unit_tests/lcls_tools/common/devices/magnet/test_reader.py @@ -1,6 +1,7 @@ from lcls_tools.common.devices.magnet.reader import create_magnet, _find_yaml_file -from lcls_tools.common.devices.magnet.magnet import Magnet +from lcls_tools.common.devices.magnet.magnet import Magnet, MagnetCollection import unittest +from unittest.mock import patch, MagicMock import os @@ -13,40 +14,48 @@ def setUp(self) -> None: def test_bad_file_location_raises_when_finding(self): with self.assertRaises(FileNotFoundError): - _find_yaml_file(yaml_filename="bad-filename.yaml") + _find_yaml_file(area="bad-area") def test_bad_file_location_when_creating_magnet_returns_none(self): - self.assertIsNone(create_magnet(yaml_filename="bad-filename.yml")) + self.assertIsNone(create_magnet("bad-area")) def test_no_file_location_when_creating_magnet_returns_none(self): - self.assertIsNone(create_magnet(yaml_filename=None)) + self.assertIsNone(create_magnet()) def test_magnet_name_not_in_file_when_creating_magnet_returns_none(self): - self.assertIsNone( - create_magnet(yaml_filename=self.typical_config, name="BAD-MAGNET-NAME") - ) - - def test_config_with_no_control_information_returns_none(self): - self.assertIsNone(create_magnet(self.bad_config, "SOL2B")) - - def test_config_with_no_metadata_returns_none(self): - self.assertIsNone(create_magnet(self.bad_config, "SOL1B")) + self.assertIsNone(create_magnet(area="GUNB", name="BAD-MAGNET-NAME")) + + @patch( + "lcls_tools.common.devices.magnet.reader._find_yaml_file", + new_callable=MagicMock(), + ) + def test_config_with_no_control_information_returns_none(self, mock_find_yaml): + mock_find_yaml.return_value = self.bad_config + self.assertIsNone(create_magnet(area="GUNX", name="CQ02B")) + + @patch( + "lcls_tools.common.devices.magnet.reader._find_yaml_file", + new_callable=MagicMock(), + ) + def test_config_with_no_metadata_returns_none(self, mock_find_yaml): + mock_find_yaml.return_value = self.bad_config + self.assertIsNone(create_magnet(area="GUNX", name="SOL1B")) def test_create_magnet_with_only_config_creates_all_magnets(self): - result = create_magnet(yaml_filename=self.typical_config) - self.assertIsInstance(result, dict) + result = create_magnet(area="GUNB") + self.assertIsInstance(result, MagnetCollection) for name in [ "SOL2B", "SOL1B", ]: - self.assertIn(name, result, msg=f"expected {name} in {result}.") - self.assertIsInstance(result[name], Magnet) + self.assertIn(name, result.magnets, msg=f"expected {name} in {result}.") + self.assertIsInstance(result.magnets[name], Magnet) def test_create_magnet_with_config_and_name_creates_one_magnet(self): name = "SOL1B" result = create_magnet( - yaml_filename=self.typical_config, + area="GUNB", name=name, ) - self.assertNotIsInstance(result, dict) + self.assertNotIsInstance(result, MagnetCollection) self.assertIsInstance(result, Magnet) diff --git a/tests/unit_tests/lcls_tools/common/devices/test_device.py b/tests/unit_tests/lcls_tools/common/devices/test_device.py index ecd84b6a..820a0259 100644 --- a/tests/unit_tests/lcls_tools/common/devices/test_device.py +++ b/tests/unit_tests/lcls_tools/common/devices/test_device.py @@ -1,11 +1,11 @@ import os import unittest from unittest.mock import MagicMock, patch +from pydantic import ValidationError from lcls_tools.common.devices.device import ( Device, ApplyDeviceCallbackError, RemoveDeviceCallbackError, - MandatoryFieldNotFoundInYAMLError, ) from epics import PV import yaml @@ -18,36 +18,21 @@ def setUp(self) -> None: with open(self.config_filename, "r") as file: self.device_name = "DEVICE_1" self.config_data = yaml.safe_load(file) - self.device = Device( - name=self.device_name, config=self.config_data[self.device_name] - ) + device_data = self.config_data[self.device_name] + device_data.update({"name": self.device_name}) + self.device = Device(**device_data) self.pv_obj = PV("SOLN:GUNB:212:BACT") return super().setUp() def test_config_with_no_control_information_field_raises(self): name_of_device_with_missing_data = "DEVICE_2" - with self.assertRaises(MandatoryFieldNotFoundInYAMLError): - Device( - name=name_of_device_with_missing_data, - config=self.config_data[name_of_device_with_missing_data], - ) + with self.assertRaises(ValidationError): + Device(**self.config_data[name_of_device_with_missing_data]) def test_config_with_no_metadata_field_raises(self): name_of_device_with_missing_data = "DEVICE_3" - with self.assertRaises(MandatoryFieldNotFoundInYAMLError): - Device( - name=name_of_device_with_missing_data, - config=self.config_data[name_of_device_with_missing_data], - ) - - def test_properties_exist(self): - """Test that all the properties we expect exist""" - # Assert that device has all auto-generated private attributes - for item in self.device._mandatory_fields: - self.assertTrue( - hasattr(self.device, item), - msg=f"expected device to have attribute {item}", - ) + with self.assertRaises(ValidationError): + Device(**self.config_data[name_of_device_with_missing_data]) def test_name_property_assigned_after_init(self): self.assertEqual(self.device.name, self.device_name)