From 7939f6b636408ec88517682b2b78c7d5aa89215c Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Thu, 13 Apr 2023 12:49:03 +0000 Subject: [PATCH 1/8] nitpicky true, update typing --- docs/conf.py | 11 +++++++-- src/pytac/cs.py | 10 ++++---- src/pytac/data_source.py | 2 +- src/pytac/device.py | 8 +++---- src/pytac/lattice.py | 10 ++++---- src/pytac/units.py | 52 ++++++++++++++++++++-------------------- src/pytac/utils.py | 4 ++-- 7 files changed, 52 insertions(+), 45 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index e764ee86..a58b0060 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -48,13 +48,14 @@ # If true, Sphinx will warn about all references where the target cannot # be found. -nitpicky = False +nitpicky = True # A list of (type, target) tuples (by default empty) that should be ignored when # generating warnings in "nitpicky mode". Note that type should include the # domain name if present. Example entries would be ('py:func', 'int') or # ('envvar', 'LD_LIBRARY_PATH'). nitpick_ignore = [ + # Defaults with skeleton. ("py:class", "NoneType"), ("py:class", "'str'"), ("py:class", "'float'"), @@ -63,6 +64,9 @@ ("py:class", "'object'"), ("py:class", "'id'"), ("py:class", "typing_extensions.Literal"), + # Added, but custom for pytac. + ("py:class", "poly1d"), + ("py:class", "PchipInterpolator"), ] # Both the class’ and the __init__ method’s docstring are concatenated and @@ -98,7 +102,10 @@ # This means you can link things like `str` and `asyncio` to the relevant # docs in the python documentation. -intersphinx_mapping = dict(python=("https://docs.python.org/3/", None)) +intersphinx_mapping = { + "python": ("https://docs.python.org/3/", None), + "numpy": ("https://numpy.org/doc/stable/", None), +} # A dictionary of graphviz graph attributes for inheritance diagrams. inheritance_graph_attrs = dict(rankdir="TB") diff --git a/src/pytac/cs.py b/src/pytac/cs.py index 858c58e7..02fe1ea5 100644 --- a/src/pytac/cs.py +++ b/src/pytac/cs.py @@ -14,7 +14,7 @@ def get_single(self, pv, throw): """Get the value of a given PV. Args: - pv (string): PV to get the value of. + pv (str): PV to get the value of. readback or a setpoint PV. throw (bool): On failure: if True, raise ControlSystemException; if False, return None and log a warning. @@ -31,7 +31,7 @@ def get_multiple(self, pvs, throw): """Get the value for given PVs. Args: - pvs (sequence): PVs to get values of. + pvs (typing.Sequence): PVs to get values of. throw (bool): On failure: if True, raise ControlSystemException; if False, None will be returned for any PV that fails and a warning will be logged. @@ -48,7 +48,7 @@ def set_single(self, pv, value, throw): """Set the value of a given PV. Args: - pv (string): The PV to set the value of. + pv (str): The PV to set the value of. value (object): The value to set the PV to. throw (bool): On failure: if True, raise ControlSystemException: if False, log a warning. @@ -62,8 +62,8 @@ def set_multiple(self, pvs, values, throw): """Set the values for given PVs. Args: - pvs (sequence): PVs to set the values of. - values (sequence): values to set no the PVs. + pvs (typing.Sequence): PVs to set the values of. + values (typing.Sequence): values to set no the PVs. throw (bool): On failure, if True raise ControlSystemException, if False return a list of True and False values corresponding to successes and failures and log a diff --git a/src/pytac/data_source.py b/src/pytac/data_source.py index 53949d73..4cc8838a 100644 --- a/src/pytac/data_source.py +++ b/src/pytac/data_source.py @@ -21,7 +21,7 @@ def get_fields(self): """Get all the fields represented by this data source. Returns: - iterable: all fields. + typing.Iterable: all fields. """ raise NotImplementedError() diff --git a/src/pytac/device.py b/src/pytac/device.py index f385f39b..01f5cb2f 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -96,7 +96,7 @@ def get_value(self, handle=None, throw=None): used, only supported to conform with the base class. Returns: - numeric: the value of the device. + Union(float, int): the value of the device. """ return self._value @@ -104,7 +104,7 @@ def set_value(self, value, throw=None): """Set the value on the device. Args: - value (numeric): the value to set. + value (Union(float, int)): the value to set. throw (bool): Irrelevant in this case as a control system is not used, only supported to conform with the base class. """ @@ -130,7 +130,7 @@ class EpicsDevice(Device): .. Private Attributes: _cs (ControlSystem): The control system object used to get and set the value of a PV. - _enabled (bool-like): Whether the device is enabled. May be a + _enabled (Union(bool, PvEnabler)): Whether the device is enabled. May be a PvEnabler object. """ @@ -140,7 +140,7 @@ def __init__(self, name, cs, enabled=True, rb_pv=None, sp_pv=None): name (str): The prefix of EPICS PV for this device. cs (ControlSystem): The control system object used to get and set the value of a PV. - enabled (bool-like): Whether the device is enabled. May be a + enabled (Union(bool, PvEnabler)): Whether the device is enabled. May be a PvEnabler object. rb_pv (str): The EPICS readback PV. sp_pv (str): The EPICS setpoint PV. diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 86e0c949..470ee1cc 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -385,7 +385,7 @@ def get_element_values( numpy array of the specified type. Returns: - list or numpy.array: The requested values. + Union(list, numpy.ndarray): The requested values. """ elements = self.get_elements(family) values = [ @@ -411,7 +411,7 @@ def set_element_values( Args: family (str): family of elements on which to set values. field (str): field to set values for. - values (sequence): A list of values to assign. + values (typing.Sequence): A list of values to assign. units (str): pytac.ENG or pytac.PHYS. data_source (str): pytac.LIVE or pytac.SIM. throw (bool): On failure, if True raise ControlSystemException, if @@ -507,7 +507,7 @@ def convert_family_values(self, family, field, values, origin, target): Args: family (str): the family of elements which the values belong to. field (str): the field on the elements which the values are from. - values (sequence): values to be converted. + values (typing.Sequence): values to be converted. origin (str): pytac.ENG or pytac.PHYS. target (str): pytac.ENG or pytac.PHYS. """ @@ -627,7 +627,7 @@ def get_element_values( numpy array of the specified type. Returns: - list or numpy.array: The requested values. + Union(list, numpy.ndarray): The requested values. """ if data_source == pytac.DEFAULT: data_source = self.get_default_data_source() @@ -663,7 +663,7 @@ def set_element_values( Args: family (str): family of elements on which to set values. field (str): field to set values for. - values (sequence): A list of values to assign. + values (typing.Sequence): A list of values to assign. units (str): pytac.ENG or pytac.PHYS. data_source (str): pytac.LIVE or pytac.SIM. throw (bool): On failure: if True, raise ControlSystemException: if diff --git a/src/pytac/units.py b/src/pytac/units.py index fc8ac1fe..98f89869 100644 --- a/src/pytac/units.py +++ b/src/pytac/units.py @@ -39,10 +39,10 @@ class UnitConv: phys_units (str): The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (function): Function to be applied after the - initial conversion. - _pre_phys_to_eng (function): Function to be applied before the - initial conversion. + _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied + after the initial conversion. + _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied + before the initial conversion. """ def __init__( @@ -55,10 +55,10 @@ def __init__( ): """ Args: - post_eng_to_phys (function): Function to be applied after the - initial conversion. - pre_phys_to_eng (function): Function to be applied before the - initial conversion. + post_eng_to_phys (typing.Callable[[float], float]): Function to be applied + after the initial conversion. + pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied + before the initial conversion. engineering_units (str): The unit type of the post conversion engineering value. physics_units (str): The unit type of the post conversion physics @@ -85,8 +85,8 @@ def set_post_eng_to_phys(self, post_eng_to_phys): """Set the function to be applied after the initial conversion. Args: - post_eng_to_phys (function): Function to be applied after the - initial conversion. + post_eng_to_phys (typing.Callable[[float], float]): Function to be applied + after the initial conversion. """ self._post_eng_to_phys = post_eng_to_phys @@ -94,8 +94,8 @@ def set_pre_phys_to_eng(self, pre_phys_to_eng): """Set the function to be applied before the initial conversion. Args: - pre_phys_to_eng (function): Function to be applied before the - initial conversion. + pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied + before the initial conversion. """ self._pre_phys_to_eng = pre_phys_to_eng @@ -275,10 +275,10 @@ class PolyUnitConv(UnitConv): phys_units (str): The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (function): Function to be applied after the - initial conversion. - _pre_phys_to_eng (function): Function to be applied before the - initial conversion. + _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied + after the initial conversion. + _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied + before the initial conversion. """ def __init__( @@ -292,8 +292,8 @@ def __init__( ): """ Args: - coef (array-like): The polynomial's coefficients, in decreasing - powers. + coef (numpy.ndarray[typing.Any, numpy.dtype[numpy.generic]]): The + polynomial's coefficients, in decreasing powers. post_eng_to_phys (float): The value after conversion between ENG and PHYS. pre_eng_to_phys (float): The value before conversion. @@ -360,10 +360,10 @@ class PchipUnitConv(UnitConv): phys_units (str): The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (function): Function to be applied after the - initial conversion. - _pre_phys_to_eng (function): Function to be applied before the - initial conversion. + _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied + after the initial conversion. + _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied + before the initial conversion. """ def __init__( @@ -456,10 +456,10 @@ class NullUnitConv(UnitConv): phys_units (str): The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (function): Always unit_function as no conversion - is performed. - _pre_phys_to_eng (function): Always unit_function as no conversion - is performed. + _post_eng_to_phys (typing.Callable[[float], float]): Always unit_function as + no conversion is performed. + _pre_phys_to_eng (typing.Callable[[float], float]): Always unit_function as + no conversion is performed. """ def __init__(self, engineering_units="", physics_units=""): diff --git a/src/pytac/utils.py b/src/pytac/utils.py index bccab0ba..57240236 100644 --- a/src/pytac/utils.py +++ b/src/pytac/utils.py @@ -28,7 +28,7 @@ def get_div_rigidity(energy): energy (int): the energy of the lattice. Returns: - function: div rigidity. + typing.Callable[[int], float]: div rigidity. """ rigidity = get_rigidity(energy) @@ -44,7 +44,7 @@ def get_mult_rigidity(energy): energy (int): the energy of the lattice. Returns: - function: mult rigidity. + typing.Callable[[int], float]: mult rigidity. """ rigidity = get_rigidity(energy) From 6dd8e7a4112d0a3ced0523885b76dc7aa46a2039 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Tue, 29 Aug 2023 14:27:56 +0000 Subject: [PATCH 2/8] Update all typing. --- pyproject.toml | 1 + src/pytac/cothread_cs.py | 81 ++++---- src/pytac/cs.py | 54 +++--- src/pytac/data_source.py | 183 +++++++++--------- src/pytac/device.py | 182 ++++++++++-------- src/pytac/element.py | 180 ++++++++++-------- src/pytac/lattice.py | 401 ++++++++++++++++++++------------------- src/pytac/load_csv.py | 34 ++-- src/pytac/units.py | 321 ++++++++++++++----------------- src/pytac/utils.py | 28 +-- 10 files changed, 763 insertions(+), 702 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 3de08a88..11bb27df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,6 +64,7 @@ profile = "black" extend-ignore = [ "E203", # See https://github.com/PyCQA/pycodestyle/issues/373 "E402", # allow isort:skip + "E502", # ignore line length "F811", # support typing.overload decorator "F722", # allow Annotated[typ, some_func("some string")] ] diff --git a/src/pytac/cothread_cs.py b/src/pytac/cothread_cs.py index cf122fed..9b78deb6 100644 --- a/src/pytac/cothread_cs.py +++ b/src/pytac/cothread_cs.py @@ -1,4 +1,5 @@ import logging +from typing import Any, List, Optional, Sequence from cothread.catools import ca_nothing, caget, caput @@ -12,24 +13,37 @@ class CothreadControlSystem(ControlSystem): N.B. this is the default control system. It is used to communicate over channel access with the hardware in the ring. - **Methods:** + Attributes: + _timeout: Timeout in seconds for the caget operations. + _wait: Caput operations will wait until the server acknowledges successful + completion before returning. """ - def __init__(self, timeout=1.0, wait=False): + _timeout: float + _wait: bool + + def __init__(self, timeout: float = 1.0, wait: bool = False) -> None: + """Initialise the Cothread control system. + + Args: + _timeout: Timeout in seconds for the caget operation. + _wait: Caput operations will wait until the server acknowledges successful + completion before returning. + """ self._timeout = timeout self._wait = wait - def get_single(self, pv, throw=True): + def get_single(self, pv: str, throw: bool = True) -> Optional[Any]: """Get the value of a given PV. Args: - pv (string): The process variable given as a string. It can be a - readback or a setpoint PV. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + pv: The process variable given as a string. It can be a readback or a + setpoint PV. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - object: the current value of the given PV. + The current value of the given PV. Raises: ControlSystemException: if it cannot connect to the specified PV. @@ -44,24 +58,23 @@ def get_single(self, pv, throw=True): logging.warning(error_msg) return None - def get_multiple(self, pvs, throw=True): + def get_multiple(self, pvs: Sequence[str], throw: bool = True) -> List[Any]: """Get the value for given PVs. Args: - pvs (sequence): PVs to get values of. - throw (bool): On failure: if True, raise ControlSystemException; if - False, None will be returned for any PV that fails - and a warning will be logged. + pvs: PVs to get values of. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - sequence: the current values of the PVs. + The current values of the PVs. Raises: ControlSystemException: if it cannot connect to one or more PVs. """ results = caget(pvs, timeout=self._timeout, throw=False) - return_values = [] - failures = [] + return_values: List[Optional[Any]] = [] + failures: List[Any] = [] for result in results: if isinstance(result, ca_nothing): logging.warning(f"Cannot connect to {result.name}.") @@ -75,17 +88,17 @@ def get_multiple(self, pvs, throw=True): raise ControlSystemException(f"{len(failures)} caget calls failed.") return return_values - def set_single(self, pv, value, throw=True): + def set_single(self, pv: str, value: Any, throw: bool = True) -> bool: """Set the value of a given PV. Args: - pv (string): PV to set the value of. - value (object): The value to set the PV to. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + pv: PV to set the value of. + value: The value to set the PV to. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - bool: True for success, False for failure + True for success, False for failure Raises: ControlSystemException: if it cannot connect to the specified PV. @@ -101,20 +114,21 @@ def set_single(self, pv, value, throw=True): logging.warning(error_msg) return False - def set_multiple(self, pvs, values, throw=True): + def set_multiple( + self, pvs: Sequence[str], values: Any, throw: bool = True + ) -> Optional[List[bool]]: """Set the values for given PVs. Args: - pvs (sequence): PVs to set the values of. - values (sequence): values to set to the PVs. - throw (bool): On failure, if True raise ControlSystemException, if - False return a list of True and False values - corresponding to successes and failures and log a - warning for each PV that fails. + pvs: PVs to set the values of. + values: values to set to the PVs. + throw: On failure, if True raise ControlSystemException, if False return + a list of True and False values corresponding to successes and failures + and log a warning for each PV that fails. Returns: - list(bool): True for success, False for failure; only returned if - throw is false and a failure occurs. + True for success, False for failure; only returned if throw is false and a + failure occurs. Raises: ValueError: if the lists of values and PVs are diffent lengths. @@ -123,8 +137,8 @@ def set_multiple(self, pvs, values, throw=True): if len(pvs) != len(values): raise ValueError("Please enter the same number of values as PVs.") status = caput(pvs, values, timeout=self._timeout, throw=False, wait=self._wait) - return_values = [] - failures = [] + return_values: List[bool] = [] + failures: List[Any] = [] for stat in status: if not stat.ok: return_values.append(False) @@ -137,3 +151,4 @@ def set_multiple(self, pvs, values, throw=True): raise ControlSystemException(f"{len(failures)} caput calls failed.") else: return return_values + return None diff --git a/src/pytac/cs.py b/src/pytac/cs.py index 02fe1ea5..249a19e3 100644 --- a/src/pytac/cs.py +++ b/src/pytac/cs.py @@ -1,77 +1,75 @@ """Class representing an abstract control system.""" +from typing import Any, List, Sequence + + class ControlSystem(object): """Abstract base class representing a control system. A specialised implementation of this class would be used to communicate over channel access with the hardware in the ring. - - **Methods:** """ - def get_single(self, pv, throw): + def get_single(self, pv: str, throw: bool) -> Any: """Get the value of a given PV. Args: - pv (str): PV to get the value of. - readback or a setpoint PV. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + pv: PV to get the value of readback or a setpoint PV. + throw: On failure: if True, raise ControlSystemException; if False, + return None and log a warning. Returns: - object: the current value of the given PV. + The current value of the given PV. Raises: ControlSystemException: if it cannot connect to the specified PVs. """ raise NotImplementedError() - def get_multiple(self, pvs, throw): + def get_multiple(self, pvs: Sequence[str], throw: bool) -> List[Any]: """Get the value for given PVs. Args: - pvs (typing.Sequence): PVs to get values of. - throw (bool): On failure: if True, raise ControlSystemException; if - False, None will be returned for any PV that fails - and a warning will be logged. + pvs: PVs to get values of. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - list(object): the current values of the PVs. + The current values of the PVs. Raises: ControlSystemException: if it cannot connect to the specified PV. """ raise NotImplementedError() - def set_single(self, pv, value, throw): + def set_single(self, pv: str, value: Any, throw: bool): """Set the value of a given PV. Args: - pv (str): The PV to set the value of. - value (object): The value to set the PV to. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + pv: The PV to set the value of. + value: The value to set the PV to. + throw: On failure: if True, raise ControlSystemException: if False, + log a warning. Raises: ControlSystemException: if it cannot connect to the specified PV. """ raise NotImplementedError() - def set_multiple(self, pvs, values, throw): + def set_multiple(self, pvs: Sequence[str], values: Sequence[Any], throw: bool): """Set the values for given PVs. Args: - pvs (typing.Sequence): PVs to set the values of. - values (typing.Sequence): values to set no the PVs. - throw (bool): On failure, if True raise ControlSystemException, if - False return a list of True and False values - corresponding to successes and failures and log a - warning for each PV that fails. + pvs: PVs to set the values of. + values: values to set no the PVs. + throw: On failure, if True raise ControlSystemException, if False return a + list of True and False values corresponding to successes and failures + and log a warning for each PV that fails. Raises: - ValueError: if the PVs or values are not passed in as sequences - or if they have different lengths + ValueError: if the PVs or values are not passed in as sequences or if they + have different lengths ControlSystemException: if it cannot connect to one or more PVs. """ raise NotImplementedError() diff --git a/src/pytac/data_source.py b/src/pytac/data_source.py index 4cc8838a..0ed0050d 100644 --- a/src/pytac/data_source.py +++ b/src/pytac/data_source.py @@ -1,6 +1,12 @@ """Module containing pytac data source classes.""" +from typing import Any, Dict, Iterable, List, Union + +from _collections_abc import KeysView + import pytac +from pytac.device import Device from pytac.exceptions import DataSourceException, FieldException +from pytac.units import UnitConv class DataSource(object): @@ -9,46 +15,44 @@ class DataSource(object): Typically an instance would represent hardware via a control system, or a simulation. - **Attributes:** - Attributes: - units (str): pytac.PHYS or pytac.ENG. - - **Methods:** + units: pytac.PHYS or pytac.ENG. """ - def get_fields(self): + units: str + + def get_fields(self) -> Iterable: """Get all the fields represented by this data source. Returns: - typing.Iterable: all fields. + All fields. """ raise NotImplementedError() - def get_value(self, field, handle, throw): + def get_value(self, field: str, handle: str, throw: bool) -> Any: """Get a value for a field. Args: - field (str): field of the requested value. - handle (str): pytac.RB or pytac.SP - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + field: field of the requested value. + handle: pytac.RB or pytac.SP + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - float: value for specified field and handle. + Value for specified field and handle. """ raise NotImplementedError() - def set_value(self, field, value, throw): + def set_value(self, field: str, value: float, throw: bool) -> Any: """Set a value for a field. This is always set to pytac.SP, never pytac.RB. Args: - field (str): field to set. - value (float): value to set. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + field: field to set. + value: value to set. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. """ raise NotImplementedError() @@ -62,42 +66,44 @@ class DataSourceManager(object): also held here. Attributes: - default_units (str): Holds the current default unit type, pytac.PHYS or - pytac.ENG, for an element or lattice. - default_data_source (str): Holds the current default data source, - pytac.LIVE or pytac.SIM, for an element or - lattice. + default_units: Holds the current default unit type, pytac.PHYS or pytac.ENG, + for an element or lattice. + default_data_source: Holds the current default data source, pytac.LIVE or + pytac.SIM, for an element or lattice. .. Private Attributes: - _data_sources (dict): A dictionary of the data sources held. - _uc (dict): A dictionary of the unit conversion objects for each - key(field). - - **Methods:** + _data_sources: A dictionary of the data sources held. + _uc: A dictionary of the unit conversion objects for each key(field). """ - def __init__(self): - self._data_sources = {} - self._uc = {} + default_units: str + default_data_source: str + + _data_sources: Dict[str, DataSource] + _uc: Dict[str, UnitConv] + + def __init__(self) -> None: self.default_units = pytac.ENG self.default_data_source = pytac.LIVE + self._data_sources = {} + self._uc = {} - def set_data_source(self, data_source, data_source_type): + def set_data_source(self, data_source: DataSource, data_source_type: str) -> None: """Add a data source to the manager. Args: - data_source (DataSource): the data source to be set. - data_source_type (str): the type of the data source being set - pytac.LIVE or pytac.SIM. + data_source: the data source to be set. + data_source_type: the type of the data source being set + pytac.LIVE or pytac.SIM. """ self._data_sources[data_source_type] = data_source - def get_data_source(self, data_source_type): + def get_data_source(self, data_source_type: str) -> DataSource: """Get a data source. Args: - data_source_type (str): the type of the data source being set - pytac.LIVE or pytac.SIM. + data_source_type: the type of the data source being set + pytac.LIVE or pytac.SIM. Raises: DataSourceException: if there is no data source on the given field. @@ -109,21 +115,21 @@ def get_data_source(self, data_source_type): f"No data source {data_source_type} on manager {self}." ) - def get_fields(self): + def get_fields(self) -> Dict[str, Iterable]: """Get all the fields defined on the manager. Includes all fields defined by all data sources. Returns: - dict: A dictionary of all the fields defined on the manager, - separated by data source(key). + A dictionary of all the fields defined on the manager, separated by + data source(key). """ - fields = {} + fields: Dict[str, Iterable] = {} for data_source in self._data_sources: fields[data_source] = self._data_sources[data_source].get_fields() return fields - def add_device(self, field, device, uc): + def add_device(self, field: str, device: Device, uc: UnitConv) -> None: """Add device and unit conversion objects to a given field. A DeviceDataSource must be set before calling this method, this @@ -131,10 +137,9 @@ def add_device(self, field, device, uc): uses devices. Args: - field (str): The key to store the unit conversion and device - objects. - device (Device): The device object used for this field. - uc (UnitConv): The unit conversion object used for this field. + field: The key to store the unit conversion and device objects. + device: The device object used for this field. + uc: The unit conversion object used for this field. Raises: DataSourceException: if no DeviceDataSource is set. @@ -142,7 +147,7 @@ def add_device(self, field, device, uc): self.get_data_source(pytac.LIVE).add_device(field, device) self.set_unitconv(field, uc) - def get_device(self, field): + def get_device(self, field: str) -> Device: """Get the device for the given field. A DeviceDataSource must be set before calling this method, this @@ -150,21 +155,21 @@ def get_device(self, field): uses devices. Args: - field (str): The lookup key to find the device on the manager. + field: The lookup key to find the device on the manager. Returns: - Device: The device on the given field. + The device on the given field. Raises: DataSourceException: if no DeviceDataSource is set. """ return self.get_data_source(pytac.LIVE).get_device(field) - def get_unitconv(self, field): + def get_unitconv(self, field: str) -> UnitConv: """Get the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. + field: The field associated with this conversion. Returns: UnitConv: The object associated with the specified field. @@ -179,12 +184,12 @@ def get_unitconv(self, field): f"No unit conversion option for field {field} on manager {self}." ) - def set_unitconv(self, field, uc): + def set_unitconv(self, field: str, uc: UnitConv) -> None: """set the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. - uc (UnitConv): The unit conversion object to be set. + field: The field associated with this conversion. + uc: The unit conversion object to be set. """ self._uc[field] = uc @@ -209,8 +214,8 @@ def get_value( handle: pytac.SP or pytac.RB. units: pytac.ENG or pytac.PHYS returned. data_source: pytac.LIVE or pytac.SIM. - throw: On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: The value of the requested field @@ -247,8 +252,8 @@ def set_value( value: The value to set. units: pytac.ENG or pytac.PHYS. data_source_type: pytac.LIVE or pytac.SIM. - throw: On failure: if True, raise ControlSystemException: if - False, log a warning. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: HandleException: if the specified handle is not pytac.SP. @@ -269,82 +274,88 @@ def set_value( class DeviceDataSource(DataSource): """Data source containing control system devices. - **Attributes:** - Attributes: - units (str): pytac.ENG or pytac.PHYS, pytac.ENG by default. + units: pytac.ENG or pytac.PHYS, pytac.ENG by default. .. Private Attributes: - _devices (dict): A dictionary of the devices for each key(field). - - **Methods:** + _devices: A dictionary of the devices for each key(field). """ + units: str + + _devices: Dict[str, Device] + def __init__(self): self._devices = {} self.units = pytac.ENG - def add_device(self, field, device): + def add_device(self, field: str, device: Device) -> None: """Add device to this data_source. Args: - field (str): field this device represents. - device (Device): device object. + field: field this device represents. + device: device object. """ self._devices[field] = device - def get_device(self, field): + def get_device(self, field: str) -> Device: """Get device from the data_source. Args: - field (str): field of the requested device. + field: field of the requested device. Returns: - Device: The device of the specified field. + The device of the specified field. Raises: - FieldException: if the specified field doesn't exist on this data - source. + FieldException: if the specified field doesn't exist on this data source. """ try: return self._devices[field] except KeyError: raise FieldException(f"No field {field} on data source {self}.") - def get_fields(self): + def get_fields(self) -> KeysView: """Get all the fields from the data_source. Returns: - list: list of strings of all the fields of the data_source. + List of strings of all the fields of the data_source. """ return self._devices.keys() - def get_value(self, field, handle, throw=True): + def get_value( + self, field: str, handle: str, throw: bool = True + ) -> Union[float, int, List[float], List[int]]: """Get the value of a readback or setpoint PV for a field from the data_source. Args: - field (str): field of the requested value. - handle (str): pytac.RB or pytac.SP. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + field: field of the requested value. + handle: pytac.RB or pytac.SP. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - float: The value of the PV. + The value of the PV. Raises: FieldException: if the device does not have the specified field. """ return self.get_device(field).get_value(handle, throw) - def set_value(self, field, value, throw=True): + def set_value( + self, + field: str, + value: Union[float, int, List[int], List[float]], + throw: bool = True, + ) -> None: """Set the value of a readback or setpoint PV for a field from the data_source. Args: - field (str): field for the requested value. - value (float): The value to set on the PV. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + field: field for the requested value. + value: The value to set on the PV. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: FieldException: if the device does not have the specified field. diff --git a/src/pytac/device.py b/src/pytac/device.py index 01f5cb2f..a1af5892 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -5,9 +5,10 @@ DLS is a sextupole magnet that contains also horizontal and vertical corrector magnets and a skew quadrupole. """ -from typing import List, Union +from typing import List, Optional, Union import pytac +from pytac.cs import ControlSystem from pytac.exceptions import DataSourceException, HandleException @@ -16,8 +17,6 @@ class Device: Typically a control system will be used to set and get values on a device. - - **Methods:** """ def is_enabled(self) -> bool: @@ -28,26 +27,30 @@ def is_enabled(self) -> bool: """ raise NotImplementedError() - def get_value(self, handle: str, throw: bool) -> float: + def get_value( + self, handle: str, throw: bool + ) -> Union[float, int, List[int], List[float]]: """Read the value from the device. Args: handle: pytac.SP or pytac.RB. - throw: On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: the value of the PV. """ raise NotImplementedError() - def set_value(self, value: float, throw: bool) -> None: + def set_value( + self, value: Union[float, int, List[int], List[float]], throw: bool + ) -> None: """Set the value on the device. Args: - value (float): the value to set. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + value: the value to set. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. """ raise NotImplementedError() @@ -59,54 +62,69 @@ class SimpleDevice(Device): with a simulator. In short this device acts as simple storage for data that rarely changes, as it is not affected by changes to other aspects of the accelerator. + + Attributes: + _value: The value of the device. May be a number or list of numbers. + _enabled: Whether the device is enabled. May be a PvEnabler Object. + _readonly: Whether the value may be changed. """ + _value: Union[float, int, List[float], List[int]] + _enabled: Union[bool, "PvEnabler"] + _readonly: bool + def __init__( self, - value: Union[float, List[float]], - enabled: bool = True, + value: Union[float, int, List[int], List[float]], + enabled: Union[bool, "PvEnabler"] = True, readonly: bool = True, - ): - """ + ) -> None: + """Initialise the SimpleDevice object. + Args: - value: can be a number or a list of numbers. - enabled: whether the device is enabled. May be a - PvEnabler object. - readonly: whether the value may be changed. + value: The value of the device. May be a number or list of numbers. + enabled: Whether the device is enabled. May be a PvEnabler Object. + readonly: Whether the value may be changed. """ self._value = value self._enabled = enabled self._readonly = readonly - def is_enabled(self): + def is_enabled(self) -> bool: """Whether the device is enabled. Returns: - bool: whether the device is enabled. + Whether the device is enabled. """ return bool(self._enabled) - def get_value(self, handle=None, throw=None): + def get_value( + self, handle: Optional[str] = None, throw: Optional[bool] = None + ) -> Union[float, int, List[int], List[float]]: """Read the value from the device. Args: - handle (str): Irrelevant in this case as a control system is not - used, only supported to conform with the base class. - throw (bool): Irrelevant in this case as a control system is not - used, only supported to conform with the base class. + handle: Irrelevant in this case as a control system is not used, only + supported to conform with the base class. + throw: Irrelevant in this case as a control system is not used, only + supported to conform with the base class. Returns: - Union(float, int): the value of the device. + The value of the device. """ return self._value - def set_value(self, value, throw=None): + def set_value( + self, + value: Union[float, int, List[int], List[float]], + throw: Optional[bool] = None, + ) -> None: """Set the value on the device. Args: - value (Union(float, int)): the value to set. - throw (bool): Irrelevant in this case as a control system is not - used, only supported to conform with the base class. + value: the value to set. + throw: Irrelevant in this case as a control system is not used, only + supported to conform with the base class. """ if self._readonly: raise DataSourceException("Cannot change value of readonly SimpleDevice") @@ -120,35 +138,41 @@ class EpicsDevice(Device): setpoint PV is required when creating an epics device otherwise a DataSourceException is raised. The device is enabled by default. - **Attributes:** - Attributes: - name (str): The prefix of EPICS PVs for this device. - rb_pv (str): The EPICS readback PV. - sp_pv (str): The EPICS setpoint PV. + name: The prefix of EPICS PVs for this device. + rb_pv: The EPICS readback PV. + sp_pv: The EPICS setpoint PV. .. Private Attributes: - _cs (ControlSystem): The control system object used to get and set - the value of a PV. - _enabled (Union(bool, PvEnabler)): Whether the device is enabled. May be a - PvEnabler object. + _cs: The control system object used to get and set the value of a PV. + _enabled: Whether the device is enabled. May be a PvEnabler object. """ - def __init__(self, name, cs, enabled=True, rb_pv=None, sp_pv=None): + name: str + rb_pv: Optional[str] + sp_pv: Optional[str] + + _cs: ControlSystem + _enabled: Union[bool, "PvEnabler"] + + def __init__( + self, + name: str, + cs: ControlSystem, + enabled: Union[bool, "PvEnabler"] = True, + rb_pv: Optional[str] = None, + sp_pv: Optional[str] = None, + ) -> None: """ Args: - name (str): The prefix of EPICS PV for this device. - cs (ControlSystem): The control system object used to get and set - the value of a PV. - enabled (Union(bool, PvEnabler)): Whether the device is enabled. May be a - PvEnabler object. - rb_pv (str): The EPICS readback PV. - sp_pv (str): The EPICS setpoint PV. + name: The prefix of EPICS PV for this device. + cs: The control system object used to get and set the value of a PV. + enabled: Whether the device is enabled. May be a PvEnabler object. + rb_pv: The EPICS readback PV. + sp_pv: The EPICS setpoint PV. Raises: DataSourceException: if no PVs are provided. - - **Methods:** """ if rb_pv is None and sp_pv is None: raise DataSourceException( @@ -161,51 +185,53 @@ def __init__(self, name, cs, enabled=True, rb_pv=None, sp_pv=None): self.sp_pv = sp_pv self._enabled = enabled - def is_enabled(self): + def is_enabled(self) -> bool: """Whether the device is enabled. Returns: - bool: whether the device is enabled. + Whether the device is enabled. """ return bool(self._enabled) - def get_value(self, handle, throw=True): + def get_value(self, handle: str, throw: bool = True) -> Union[float, int]: """Read the value of a readback or setpoint PV. Args: - handle (str): pytac.SP or pytac.RB. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + handle: pytac.SP or pytac.RB. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - float: The value of the PV. + The value of the PV. Raises: HandleException: if the requested PV doesn't exist. """ return self._cs.get_single(self.get_pv_name(handle), throw) - def set_value(self, value, throw=True): + def set_value( + self, value: Union[float, int, List[int], List[float]], throw: bool = True + ) -> None: """Set the device value. Args: - value (float): The value to set. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + value: The value to set. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: HandleException: if no setpoint PV exists. """ self._cs.set_single(self.get_pv_name(pytac.SP), value, throw) - def get_pv_name(self, handle): + def get_pv_name(self, handle: str) -> str: """Get the PV name for the specified handle. Args: - handle (str): The readback or setpoint handle to be returned. + handle: The readback or setpoint handle to be returned. Returns: - str: A readback or setpoint PV. + A readback or setpoint PV. Raises: HandleException: if the PV doesn't exist. @@ -225,31 +251,33 @@ class PvEnabler(object): and False otherwise. .. Private Attributes: - _pv (str): The PV name. - _enabled_value (str): The value for PV for which the device should - be considered enabled. - _cs (ControlSystem): The control system object. + _pv: The PV name. + _enabled_value: The value for PV for which the device should be + considered enabled. + _cs: The control system object. """ - def __init__(self, pv, enabled_value, cs): + _pv: str + _enabled_value: str + _cs: ControlSystem + + def __init__(self, pv: str, enabled_value: str, cs: ControlSystem): """ Args: - pv (str): The PV name. - enabled_value (str): The value for PV for which the device should - be considered enabled. - cs (ControlSystem): The control system object. - - **Methods:** + pv: The PV name. + enabled_value: The value for PV for which the device should be + considered enabled. + cs: The control system object. """ self._pv = pv self._enabled_value = str(int(float(enabled_value))) self._cs = cs - def __bool__(self): + def __bool__(self) -> bool: """Used to override the 'if object' clause. Returns: - bool: True if the device should be considered enabled. + True if the device should be considered enabled. """ - pv_value = self._cs.get_single(self._pv) + pv_value = self._cs.get_single(self._pv, True) return self._enabled_value == str(int(float(pv_value))) diff --git a/src/pytac/element.py b/src/pytac/element.py index a3632298..cb06d370 100644 --- a/src/pytac/element.py +++ b/src/pytac/element.py @@ -1,7 +1,14 @@ """Module containing the element class.""" +from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Set + import pytac from pytac.data_source import DataSource, DataSourceManager +from pytac.device import Device from pytac.exceptions import DataSourceException, FieldException +from pytac.units import UnitConv + +if TYPE_CHECKING: + from pytac.lattice import Lattice class Element(object): @@ -10,33 +17,43 @@ class Element(object): An element has zero or more devices (e.g. quadrupole magnet) associated with each of its fields (e.g. 'b1' for a quadrupole). - **Attributes:** - Attributes: - name (str): The name identifying the element. The user is free to define - this for their own purposes. - type_ (str): The type of the element. The user is free to define this for - their own purposes. - length (float): The length of the element in metres. + name: The name identifying the element. The user is free to define this for + their own purposes. + type_: The type of the element. The user is free to define this for their own + purposes. + length: The length of the element in metres. .. Private Attributes: - _lattice (Lattice): The lattice to which the element belongs. - _data_source_manager (DataSourceManager): A class that manages the - data sources associated - with this element. - _families (set): The families this element is a member of, stored - as lowercase strings. + _lattice: The lattice to which the element belongs. + _data_source_manager: A class that manages the data sources associated with + this element. + _families The families this element is a member of, stored as lowercase + strings. """ - def __init__(self, length, element_type, name=None, lattice=None): - """ - Args: - length (float): The length of the element. - element_type (str): The type of the element. - name (str): The unique identifier for the element in the ring. - lattice (Lattice): The lattice to which the element belongs. + name: Optional[str] + type_: str + length: float - **Methods:** + _lattice: Optional["Lattice"] + _data_source_manager: DataSourceManager + _families: Set[Any] + + def __init__( + self, + length: float, + element_type: str, + name: Optional[str] = None, + lattice: Optional["Lattice"] = None, + ) -> None: + """Initialise the Element object. + + Args: + length: The length of the element. + element_type: The type of the element. + name: The unique identifier for the element in the ring. + lattice: The lattice to which the element belongs. """ self.name = name self.type_ = element_type @@ -47,24 +64,24 @@ def __init__(self, length, element_type, name=None, lattice=None): self._data_source_manager = DataSourceManager() @property - def index(self): - """int: The element's index within the ring, starting at 1.""" + def index(self) -> Optional[int]: + """The element's index within the ring, starting at 1.""" if self._lattice is None: return None else: return self._lattice._elements.index(self) + 1 @property - def s(self): - """float: The element's start position within the lattice in metres.""" + def s(self) -> Optional[float]: + """The element's start position within the lattice in metres.""" if self._lattice is None: return None else: return sum([el.length for el in self._lattice[: self.index - 1]]) @property - def cell(self): - """int: The lattice cell this element is within. + def cell(self) -> Optional[int]: + """The lattice cell this element is within. N.B. If the element spans multiple cells then the cell it begins in is returned (lowest cell number). @@ -77,15 +94,15 @@ def cell(self): return int(self.s / self._lattice.cell_length) + 1 @property - def families(self): + def families(self) -> Set[Any]: """set(str): All families that this element is in.""" return set(self._families) - def __str__(self): + def __str__(self) -> str: """Return a representation of an element, as a string. Returns: - str: A representation of an element. + A representation of an element. """ repn = " Non """ self._data_source_manager.set_data_source(data_source, data_source_type) - def get_fields(self): + def get_fields(self) -> Dict[str, Iterable]: """Get the all fields defined on an element. Includes all fields defined by all data sources. Returns: - dict: A dictionary of all the fields defined on an element, + A dictionary of all the fields defined on an element, separated by data source(key). """ return self._data_source_manager.get_fields() - def add_device(self, field, device, uc): + def add_device(self, field: str, device: Device, uc: UnitConv) -> None: """Add device and unit conversion objects to a given field. A DeviceDataSource must be set before calling this method, this @@ -146,10 +163,9 @@ def add_device(self, field, device, uc): uses devices. Args: - field (str): The key to store the unit conversion and device - objects. - device (Device): The device object used for this field. - uc (UnitConv): The unit conversion object used for this field. + field: The key to store the unit conversion and device objects. + device: The device object used for this field. + uc: The unit conversion object used for this field. Raises: DataSourceException: if no DeviceDataSource is set. @@ -159,7 +175,7 @@ def add_device(self, field, device, uc): except DataSourceException as e: raise DataSourceException(f"{self}: {e}.") - def get_device(self, field): + def get_device(self, field: str) -> Device: """Get the device for the given field. A DeviceDataSource must be set before calling this method, this @@ -167,10 +183,10 @@ def get_device(self, field): uses devices. Args: - field (str): The lookup key to find the device on an element. + field: The lookup key to find the device on an element. Returns: - Device: The device on the given field. + The device on the given field. Raises: DataSourceException: if no DeviceDataSource is set. @@ -180,14 +196,14 @@ def get_device(self, field): except DataSourceException as e: raise DataSourceException(f"{self}: {e}.") - def get_unitconv(self, field): + def get_unitconv(self, field: str) -> UnitConv: """Get the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. + field: The field associated with this conversion. Returns: - UnitConv: The object associated with the specified field. + The object associated with the specified field. Raises: FieldException: if no unit conversion object is present. @@ -197,28 +213,28 @@ def get_unitconv(self, field): except FieldException as e: raise FieldException(f"{self}: {e}") - def set_unitconv(self, field, uc): + def set_unitconv(self, field: str, uc: UnitConv) -> None: """Set the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. - uc (UnitConv): The unit conversion object to be set. + field: The field associated with this conversion. + uc: The unit conversion object to be set. """ self._data_source_manager.set_unitconv(field, uc) - def add_to_family(self, family): + def add_to_family(self, family: str) -> None: """Add the element to the specified family. Args: - family (str): Represents the name of the family. + family: Represents the name of the family. """ self._families.add(family.lower()) - def is_in_family(self, family): + def is_in_family(self, family: str) -> bool: """Return true if the element is in the specified family. Args: - family (str): Family to check. + family: Family to check. Returns: true if element is in the specified family. @@ -227,12 +243,12 @@ def is_in_family(self, family): def get_value( self, - field, - handle=pytac.RB, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - ): + field: str, + handle: str = pytac.RB, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + ) -> float: """Get the value for a field. Returns the value of a field on the element. This value is uniquely @@ -241,15 +257,15 @@ def get_value( real or simulated values. Args: - field (str): The requested field. - handle (str): pytac.SP or pytac.RB. - units (str): pytac.ENG or pytac.PHYS returned. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + field: The requested field. + handle: pytac.SP or pytac.RB. + units: pytac.ENG or pytac.PHYS returned. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - float: The value of the requested field + The value of the requested field Raises: DataSourceException: if there is no data source on the given field. @@ -266,23 +282,23 @@ def get_value( def set_value( self, - field, - value, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - ): + field: str, + value: float, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + ) -> None: """Set the value for a field. This value can be set on the machine or the simulation. Args: - field (str): The requested field. - value (float): The value to set. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + field: The requested field. + value: The value to set. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: DataSourceException: if arguments are incorrect. @@ -295,12 +311,12 @@ def set_value( except FieldException as e: raise FieldException(f"{self}: {e}") - def set_lattice(self, lattice): + def set_lattice(self, lattice: Lattice) -> None: """Set the stored lattice reference for this element to the passed lattice object. Args: - lattice (Lattice): lattice object to store a reference to. + lattice: lattice object to store a reference to. """ self._lattice = lattice @@ -309,19 +325,17 @@ class EpicsElement(Element): """EPICS-aware element. Adds get_pv_name() method. - - **Methods:** """ - def get_pv_name(self, field, handle): + def get_pv_name(self, field: str, handle: str) -> str: """Get PV name for the specified field and handle. Args: - field (str): The requested field. - handle (str): pytac.RB or pytac.SP. + field: The requested field. + handle: pytac.RB or pytac.SP. Returns: - str: The readback or setpoint PV for the specified field. + The readback or setpoint PV for the specified field. Raises: DataSourceException: if there is no data source for this field. diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 470ee1cc..3efdc45d 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -2,7 +2,17 @@ machine. """ import logging -from typing import List, Optional +from typing import ( + TYPE_CHECKING, + Any, + Dict, + Iterable, + List, + Optional, + Sequence, + Set, + Union, +) import numpy @@ -11,6 +21,11 @@ from pytac.element import Element from pytac.exceptions import DataSourceException, UnitsException +if TYPE_CHECKING: + from pytac.cs import ControlSystem + from pytac.device import Device + from pytac.units import UnitConv + class Lattice: """Representation of a lattice. @@ -18,29 +33,32 @@ class Lattice: Represents a lattice object that contains all elements of the ring. It has a name and a control system to be used for unit conversion. - **Attributes:** - Attributes: name: The name of the lattice. symmetry: The symmetry of the lattice (the number of cells). .. Private Attributes: - _elements: The list of all the element objects in the lattice - _data_source_manager: A class that manages the - data sources associated - with this lattice. + _elements: The list of all the element objects in the lattice + _data_source_manager: A class that manages the data sources associated + with this lattice. """ + name: str + symmetry: Optional[int] + + _elements: List[Element] + _data_source_manager: DataSourceManager + def __init__(self, name: str, symmetry: Optional[int] = None) -> None: - """Args: + """Initialise the Lattice object. + + Args: name: The name of the lattice. symmetry: The symmetry of the lattice (the number of cells). - - **Methods:** """ self.name = name self.symmetry = symmetry - self._elements: List[Element] = [] + self._elements = [] self._data_source_manager = DataSourceManager() def __str__(self) -> str: @@ -85,10 +103,10 @@ def __getitem__(self, n: int) -> Element: i.e. index 0 represents the first element in the lattice. Args: - n: index. + n: Index. Returns: - indexed element + Indexed element. """ return self._elements[n] @@ -105,23 +123,23 @@ def set_data_source(self, data_source: DataSource, data_source_type: str) -> Non Args: data_source: the data source to be set. - data_source_type: the type of the data source being set: - pytac.LIVE or pytac.SIM. + data_source_type: the type of the data source being set; + pytac.LIVE or pytac.SIM. """ self._data_source_manager.set_data_source(data_source, data_source_type) - def get_fields(self): + def get_fields(self) -> Dict[str, Iterable]: """Get the fields defined on the lattice. Includes all fields defined by all data sources. Returns: - dict: A dictionary of all the fields defined on the lattice, - separated by data source(key). + A dictionary of all the fields defined on the lattice, separated by + data source(key). """ return self._data_source_manager.get_fields() - def add_device(self, field, device, uc): + def add_device(self, field: str, device: Device, uc: UnitConv) -> None: """Add device and unit conversion objects to a given field. A DeviceDataSource must be set before calling this method, this @@ -129,17 +147,16 @@ def add_device(self, field, device, uc): uses devices. Args: - field (str): The key to store the unit conversion and device - objects. - device (Device): The device object used for this field. - uc (UnitConv): The unit conversion object used for this field. + field: The key to store the unit conversion and device objects. + device: The device object used for this field. + uc: The unit conversion object used for this field. Raises: DataSourceException: if no DeviceDataSource is set. """ self._data_source_manager.add_device(field, device, uc) - def get_device(self, field): + def get_device(self, field: str) -> Device: """Get the device for the given field. A DeviceDataSource must be set before calling this method, this @@ -147,47 +164,47 @@ def get_device(self, field): uses devices. Args: - field (str): The lookup key to find the device on the lattice. + field: The lookup key to find the device on the lattice. Returns: - Device: The device on the given field. + The device on the given field. Raises: DataSourceException: if no DeviceDataSource is set. """ return self._data_source_manager.get_device(field) - def get_unitconv(self, field): + def get_unitconv(self, field: str) -> UnitConv: """Get the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. + field: The field associated with this conversion. Returns: - UnitConv: The object associated with the specified field. + The object associated with the specified field. Raises: FieldException: if no unit conversion object is present. """ return self._data_source_manager.get_unitconv(field) - def set_unitconv(self, field, uc): + def set_unitconv(self, field: str, uc: UnitConv) -> None: """Set the unit conversion option for the specified field. Args: - field (str): The field associated with this conversion. - uc (UnitConv): The unit conversion object to be set. + field: The field associated with this conversion. + uc: The unit conversion object to be set. """ self._data_source_manager.set_unitconv(field, uc) def get_value( self, - field, - handle=pytac.RB, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - ): + field: str, + handle: str = pytac.RB, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + ) -> float: """Get the value for a field on the lattice. Returns the value of a field on the lattice. This value is uniquely @@ -196,15 +213,15 @@ def get_value( real or simulated values. Args: - field (str): The requested field. - handle (str): pytac.SP or pytac.RB. - units (str): pytac.ENG or pytac.PHYS returned. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException; if - False, return None and log a warning. + field: The requested field. + handle: pytac.SP or pytac.RB. + units: pytac.ENG or pytac.PHYS returned. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Returns: - float: The value of the requested field + The value of the requested field Raises: DataSourceException: if there is no data source on the given field. @@ -216,23 +233,23 @@ def get_value( def set_value( self, - field, - value, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, + field: str, + value: float, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, ): """Set the value for a field. This value can be set on the machine or the simulation. Args: - field (str): The requested field. - value (float): The value to set. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + field: The requested field. + value: The value to set. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: DataSourceException: if arguments are incorrect. @@ -240,42 +257,43 @@ def set_value( """ self._data_source_manager.set_value(field, value, units, data_source, throw) - def get_length(self): + def get_length(self) -> float: """Returns the length of the lattice, in meters. Returns: - float: The length of the lattice (m). + The length of the lattice (m). """ total_length = 0.0 for e in self._elements: total_length += e.length return total_length - def add_element(self, element): + def add_element(self, element: Element) -> None: """Append an element to the lattice and update its lattice reference. Args: - element (Element): element to append. + element: element to append. """ element.set_lattice(self) self._elements.append(element) - def get_elements(self, family=None, cell=None): + def get_elements( + self, family: Optional[str] = None, cell: Optional[int] = None + ) -> List[Element]: """Get the elements of a family from the lattice. If no family is specified it returns all elements. Elements are returned in the order they exist in the ring. Args: - family (str): requested family. - cell (int): restrict elements to those in the specified cell. + family: requested family. + cell: restrict elements to those in the specified cell. Returns: - list: list containing all elements of the specified family. + List containing all elements of the specified family. Raises: - ValueError: if there are no elements in the specified cell or - family. + ValueError: if there are no elements in the specified cell or family. """ if family is None: elements = self._elements[:] @@ -291,7 +309,7 @@ def get_elements(self, family=None, cell=None): raise ValueError(f"{self}: no elements in cell {cell}.") return elements - def get_all_families(self): + def get_all_families(self) -> Set[Any]: """Get all families of elements in the lattice. Returns: @@ -302,22 +320,23 @@ def get_all_families(self): families.update(element.families) return families - def get_family_s(self, family): + def get_family_s(self, family: str) -> List[float]: """Get s positions for all elements from the same family. Args: - family (str): requested family. + family: requested family. Returns: - list: list of s positions for each element. + List of s positions for each element. """ elements = self.get_elements(family) s_positions = [] for element in elements: + assert element.s is not None s_positions.append(element.s) return s_positions - def get_element_devices(self, family, field): + def get_element_devices(self, family: str, field: str) -> List[Device]: """Get devices for a specific field for elements in the specfied family. @@ -326,11 +345,11 @@ def get_element_devices(self, family, field): and 'y'. Args: - family (str): family of elements. - field (str): field specifying the devices. + family: family of elements. + field: field specifying the devices. Returns: - list: devices for specified family and field. + Devices for specified family and field. """ elements = self.get_elements(family) devices = [] @@ -341,7 +360,7 @@ def get_element_devices(self, family, field): logging.warning(f"No device for field {field} on element {element}.") return devices - def get_element_device_names(self, family, field): + def get_element_device_names(self, family: str, field: str) -> List[str]: """Get the names for devices attached to a specific field for elements in the specfied family. @@ -350,8 +369,8 @@ def get_element_device_names(self, family, field): and 'y'. Args: - family (str): family of elements. - field (str): field specifying the devices. + family: family of elements. + field: field specifying the devices. Returns: list: device names for specified family and field. @@ -361,31 +380,30 @@ def get_element_device_names(self, family, field): def get_element_values( self, - family, - field, - handle=pytac.RB, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - dtype=None, - ): + family: str, + field: str, + handle: str = pytac.RB, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + dtype: Optional[numpy.dtype] = None, + ) -> Union[list, numpy.ndarray]: """Get the value of the given field for all elements in the given family in the lattice. Args: - family (str): family of elements to request the values of. - field (str): field to request values for. - handle (str): pytac.RB or pytac.SP. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException; if - False, None will be returned for any PV that fails - and a warning will be logged. - dtype (numpy.dtype): if None, return a list. If not None, return a - numpy array of the specified type. + family: family of elements to request the values of. + field: field to request values for. + handle: pytac.RB or pytac.SP. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. + dtype: if None, return a list. If not None, return a numpy array of the + specified type. Returns: - Union(list, numpy.ndarray): The requested values. + The requested values. """ elements = self.get_elements(family) values = [ @@ -398,30 +416,28 @@ def get_element_values( def set_element_values( self, - family, - field, - values, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - ): + family: str, + field: str, + values: Sequence[Any], + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + ) -> None: """Set the value of the given field for all elements in the given family in the lattice to the given values. Args: - family (str): family of elements on which to set values. - field (str): field to set values for. - values (typing.Sequence): A list of values to assign. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure, if True raise ControlSystemException, if - False return a list of True and False values - corresponding to successes and failures and log a - warning for each PV that fails. + family: family of elements on which to set values. + field: field to set values for. + values: A list of values to assign. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: IndexError: if the given list of values doesn't match the number of - elements in the family. + elements in the family. """ elements = self.get_elements(family) if len(elements) != len(values): @@ -444,12 +460,11 @@ def set_default_units(self, units: str) -> None: """Sets the default unit type for the lattice and all its elements. Args: - default_units: The default unit type to be set across the - entire lattice, pytac.ENG or pytac.PHYS. + default_units: The default unit type to be set across the entire lattice, + pytac.ENG or pytac.PHYS. Raises: - UnitsException: if specified default unit type is not a valid unit - type. + UnitsException: if specified default unit type is not a valid unit type. """ if units == pytac.ENG or units == pytac.PHYS: self._data_source_manager.default_units = units @@ -465,12 +480,12 @@ def set_default_data_source(self, data_source_type: str) -> None: """Sets the default data source for the lattice and all its elements. Args: - data_source_type: The default data source to be set across the - entire lattice, pytac.LIVE or pytac.SIM. + data_source_type: The default data source to be set across the entire + lattice, pytac.LIVE or pytac.SIM. Raises: - DataSourceException: if specified default data source is not a - valid data source. + DataSourceException: if specified default data source is not a valid + data source. """ if (data_source_type == pytac.LIVE) or (data_source_type == pytac.SIM): self._data_source_manager.default_data_source = data_source_type @@ -483,33 +498,35 @@ def set_default_data_source(self, data_source_type: str) -> None: f"Please enter {pytac.LIVE} or {pytac.SIM}." ) - def get_default_units(self): + def get_default_units(self) -> str: """Get the default unit type, pytac.ENG or pytac.PHYS. Returns: - str: the default unit type for the entire lattice. + The default unit type for the entire lattice. """ return self._data_source_manager.default_units - def get_default_data_source(self): + def get_default_data_source(self) -> str: """Get the default data source, pytac.LIVE or pytac.SIM. Returns: - str: the default data source for the entire lattice. + The default data source for the entire lattice. """ return self._data_source_manager.default_data_source - def convert_family_values(self, family, field, values, origin, target): + def convert_family_values( + self, family: str, field: str, values: Sequence[Any], origin: str, target: str + ) -> List[float]: """Convert the given values according to the given origin and target units, using the unit conversion objects for the given field on the elements in the given family. Args: - family (str): the family of elements which the values belong to. - field (str): the field on the elements which the values are from. - values (typing.Sequence): values to be converted. - origin (str): pytac.ENG or pytac.PHYS. - target (str): pytac.ENG or pytac.PHYS. + family: the family of elements which the values belong to. + field: the field on the elements which the values are from. + values: values to be converted. + origin: pytac.ENG or pytac.PHYS. + target: pytac.ENG or pytac.PHYS. """ elements = self.get_elements(family) if len(elements) != len(values): @@ -517,7 +534,7 @@ def convert_family_values(self, family, field, values, origin, target): f"Number of elements in given sequence({len(values)}) must " f"be equal to the number of elements in the family({len(elements)})." ) - converted_values = [] + converted_values: List[float] = [] for elem, value in zip(elements, values): uc = elem.get_unitconv(field) converted_values.append(uc.convert(value, origin, target)) @@ -530,43 +547,41 @@ class EpicsLattice(Lattice): Allows efficient get_element_values() and set_element_values() methods, and adds get_pv_names() method. - **Attributes:** - Attributes: - name (str): The name of the lattice. - symmetry (int): The symmetry of the lattice (the number of cells). + name: The name of the lattice. + symmetry: The symmetry of the lattice (the number of cells). .. Private Attributes: - _elements (list): The list of all the element objects in the lattice - _cs (ControlSystem): The control system to use for the more - efficient batch getting and setting of PVs. - _data_source_manager (DataSourceManager): A class that manages the - data sources associated - with this lattice. + _elements: The list of all the element objects in the lattice + _cs: The control system to use for the more efficient batch getting and + setting of PVs. + _data_source_manager: A class that manages the data sources associated + with this lattice. """ - def __init__(self, name, epics_cs, symmetry=None): + _cs: ControlSystem + + def __init__( + self, name: str, epics_cs: ControlSystem, symmetry: Optional[int] = None + ) -> None: """ Args: - name (str): The name of the epics lattice. - epics_cs (ControlSystem): The control system used to store the - values on a PV. - symmetry (int): The symmetry of the lattice (the number of cells). - - **Methods:** + name: The name of the epics lattice. + epics_cs: The control system used to store the values on a PV. + symmetry: The symmetry of the lattice (the number of cells). """ super(EpicsLattice, self).__init__(name, symmetry) self._cs = epics_cs - def get_pv_name(self, field, handle): + def get_pv_name(self, field: str, handle: str) -> str: """Get the PV name for a specific field, and handle on this lattice. Args: - field (str): The requested field. - handle (str): pytac.RB or pytac.SP. + field: The requested field. + handle: pytac.RB or pytac.SP. Returns: - str: The readback or setpoint PV for the specified field. + The readback or setpoint PV for the specified field. """ try: return ( @@ -580,20 +595,19 @@ def get_pv_name(self, field, handle): f"{self}, as the device does not have associated PVs." ) - def get_element_pv_names(self, family, field, handle): + def get_element_pv_names(self, family: str, field: str, handle: str) -> List[str]: """Get the PV names for the given field, and handle, on all elements in the given family in the lattice. - Assume that the elements are EpicsElements that have the get_pv_name() - method. + Assume that the elements are EpicsElements that have the get_pv_name() method. Args: - family (str): The requested family. - field (str): The requested field. - handle (str): pytac.RB or pytac.SP. + family: The requested family. + field: The requested field. + handle: pytac.RB or pytac.SP. Returns: - list: A list of PV names, strings. + A list of PV names, strings. """ elements = self.get_elements(family) pv_names = [] @@ -603,31 +617,30 @@ def get_element_pv_names(self, family, field, handle): def get_element_values( self, - family, - field, - handle=pytac.RB, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - dtype=None, - ): + family: str, + field: str, + handle: str = pytac.RB, + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + dtype: Optional[numpy.dtype] = None, + ) -> Union[list, numpy.ndarray]: """Get the value of the given field for all elements in the given family in the lattice. Args: - family (str): family of elements to request the values of. - field (str): field to request values for. - handle (str): pytac.RB or pytac.SP. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException; if - False, None will be returned for any PV that fails - and a warning will be logged. - dtype (numpy.dtype): if None, return a list. If not None, return a - numpy array of the specified type. + family: family of elements to request the values of. + field: field to request values for. + handle: pytac.RB or pytac.SP. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. + dtype: if None, return a list. If not None, return a numpy array of the + specified type. Returns: - Union(list, numpy.ndarray): The requested values. + The requested values. """ if data_source == pytac.DEFAULT: data_source = self.get_default_data_source() @@ -650,28 +663,28 @@ def get_element_values( def set_element_values( self, - family, - field, - values, - units=pytac.DEFAULT, - data_source=pytac.DEFAULT, - throw=True, - ): + family: str, + field: str, + values: Sequence[Any], + units: str = pytac.DEFAULT, + data_source: str = pytac.DEFAULT, + throw: bool = True, + ) -> None: """Set the value of the given field for all elements in the given family in the lattice to the given values. Args: - family (str): family of elements on which to set values. - field (str): field to set values for. - values (typing.Sequence): A list of values to assign. - units (str): pytac.ENG or pytac.PHYS. - data_source (str): pytac.LIVE or pytac.SIM. - throw (bool): On failure: if True, raise ControlSystemException: if - False, log a warning. + family: family of elements on which to set values. + field: field to set values for. + values: A list of values to assign. + units: pytac.ENG or pytac.PHYS. + data_source: pytac.LIVE or pytac.SIM. + throw: On failure: if True, raise ControlSystemException; if False, None + will be returned for any PV that fails and a warning will be logged. Raises: IndexError: if the given list of values doesn't match the number of - elements in the family. + elements in the family. """ if data_source == pytac.DEFAULT: data_source = self.get_default_data_source() diff --git a/src/pytac/load_csv.py b/src/pytac/load_csv.py index 6c7eba09..b03f00d5 100644 --- a/src/pytac/load_csv.py +++ b/src/pytac/load_csv.py @@ -14,7 +14,7 @@ import copy import csv from pathlib import Path -from typing import Dict, Iterator +from typing import TYPE_CHECKING, Dict, Iterator, Optional, Union import pytac from pytac import data_source, element, utils @@ -23,6 +23,9 @@ from pytac.lattice import EpicsLattice, Lattice from pytac.units import NullUnitConv, PchipUnitConv, PolyUnitConv, UnitConv +if TYPE_CHECKING: + from pytac.cs import ControlSystem + # Create a default unit conversion object that returns the input unchanged. DEFAULT_UC = NullUnitConv() @@ -49,7 +52,7 @@ def load_poly_unitconv(filepath: Path) -> Dict[int, PolyUnitConv]: filepath: The file from which to load. Returns: - dict: A dictionary of the unit conversions. + A dictionary of the unit conversions. """ unitconvs: Dict[int, PolyUnitConv] = {} data = collections.defaultdict(list) @@ -70,7 +73,7 @@ def load_pchip_unitconv(filepath: Path) -> Dict[int, PchipUnitConv]: filename: The file from which to load. Returns: - dict: A dictionary of the unit conversions. + A dictionary of the unit conversions. """ unitconvs: Dict[int, PchipUnitConv] = {} data = collections.defaultdict(list) @@ -144,25 +147,28 @@ def load_unitconv(mode_dir: Path, lattice: Lattice) -> None: element.set_unitconv(item["field"], uc) -def load(mode, control_system=None, directory=None, symmetry=None): +def load( + mode: str, + control_system: Optional[ControlSystem] = None, + directory: Optional[Union[str, Path]] = None, + symmetry: Optional[int] = None, +) -> Lattice: """Load the elements of a lattice from a directory. Args: - mode (str): The name of the mode to be loaded. - control_system (ControlSystem): The control system to be used. If none - is provided an EpicsControlSystem will - be created. - directory (str): Directory where to load the files from. If no - directory is given the data directory at the root of - the repository is used. - symmetry (int): The symmetry of the lattice (the number of cells). + mode: The name of the mode to be loaded. + control_system: The control system to be used. If none is provided an + EpicsControlSystem will be created. + directory: Directory where to load the files from. If no directory is + given the data directory at the root of the repository is used. + symmetry: The symmetry of the lattice (the number of cells). Returns: - Lattice: The lattice containing all elements. + The lattice containing all elements. Raises: ControlSystemException: if the default control system, cothread, is not - installed. + installed. """ try: if control_system is None: diff --git a/src/pytac/units.py b/src/pytac/units.py index 98f89869..fd5bb25f 100644 --- a/src/pytac/units.py +++ b/src/pytac/units.py @@ -1,4 +1,6 @@ """Classes for use in unit conversion.""" +from typing import Any, Callable, List, Optional + import numpy from scipy.interpolate import PchipInterpolator @@ -6,14 +8,14 @@ from pytac.exceptions import UnitsException -def unit_function(value): +def unit_function(value: float) -> float: """Default value for the pre and post functions used in unit conversion. Args: - value (float): The value to be converted. + value: The value to be converted. Returns: - float: The result of the conversion. + The result of the conversion. """ return value @@ -30,100 +32,93 @@ class UnitConv: applied to the result of the initial conversion. One happens after the conversion, the other happens before the conversion back. - **Attributes:** - Attributes: - name (str): An identifier for the unit conversion object. - eng_units (str): The unit type of the post conversion engineering - value. - phys_units (str): The unit type of the post conversion physics value. + name: An identifier for the unit conversion object. + eng_units: The unit type of the post conversion engineering value. + phys_units: The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied - after the initial conversion. - _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied - before the initial conversion. + _post_eng_to_phys: Function to be applied after the initial conversion. + _pre_phys_to_eng: Function to be applied before the initial conversion. """ + name: Optional[str] + eng_units: str + phys_units: str + + _post_eng_to_phys: Callable[[float], float] + _pre_phys_to_eng: Callable[[float], float] + def __init__( self, - post_eng_to_phys=unit_function, - pre_phys_to_eng=unit_function, - engineering_units="", - physics_units="", - name=None, - ): - """ + post_eng_to_phys: Callable[[float], float] = unit_function, + pre_phys_to_eng: Callable[[float], float] = unit_function, + engineering_units: str = "", + physics_units: str = "", + name: Optional[str] = None, + ) -> None: + """Initialise the UnitConv Object. + Args: - post_eng_to_phys (typing.Callable[[float], float]): Function to be applied - after the initial conversion. - pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied - before the initial conversion. - engineering_units (str): The unit type of the post conversion - engineering value. - physics_units (str): The unit type of the post conversion physics - value. - name (str): An identifier for the unit conversion object. - - **Methods:** + post_eng_to_phys: Function to be applied after the initial conversion. + pre_phys_to_eng: Function to be applied before the initial conversion. + engineering_units: The unit type of the post conversion engineering value. + physics_units: The unit type of the post conversion physics value. + name: An identifier for the unit conversion object. """ self.name = name self._post_eng_to_phys = post_eng_to_phys self._pre_phys_to_eng = pre_phys_to_eng self.eng_units = engineering_units self.phys_units = physics_units - self.lower_limit = None - self.upper_limit = None + self.lower_limit: Optional[float] = None + self.upper_limit: Optional[float] = None - def __str__(self): + def __str__(self) -> str: string_rep = self.__class__.__name__ if self.name is not None: string_rep += f" {self.name}" return string_rep - def set_post_eng_to_phys(self, post_eng_to_phys): + def set_post_eng_to_phys(self, post_eng_to_phys: Callable[[float], float]) -> None: """Set the function to be applied after the initial conversion. Args: - post_eng_to_phys (typing.Callable[[float], float]): Function to be applied - after the initial conversion. + post_eng_to_phys: Function to be applied after the initial conversion. """ self._post_eng_to_phys = post_eng_to_phys - def set_pre_phys_to_eng(self, pre_phys_to_eng): + def set_pre_phys_to_eng(self, pre_phys_to_eng: Callable[[float], float]) -> None: """Set the function to be applied before the initial conversion. Args: - pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied - before the initial conversion. + pre_phys_to_eng: Function to be applied before the initial conversion. """ self._pre_phys_to_eng = pre_phys_to_eng - def _raw_eng_to_phys(self, value): + def _raw_eng_to_phys(self, value: float): """Function to be implemented by child classes. Args: - value (float): The engineering value to be converted to physics - units. + value: The engineering value to be converted to physics units. """ raise NotImplementedError(f"{self}: No eng-to-phys conversion provided") - def eng_to_phys(self, value): + def eng_to_phys(self, value: float) -> float: """Function that does the unit conversion. Conversion from engineering to physics units. An additional function may be cast on the initial conversion. Args: - value (float): Value to be converted from engineering to physics - units. + value: Value to be converted from engineering to physics units. Returns: - float: The result value. + The result value. Raises: UnitsException: If the conversion is invalid; i.e. if there are no - solutions, or multiple, within conversion limits. + solutions, or multiple, within conversion limits. """ if self.lower_limit is not None and value < self.lower_limit: raise UnitsException( @@ -149,31 +144,29 @@ def eng_to_phys(self, value): ) return valid_results[0] - def _raw_phys_to_eng(self, value): + def _raw_phys_to_eng(self, value: float): """Function to be implemented by child classes. Args: - value (float): The physics value to be converted to engineering - units. + value: The physics value to be converted to engineering units. """ raise NotImplementedError(f"{self}: No phys-to-eng conversion provided") - def phys_to_eng(self, value): + def phys_to_eng(self, value: float) -> float: """Function that does the unit conversion. Conversion from physics to engineering units. An additional function may be cast on the initial conversion. Args: - value (float): Value to be converted from physics to engineering - units. + value: Value to be converted from physics to engineering units. Returns: - float: The result value. + The result value. Raises: UnitsException: If the conversion is invalid; i.e. if there are no - solutions, or multiple, within conversion limits. + solutions, or multiple, within conversion limits. """ adjusted_value = self._pre_phys_to_eng(value) results = self._raw_phys_to_eng(adjusted_value) @@ -196,21 +189,21 @@ def phys_to_eng(self, value): ) return valid_results[0] - def convert(self, value, origin, target): + def convert(self, value: float, origin: str, target: str) -> float: """Convert between two different unit types and check the validity of the result. Args: - value (float): the value to be converted - origin (str): pytac.ENG or pytac.PHYS - target (str): pytac.ENG or pytac.PHYS + value: the value to be converted + origin: pytac.ENG or pytac.PHYS + target: pytac.ENG or pytac.PHYS Returns: - float: The resulting value. + The resulting value. Raises: UnitsException: If the conversion is invalid; i.e. if there are no - solutions, or multiple, within conversion limits. + solutions, or multiple, within conversion limits. """ if origin == target: return value @@ -223,13 +216,13 @@ def convert(self, value, origin, target): f"{self}: Conversion from {origin} to {target} not understood." ) - def set_conversion_limits(self, lower_limit, upper_limit): + def set_conversion_limits(self, lower_limit: float, upper_limit: float) -> None: """Conversion limits to be applied before or after a conversion take place. Limits should be set in in engineering units. Args: - lower_limit (float): the lower conversion limit - upper_limit (float): the upper conversion limit + lower_limit: the lower conversion limit + upper_limit: the upper conversion limit """ if (lower_limit is not None) and (upper_limit is not None): if lower_limit >= upper_limit: @@ -240,15 +233,15 @@ def set_conversion_limits(self, lower_limit, upper_limit): self.lower_limit = lower_limit self.upper_limit = upper_limit - def get_conversion_limits(self, units=pytac.ENG): + def get_conversion_limits(self, units: str = pytac.ENG) -> List[float]: """Return the current conversion limits in the specified unit type. Args: - units: + units: The unit type. Returns: - list: the conversion limits in the desired unit type, - format: [lower_limit, upper_limit] + The conversion limits in the desired unit type, + format: [lower_limit, upper_limit] """ if units == pytac.ENG: return [self.lower_limit, self.upper_limit] @@ -265,72 +258,65 @@ class PolyUnitConv(UnitConv): """Linear interpolation for converting between physics and engineering units. - **Attributes:** - Attributes: - p (poly1d): A one-dimensional polynomial of coefficients. - name (str): An identifier for the unit conversion object. - eng_units (str): The unit type of the post conversion engineering - value. - phys_units (str): The unit type of the post conversion physics value. + p: A one-dimensional polynomial of coefficients. + name: An identifier for the unit conversion object. + eng_units: The unit type of the post conversion engineering value. + phys_units: The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied - after the initial conversion. - _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied - before the initial conversion. + _post_eng_to_phys: Function to be appliedafter the initial conversion. + _pre_phys_to_eng: Function to be applied before the initial conversion. """ + p: numpy.poly1d + def __init__( self, - coef, - post_eng_to_phys=unit_function, - pre_phys_to_eng=unit_function, - engineering_units="", - physics_units="", - name=None, + coef: numpy.ndarray[Any, numpy.dtype[numpy.generic]], + post_eng_to_phys: Callable[[float], float] = unit_function, + pre_phys_to_eng: Callable[[float], float] = unit_function, + engineering_units: str = "", + physics_units: str = "", + name: Optional[str] = None, ): - """ + """Initialise the PolyUnitConv Object. + Args: - coef (numpy.ndarray[typing.Any, numpy.dtype[numpy.generic]]): The - polynomial's coefficients, in decreasing powers. - post_eng_to_phys (float): The value after conversion between ENG - and PHYS. - pre_eng_to_phys (float): The value before conversion. - engineering_units (str): The unit type of the post conversion - engineering value. - physics_units (str): The unit type of the post conversion physics - value. - name (str): An identifier for the unit conversion object. + coef: The polynomial's coefficients, in decreasing powers. + post_eng_to_phys: The value after conversion between ENG and PHYS. + pre_eng_to_phys: The value before conversion. + engineering_units: The unit type of the post conversion engineering value. + physics_units: The unit type of the post conversion physics value. + name: An identifier for the unit conversion object. """ super(self.__class__, self).__init__( post_eng_to_phys, pre_phys_to_eng, engineering_units, physics_units, name ) self.p = numpy.poly1d(coef) - def _raw_eng_to_phys(self, eng_value): + def _raw_eng_to_phys(self, eng_value: float) -> List[float]: """Convert between engineering and physics units. Args: - eng_value (float): The engineering value to be converted to physics - units. + eng_value: The engineering value to be converted to physics units. Returns: - list: Containing the converted physics value from the given - engineering value. + Containing the converted physics value from the given + engineering value. """ return [self.p(eng_value)] - def _raw_phys_to_eng(self, physics_value): + def _raw_phys_to_eng(self, physics_value: float) -> List[float]: """Convert between physics and engineering units. Args: - physics_value (float): The physics value to be converted to - engineering units. + physics_value: The physics value to be converted to + engineering units. Returns: - list: Containing all posible real engineering values converted - from the given physics value. + Containing all posible real engineering values converted + from the given physics value. """ roots = set((self.p - physics_value).roots) # remove duplicates valid_roots = [] @@ -343,52 +329,45 @@ def _raw_phys_to_eng(self, physics_value): class PchipUnitConv(UnitConv): """Piecewise Cubic Hermite Interpolating Polynomial unit conversion. - **Attributes:** - Attributes: - x (list): A list of points on the x axis. These must be in increasing - order for the interpolation to work. Otherwise, a ValueError - is raised. - y (list): A list of points on the y axis. These must be in increasing - or decreasing order. Otherwise, a ValueError is raised. - pp (PchipInterpolator): A pchip one-dimensional monotonic cubic - interpolation of points on both x and y axes. - - name (str): An identifier for the unit conversion object. - eng_units (str): The unit type of the post conversion engineering - value. - phys_units (str): The unit type of the post conversion physics value. + x: A list of points on the x axis. These must be in increasing order + for the interpolation to work. Otherwise, a ValueError is raised. + y: A list of points on the y axis. These must be in increasing or + decreasing order. Otherwise, a ValueError is raised. + pp: A pchip one-dimensional monotonic cubic interpolation of points on + both x and y axes. + name: An identifier for the unit conversion object. + eng_units: The unit type of the post conversion engineering value. + phys_units: The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (typing.Callable[[float], float]): Function to be applied - after the initial conversion. - _pre_phys_to_eng (typing.Callable[[float], float]): Function to be applied - before the initial conversion. + _post_eng_to_phys: Function to be applied after the initial conversion. + _pre_phys_to_eng: Function to be applied before the initial conversion. """ + x: List[Any] + y: List[Any] + pp: PchipInterpolator + def __init__( self, - x, - y, - post_eng_to_phys=unit_function, - pre_phys_to_eng=unit_function, - engineering_units="", - physics_units="", - name=None, - ): + x: List[Any], + y: List[Any], + post_eng_to_phys: Callable[[float], float] = unit_function, + pre_phys_to_eng: Callable[[float], float] = unit_function, + engineering_units: str = "", + physics_units: str = "", + name: Optional[str] = None, + ) -> None: """ Args: - x (list): A list of points on the x axis. These must be in - increasing order for the interpolation to work. - Otherwise, a ValueError is raised. - y (list): A list of points on the y axis. These must be in - increasing or decreasing order. Otherwise, a ValueError - is raised. - engineering_units (str): The unit type of the post conversion - engineering value. - physics_units (str): The unit type of the post conversion physics - value. - name (str): An identifier for the unit conversion object. + x: A list of points on the x axis. These must be in increasing order for + the interpolation to work. Otherwise, a ValueError is raised. + y: A list of points on the y axis. These must be in increasing or + decreasing order. Otherwise, a ValueError is raised. + engineering_units: The unit type of the post conversion engineering value. + physics_units: The unit type of the post conversion physics value. + name: An identifier for the unit conversion object. Raises: ValueError: if coefficients are not appropriately monotonic. @@ -412,28 +391,25 @@ def __init__( "y coefficients must be monotonically increasing or decreasing." ) - def _raw_eng_to_phys(self, eng_value): + def _raw_eng_to_phys(self, eng_value: float) -> List[Any]: """Convert between engineering and physics units. Args: - eng_value (float): The engineering value to be converted to physics - units. + eng_value: The engineering value to be converted to physics units. Returns: - list: Containing the converted physics value from the given - engineering value. + Containing the converted physics value from the given engineering value. """ return [self.pp(eng_value)] - def _raw_phys_to_eng(self, physics_value): + def _raw_phys_to_eng(self, physics_value: float): """Convert between physics and engineering units. Args: - physics_value (float): The physics value to be converted to - engineering units. + physics_value: The physics value to be converted to engineering units. Returns: - list: Containing all posible real engineering values converted - from the given physics value. + Containing all posible real engineering values converted from the given + physics value. """ y = [val - physics_value for val in self.y] new_pp = PchipInterpolator(self.x, y) @@ -448,55 +424,50 @@ def _raw_phys_to_eng(self, physics_value): class NullUnitConv(UnitConv): """Returns input value without performing any conversions. - **Attributes:** - Attributes: - eng_units (str): The unit type of the post conversion engineering - value. - phys_units (str): The unit type of the post conversion physics value. + eng_units: The unit type of the post conversion engineering value. + phys_units: The unit type of the post conversion physics value. .. Private Attributes: - _post_eng_to_phys (typing.Callable[[float], float]): Always unit_function as - no conversion is performed. - _pre_phys_to_eng (typing.Callable[[float], float]): Always unit_function as - no conversion is performed. + _post_eng_to_phys: Always unit_function as no conversion is performed. + _pre_phys_to_eng: Always unit_function as no conversion is performed. """ - def __init__(self, engineering_units="", physics_units=""): - """ + def __init__(self, engineering_units: str = "", physics_units: str = "") -> None: + """Initialise the NullUnitConv Object. + Args: - engineering_units (str): The unit type of the post conversion - engineering value. - physics_units (str): The unit type of the post conversion physics - value. + engineering_units: The unit type of the post conversion engineering value. + physics_units: The unit type of the post conversion physics value. """ super(self.__class__, self).__init__( unit_function, unit_function, engineering_units, physics_units ) - def _raw_eng_to_phys(self, eng_value): + def _raw_eng_to_phys(self, eng_value: float) -> List[Any]: """Doesn't convert between engineering and physics units. Maintains the same syntax as the other UnitConv classes for compatibility, but does not perform any conversion. Args: - eng_value (float): The engineering value to be returned unchanged. + eng_value: The engineering value to be returned unchanged. + Returns: - list: Containing the unconverted given engineering value. + Containing the unconverted given engineering value. """ return [eng_value] - def _raw_phys_to_eng(self, phys_value): + def _raw_phys_to_eng(self, phys_value: float) -> List[Any]: """Doesn't convert between physics and engineering units. Maintains the same syntax as the other UnitConv classes for compatibility, but does not perform any conversion. Args: - physics_value (float): The physics value to be returned unchanged. + physics_value: The physics value to be returned unchanged. Returns: - list: Containing the unconverted given physics value. + Containing the unconverted given physics value. """ return [phys_value] diff --git a/src/pytac/utils.py b/src/pytac/utils.py index 57240236..192fa60b 100644 --- a/src/pytac/utils.py +++ b/src/pytac/utils.py @@ -1,5 +1,6 @@ """Utility functions.""" import math +from typing import Callable import scipy.constants @@ -7,13 +8,14 @@ electron_mass_mev, _, _ = scipy.constants.physical_constants[electron_mass_name] -def get_rigidity(energy_mev): - """ +def get_rigidity(energy_mev: int) -> float: + """Get rigidity function. + Args: - energy_mev (int): the energy of the lattice. + energy_mev: the energy of the lattice. Returns: - float: p devided by the elementary charge. + p devided by the elementary charge. """ gamma = energy_mev / electron_mass_mev beta = math.sqrt(1 - gamma ** (-2)) @@ -22,13 +24,14 @@ def get_rigidity(energy_mev): return p / scipy.constants.e -def get_div_rigidity(energy): - """ +def get_div_rigidity(energy: int) -> Callable[[int], float]: + """Return the function div_rigidity. + Args: - energy (int): the energy of the lattice. + energy: the energy of the lattice. Returns: - typing.Callable[[int], float]: div rigidity. + div rigidity. """ rigidity = get_rigidity(energy) @@ -38,13 +41,14 @@ def div_rigidity(value): return div_rigidity -def get_mult_rigidity(energy): - """ +def get_mult_rigidity(energy: int) -> Callable[[int], float]: + """Return the function mult_rigidity. + Args: - energy (int): the energy of the lattice. + energy: the energy of the lattice. Returns: - typing.Callable[[int], float]: mult rigidity. + mult rigidity. """ rigidity = get_rigidity(energy) From 033297475407c5ce33a76af1752fc78e7263bb5d Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Tue, 29 Aug 2023 15:40:33 +0000 Subject: [PATCH 3/8] Fix circular import --- docs/conf.py | 1 + src/pytac/element.py | 2 +- src/pytac/lattice.py | 24 ++++++------------------ src/pytac/load_csv.py | 8 +++----- src/pytac/utils.py | 6 +++--- 5 files changed, 14 insertions(+), 27 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index a58b0060..9c6b871f 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -105,6 +105,7 @@ intersphinx_mapping = { "python": ("https://docs.python.org/3/", None), "numpy": ("https://numpy.org/doc/stable/", None), + "scipy": ("https://docs.scipy.org/doc/scipy/", None), } # A dictionary of graphviz graph attributes for inheritance diagrams. diff --git a/src/pytac/element.py b/src/pytac/element.py index cb06d370..b4dba368 100644 --- a/src/pytac/element.py +++ b/src/pytac/element.py @@ -311,7 +311,7 @@ def set_value( except FieldException as e: raise FieldException(f"{self}: {e}") - def set_lattice(self, lattice: Lattice) -> None: + def set_lattice(self, lattice: "Lattice") -> None: """Set the stored lattice reference for this element to the passed lattice object. diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 3efdc45d..5c3f088b 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -2,29 +2,17 @@ machine. """ import logging -from typing import ( - TYPE_CHECKING, - Any, - Dict, - Iterable, - List, - Optional, - Sequence, - Set, - Union, -) +from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Union import numpy import pytac +from pytac.cs import ControlSystem from pytac.data_source import DataSource, DataSourceManager +from pytac.device import Device, EpicsDevice from pytac.element import Element from pytac.exceptions import DataSourceException, UnitsException - -if TYPE_CHECKING: - from pytac.cs import ControlSystem - from pytac.device import Device - from pytac.units import UnitConv +from pytac.units import UnitConv class Lattice: @@ -336,7 +324,7 @@ def get_family_s(self, family: str) -> List[float]: s_positions.append(element.s) return s_positions - def get_element_devices(self, family: str, field: str) -> List[Device]: + def get_element_devices(self, family: str, field: str) -> List[EpicsDevice]: """Get devices for a specific field for elements in the specfied family. @@ -387,7 +375,7 @@ def get_element_values( data_source: str = pytac.DEFAULT, throw: bool = True, dtype: Optional[numpy.dtype] = None, - ) -> Union[list, numpy.ndarray]: + ) -> Union[List[float], numpy.ndarray]: """Get the value of the given field for all elements in the given family in the lattice. diff --git a/src/pytac/load_csv.py b/src/pytac/load_csv.py index b03f00d5..80b46f7b 100644 --- a/src/pytac/load_csv.py +++ b/src/pytac/load_csv.py @@ -14,18 +14,16 @@ import copy import csv from pathlib import Path -from typing import TYPE_CHECKING, Dict, Iterator, Optional, Union +from typing import Dict, Iterator, Optional import pytac from pytac import data_source, element, utils +from pytac.cs import ControlSystem from pytac.device import EpicsDevice, SimpleDevice from pytac.exceptions import ControlSystemException from pytac.lattice import EpicsLattice, Lattice from pytac.units import NullUnitConv, PchipUnitConv, PolyUnitConv, UnitConv -if TYPE_CHECKING: - from pytac.cs import ControlSystem - # Create a default unit conversion object that returns the input unchanged. DEFAULT_UC = NullUnitConv() @@ -150,7 +148,7 @@ def load_unitconv(mode_dir: Path, lattice: Lattice) -> None: def load( mode: str, control_system: Optional[ControlSystem] = None, - directory: Optional[Union[str, Path]] = None, + directory: Optional[Path] = None, symmetry: Optional[int] = None, ) -> Lattice: """Load the elements of a lattice from a directory. diff --git a/src/pytac/utils.py b/src/pytac/utils.py index 192fa60b..331c3be3 100644 --- a/src/pytac/utils.py +++ b/src/pytac/utils.py @@ -8,7 +8,7 @@ electron_mass_mev, _, _ = scipy.constants.physical_constants[electron_mass_name] -def get_rigidity(energy_mev: int) -> float: +def get_rigidity(energy_mev: float) -> float: """Get rigidity function. Args: @@ -24,7 +24,7 @@ def get_rigidity(energy_mev: int) -> float: return p / scipy.constants.e -def get_div_rigidity(energy: int) -> Callable[[int], float]: +def get_div_rigidity(energy: float) -> Callable[[float], float]: """Return the function div_rigidity. Args: @@ -41,7 +41,7 @@ def div_rigidity(value): return div_rigidity -def get_mult_rigidity(energy: int) -> Callable[[int], float]: +def get_mult_rigidity(energy: float) -> Callable[[float], float]: """Return the function mult_rigidity. Args: From 5b933226291a992fa817e3336239774728a23a49 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Tue, 29 Aug 2023 15:58:34 +0000 Subject: [PATCH 4/8] Fix attribute declaration for sphinx --- src/pytac/data_source.py | 31 +++++++-------------- src/pytac/device.py | 31 ++++++++------------- src/pytac/element.py | 22 ++++++--------- src/pytac/lattice.py | 13 +++------ src/pytac/units.py | 60 ++++++++++------------------------------ 5 files changed, 47 insertions(+), 110 deletions(-) diff --git a/src/pytac/data_source.py b/src/pytac/data_source.py index 0ed0050d..59c540e5 100644 --- a/src/pytac/data_source.py +++ b/src/pytac/data_source.py @@ -14,12 +14,10 @@ class DataSource(object): Typically an instance would represent hardware via a control system, or a simulation. - - Attributes: - units: pytac.PHYS or pytac.ENG. """ units: str + """Units of DataSource in pytac.PHYS or pytac.ENG.""" def get_fields(self) -> Iterable: """Get all the fields represented by this data source. @@ -64,23 +62,19 @@ class DataSourceManager(object): It receives requests from a lattice or element object and directs them to the correct data source. The unit conversion objects for all fields are also held here. - - Attributes: - default_units: Holds the current default unit type, pytac.PHYS or pytac.ENG, - for an element or lattice. - default_data_source: Holds the current default data source, pytac.LIVE or - pytac.SIM, for an element or lattice. - - .. Private Attributes: - _data_sources: A dictionary of the data sources held. - _uc: A dictionary of the unit conversion objects for each key(field). """ default_units: str + """Holds the current default unit type, pytac.PHYS or pytac.ENG, for an element or + lattice.""" default_data_source: str + """Holds the current default data source, pytac.LIVE or pytac.SIM, for an element + or lattice.""" _data_sources: Dict[str, DataSource] + """A dictionary of the data sources held.""" _uc: Dict[str, UnitConv] + """A dictionary of the unit conversion objects for each key(field).""" def __init__(self) -> None: self.default_units = pytac.ENG @@ -272,18 +266,13 @@ def set_value( class DeviceDataSource(DataSource): - """Data source containing control system devices. - - Attributes: - units: pytac.ENG or pytac.PHYS, pytac.ENG by default. - - .. Private Attributes: - _devices: A dictionary of the devices for each key(field). - """ + """Data source containing control system devices.""" units: str + """pytac.ENG or pytac.PHYS, pytac.ENG by default.""" _devices: Dict[str, Device] + """A dictionary of the devices for each key(field).""" def __init__(self): self._devices = {} diff --git a/src/pytac/device.py b/src/pytac/device.py index a1af5892..0623c6d3 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -62,16 +62,14 @@ class SimpleDevice(Device): with a simulator. In short this device acts as simple storage for data that rarely changes, as it is not affected by changes to other aspects of the accelerator. - - Attributes: - _value: The value of the device. May be a number or list of numbers. - _enabled: Whether the device is enabled. May be a PvEnabler Object. - _readonly: Whether the value may be changed. """ _value: Union[float, int, List[float], List[int]] + """The value of the device. May be a number or list of numbers.""" _enabled: Union[bool, "PvEnabler"] + """Whether the device is enabled. May be a PvEnabler Object.""" _readonly: bool + """Whether the value may be changed.""" def __init__( self, @@ -137,23 +135,19 @@ class EpicsDevice(Device): Contains a control system, readback and setpoint PVs. A readback or setpoint PV is required when creating an epics device otherwise a DataSourceException is raised. The device is enabled by default. - - Attributes: - name: The prefix of EPICS PVs for this device. - rb_pv: The EPICS readback PV. - sp_pv: The EPICS setpoint PV. - - .. Private Attributes: - _cs: The control system object used to get and set the value of a PV. - _enabled: Whether the device is enabled. May be a PvEnabler object. """ name: str + """The prefix of EPICS PVs for this device.""" rb_pv: Optional[str] + """The EPICS readback PV.""" sp_pv: Optional[str] + """The EPICS setpoint PV.""" _cs: ControlSystem + """The control system object used to get and set the value of a PV.""" _enabled: Union[bool, "PvEnabler"] + """Whether the device is enabled. May be a PvEnabler object.""" def __init__( self, @@ -249,17 +243,14 @@ class PvEnabler(object): The class will behave like True if the PV value equals enabled_value, and False otherwise. - - .. Private Attributes: - _pv: The PV name. - _enabled_value: The value for PV for which the device should be - considered enabled. - _cs: The control system object. """ _pv: str + """The PV name.""" _enabled_value: str + """The value for PV for which the device should be considered enabled.""" _cs: ControlSystem + """The control system object.""" def __init__(self, pv: str, enabled_value: str, cs: ControlSystem): """ diff --git a/src/pytac/element.py b/src/pytac/element.py index b4dba368..a4a2f70e 100644 --- a/src/pytac/element.py +++ b/src/pytac/element.py @@ -16,29 +16,23 @@ class Element(object): An element has zero or more devices (e.g. quadrupole magnet) associated with each of its fields (e.g. 'b1' for a quadrupole). - - Attributes: - name: The name identifying the element. The user is free to define this for - their own purposes. - type_: The type of the element. The user is free to define this for their own - purposes. - length: The length of the element in metres. - - .. Private Attributes: - _lattice: The lattice to which the element belongs. - _data_source_manager: A class that manages the data sources associated with - this element. - _families The families this element is a member of, stored as lowercase - strings. """ name: Optional[str] + """The name identifying the element. The user is free to define this for their + own purposes.""" type_: str + """The type of the element. The user is free to define this for their own + purposes.""" length: float + """The length of the element in metres.""" _lattice: Optional["Lattice"] + """The lattice to which the element belongs.""" _data_source_manager: DataSourceManager + """A class that manages the data sources associated with this element.""" _families: Set[Any] + """The families this element is a member of, stored as lowercase strings.""" def __init__( self, diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 5c3f088b..1e4186ce 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -20,22 +20,17 @@ class Lattice: Represents a lattice object that contains all elements of the ring. It has a name and a control system to be used for unit conversion. - - Attributes: - name: The name of the lattice. - symmetry: The symmetry of the lattice (the number of cells). - - .. Private Attributes: - _elements: The list of all the element objects in the lattice - _data_source_manager: A class that manages the data sources associated - with this lattice. """ name: str + """The name of the lattice.""" symmetry: Optional[int] + """The symmetry of the lattice (the number of cells).""" _elements: List[Element] + """The list of all the element objects in the lattice""" _data_source_manager: DataSourceManager + """A class that manages the data sources associated with this lattice.""" def __init__(self, name: str, symmetry: Optional[int] = None) -> None: """Initialise the Lattice object. diff --git a/src/pytac/units.py b/src/pytac/units.py index fd5bb25f..b830edf6 100644 --- a/src/pytac/units.py +++ b/src/pytac/units.py @@ -31,23 +31,19 @@ class UnitConv: The two arguments to this function represent functions that are applied to the result of the initial conversion. One happens after the conversion, the other happens before the conversion back. - - Attributes: - name: An identifier for the unit conversion object. - eng_units: The unit type of the post conversion engineering value. - phys_units: The unit type of the post conversion physics value. - - .. Private Attributes: - _post_eng_to_phys: Function to be applied after the initial conversion. - _pre_phys_to_eng: Function to be applied before the initial conversion. """ name: Optional[str] + """An identifier for the unit conversion object.""" eng_units: str + """The unit type of the post conversion engineering value.""" phys_units: str + """The unit type of the post conversion physics value.""" _post_eng_to_phys: Callable[[float], float] + """Function to be applied after the initial conversion.""" _pre_phys_to_eng: Callable[[float], float] + """Function to be applied before the initial conversion.""" def __init__( self, @@ -257,19 +253,10 @@ def get_conversion_limits(self, units: str = pytac.ENG) -> List[float]: class PolyUnitConv(UnitConv): """Linear interpolation for converting between physics and engineering units. - - Attributes: - p: A one-dimensional polynomial of coefficients. - name: An identifier for the unit conversion object. - eng_units: The unit type of the post conversion engineering value. - phys_units: The unit type of the post conversion physics value. - - .. Private Attributes: - _post_eng_to_phys: Function to be appliedafter the initial conversion. - _pre_phys_to_eng: Function to be applied before the initial conversion. """ p: numpy.poly1d + """A one-dimensional polynomial of coefficients.""" def __init__( self, @@ -327,27 +314,17 @@ def _raw_phys_to_eng(self, physics_value: float) -> List[float]: class PchipUnitConv(UnitConv): - """Piecewise Cubic Hermite Interpolating Polynomial unit conversion. - - Attributes: - x: A list of points on the x axis. These must be in increasing order - for the interpolation to work. Otherwise, a ValueError is raised. - y: A list of points on the y axis. These must be in increasing or - decreasing order. Otherwise, a ValueError is raised. - pp: A pchip one-dimensional monotonic cubic interpolation of points on - both x and y axes. - name: An identifier for the unit conversion object. - eng_units: The unit type of the post conversion engineering value. - phys_units: The unit type of the post conversion physics value. - - .. Private Attributes: - _post_eng_to_phys: Function to be applied after the initial conversion. - _pre_phys_to_eng: Function to be applied before the initial conversion. - """ + """Piecewise Cubic Hermite Interpolating Polynomial unit conversion.""" x: List[Any] + """A list of points on the x axis. These must be in increasing order for the + interpolation to work. Otherwise, a ValueError is raised.""" y: List[Any] + """A list of points on the y axis. These must be in increasing or decreasing + order. Otherwise, a ValueError is raised.""" pp: PchipInterpolator + """A pchip one-dimensional monotonic cubic interpolation of points on both x + and y axes.""" def __init__( self, @@ -422,16 +399,7 @@ def _raw_phys_to_eng(self, physics_value: float): class NullUnitConv(UnitConv): - """Returns input value without performing any conversions. - - Attributes: - eng_units: The unit type of the post conversion engineering value. - phys_units: The unit type of the post conversion physics value. - - .. Private Attributes: - _post_eng_to_phys: Always unit_function as no conversion is performed. - _pre_phys_to_eng: Always unit_function as no conversion is performed. - """ + """Returns input value without performing any conversions.""" def __init__(self, engineering_units: str = "", physics_units: str = "") -> None: """Initialise the NullUnitConv Object. From a9c168cab3011f9c6f6dcccfdaece99697ac2c98 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Wed, 30 Aug 2023 15:20:17 +0000 Subject: [PATCH 5/8] Adding AugmentedType and AugmentedValue --- docs/conf.py | 3 ++ src/pytac/cothread_cs.py | 30 +++++++++++++---- src/pytac/cs.py | 26 ++++++++++++--- src/pytac/data_source.py | 25 ++++++++------ src/pytac/device.py | 39 +++++++++++----------- src/pytac/element.py | 24 ++++++++++---- src/pytac/lattice.py | 72 +++++++++++++++++++++++++--------------- src/pytac/units.py | 62 ++++++++++++++++++++++++---------- src/pytac/utils.py | 21 ++++++++---- 9 files changed, 204 insertions(+), 98 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index 9c6b871f..448c8a3f 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -67,6 +67,9 @@ # Added, but custom for pytac. ("py:class", "poly1d"), ("py:class", "PchipInterpolator"), + ("py:class", "numpy._typing._dtype_like._SupportsDType"), + ("py:class", "numpy._typing._dtype_like._DTypeDict"), + ("py:class", "numpy._typing._array_like._ScalarType_co"), ] # Both the class’ and the __init__ method’s docstring are concatenated and diff --git a/src/pytac/cothread_cs.py b/src/pytac/cothread_cs.py index 9b78deb6..d5f978fc 100644 --- a/src/pytac/cothread_cs.py +++ b/src/pytac/cothread_cs.py @@ -3,10 +3,21 @@ from cothread.catools import ca_nothing, caget, caput -from pytac.cs import ControlSystem +from pytac.cs import AugmentedType, ControlSystem from pytac.exceptions import ControlSystemException +class AugmentedValue(AugmentedType): + """A dummy class for typehinting, + taken from aioca.aioca.types and cothread. + """ + + name: str + """Name of the PV used to create this value.""" + ok: bool + """True for normal data, False for error code.""" + + class CothreadControlSystem(ControlSystem): """A control system using cothread to communicate with EPICS. @@ -33,7 +44,7 @@ def __init__(self, timeout: float = 1.0, wait: bool = False) -> None: self._timeout = timeout self._wait = wait - def get_single(self, pv: str, throw: bool = True) -> Optional[Any]: + def get_single(self, pv: str, throw: bool = True) -> Optional[AugmentedType]: """Get the value of a given PV. Args: @@ -58,7 +69,9 @@ def get_single(self, pv: str, throw: bool = True) -> Optional[Any]: logging.warning(error_msg) return None - def get_multiple(self, pvs: Sequence[str], throw: bool = True) -> List[Any]: + def get_multiple( + self, pvs: Sequence[str], throw: bool = True + ) -> List[Optional[AugmentedType]]: """Get the value for given PVs. Args: @@ -73,8 +86,8 @@ def get_multiple(self, pvs: Sequence[str], throw: bool = True) -> List[Any]: ControlSystemException: if it cannot connect to one or more PVs. """ results = caget(pvs, timeout=self._timeout, throw=False) - return_values: List[Optional[Any]] = [] - failures: List[Any] = [] + return_values: List[Optional[AugmentedType]] = [] + failures = [] for result in results: if isinstance(result, ca_nothing): logging.warning(f"Cannot connect to {result.name}.") @@ -88,7 +101,7 @@ def get_multiple(self, pvs: Sequence[str], throw: bool = True) -> List[Any]: raise ControlSystemException(f"{len(failures)} caget calls failed.") return return_values - def set_single(self, pv: str, value: Any, throw: bool = True) -> bool: + def set_single(self, pv: str, value: AugmentedType, throw: bool = True) -> bool: """Set the value of a given PV. Args: @@ -115,7 +128,10 @@ def set_single(self, pv: str, value: Any, throw: bool = True) -> bool: return False def set_multiple( - self, pvs: Sequence[str], values: Any, throw: bool = True + self, + pvs: Sequence[str], + values: Sequence[AugmentedType], + throw: bool = True, ) -> Optional[List[bool]]: """Set the values for given PVs. diff --git a/src/pytac/cs.py b/src/pytac/cs.py index 249a19e3..5b89eda1 100644 --- a/src/pytac/cs.py +++ b/src/pytac/cs.py @@ -1,7 +1,19 @@ """Class representing an abstract control system.""" -from typing import Any, List, Sequence +from typing import ( + List, + Optional, + Sequence, + Sized, + SupportsFloat, + SupportsIndex, + SupportsInt, +) + + +class AugmentedType(SupportsFloat, SupportsInt, SupportsIndex, Sized): + pass class ControlSystem(object): @@ -11,7 +23,7 @@ class ControlSystem(object): over channel access with the hardware in the ring. """ - def get_single(self, pv: str, throw: bool) -> Any: + def get_single(self, pv: str, throw: bool) -> Optional[AugmentedType]: """Get the value of a given PV. Args: @@ -27,7 +39,9 @@ def get_single(self, pv: str, throw: bool) -> Any: """ raise NotImplementedError() - def get_multiple(self, pvs: Sequence[str], throw: bool) -> List[Any]: + def get_multiple( + self, pvs: Sequence[str], throw: bool + ) -> List[Optional[AugmentedType]]: """Get the value for given PVs. Args: @@ -43,7 +57,7 @@ def get_multiple(self, pvs: Sequence[str], throw: bool) -> List[Any]: """ raise NotImplementedError() - def set_single(self, pv: str, value: Any, throw: bool): + def set_single(self, pv: str, value: AugmentedType, throw: bool) -> bool: """Set the value of a given PV. Args: @@ -57,7 +71,9 @@ def set_single(self, pv: str, value: Any, throw: bool): """ raise NotImplementedError() - def set_multiple(self, pvs: Sequence[str], values: Sequence[Any], throw: bool): + def set_multiple( + self, pvs: Sequence[str], values: Sequence[AugmentedType], throw: bool + ) -> Optional[List[bool]]: """Set the values for given PVs. Args: diff --git a/src/pytac/data_source.py b/src/pytac/data_source.py index 59c540e5..d23e4ed2 100644 --- a/src/pytac/data_source.py +++ b/src/pytac/data_source.py @@ -1,9 +1,10 @@ """Module containing pytac data source classes.""" -from typing import Any, Dict, Iterable, List, Union +from typing import Dict, Iterable, Optional from _collections_abc import KeysView import pytac +from pytac.cs import AugmentedType from pytac.device import Device from pytac.exceptions import DataSourceException, FieldException from pytac.units import UnitConv @@ -27,7 +28,9 @@ def get_fields(self) -> Iterable: """ raise NotImplementedError() - def get_value(self, field: str, handle: str, throw: bool) -> Any: + def get_value( + self, field: str, handle: str, throw: bool + ) -> Optional[AugmentedType]: """Get a value for a field. Args: @@ -41,7 +44,7 @@ def get_value(self, field: str, handle: str, throw: bool) -> Any: """ raise NotImplementedError() - def set_value(self, field: str, value: float, throw: bool) -> Any: + def set_value(self, field: str, value: AugmentedType, throw: bool) -> None: """Set a value for a field. This is always set to pytac.SP, never pytac.RB. @@ -71,7 +74,7 @@ class DataSourceManager(object): """Holds the current default data source, pytac.LIVE or pytac.SIM, for an element or lattice.""" - _data_sources: Dict[str, DataSource] + _data_sources: Dict[str, "DeviceDataSource"] """A dictionary of the data sources held.""" _uc: Dict[str, UnitConv] """A dictionary of the unit conversion objects for each key(field).""" @@ -82,7 +85,9 @@ def __init__(self) -> None: self._data_sources = {} self._uc = {} - def set_data_source(self, data_source: DataSource, data_source_type: str) -> None: + def set_data_source( + self, data_source: "DeviceDataSource", data_source_type: str + ) -> None: """Add a data source to the manager. Args: @@ -92,7 +97,7 @@ def set_data_source(self, data_source: DataSource, data_source_type: str) -> Non """ self._data_sources[data_source_type] = data_source - def get_data_source(self, data_source_type: str) -> DataSource: + def get_data_source(self, data_source_type: str) -> "DeviceDataSource": """Get a data source. Args: @@ -194,7 +199,7 @@ def get_value( units: str = pytac.DEFAULT, data_source_type: str = pytac.DEFAULT, throw: bool = True, - ) -> float: + ) -> Optional[AugmentedType]: """Get the value for a field. Returns the value of a field on the manager. This value is uniquely @@ -231,7 +236,7 @@ def get_value( def set_value( self, field: str, - value: float, + value: AugmentedType, units: str = pytac.DEFAULT, data_source_type: str = pytac.DEFAULT, throw: bool = True, @@ -313,7 +318,7 @@ def get_fields(self) -> KeysView: def get_value( self, field: str, handle: str, throw: bool = True - ) -> Union[float, int, List[float], List[int]]: + ) -> Optional[AugmentedType]: """Get the value of a readback or setpoint PV for a field from the data_source. @@ -334,7 +339,7 @@ def get_value( def set_value( self, field: str, - value: Union[float, int, List[int], List[float]], + value: AugmentedType, throw: bool = True, ) -> None: """Set the value of a readback or setpoint PV for a field from the diff --git a/src/pytac/device.py b/src/pytac/device.py index 0623c6d3..52ca648d 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -5,10 +5,10 @@ DLS is a sextupole magnet that contains also horizontal and vertical corrector magnets and a skew quadrupole. """ -from typing import List, Optional, Union +from typing import Optional, Union, cast import pytac -from pytac.cs import ControlSystem +from pytac.cs import AugmentedType, ControlSystem from pytac.exceptions import DataSourceException, HandleException @@ -27,9 +27,7 @@ def is_enabled(self) -> bool: """ raise NotImplementedError() - def get_value( - self, handle: str, throw: bool - ) -> Union[float, int, List[int], List[float]]: + def get_value(self, handle: Optional[str], throw: bool) -> Optional[AugmentedType]: """Read the value from the device. Args: @@ -42,9 +40,7 @@ def get_value( """ raise NotImplementedError() - def set_value( - self, value: Union[float, int, List[int], List[float]], throw: bool - ) -> None: + def set_value(self, value: AugmentedType, throw: bool) -> None: """Set the value on the device. Args: @@ -64,7 +60,7 @@ class SimpleDevice(Device): the accelerator. """ - _value: Union[float, int, List[float], List[int]] + _value: Optional[AugmentedType] """The value of the device. May be a number or list of numbers.""" _enabled: Union[bool, "PvEnabler"] """Whether the device is enabled. May be a PvEnabler Object.""" @@ -73,7 +69,7 @@ class SimpleDevice(Device): def __init__( self, - value: Union[float, int, List[int], List[float]], + value: Optional[AugmentedType], enabled: Union[bool, "PvEnabler"] = True, readonly: bool = True, ) -> None: @@ -98,7 +94,7 @@ def is_enabled(self) -> bool: def get_value( self, handle: Optional[str] = None, throw: Optional[bool] = None - ) -> Union[float, int, List[int], List[float]]: + ) -> Optional[AugmentedType]: """Read the value from the device. Args: @@ -114,7 +110,7 @@ def get_value( def set_value( self, - value: Union[float, int, List[int], List[float]], + value: Optional[AugmentedType], throw: Optional[bool] = None, ) -> None: """Set the value on the device. @@ -187,7 +183,9 @@ def is_enabled(self) -> bool: """ return bool(self._enabled) - def get_value(self, handle: str, throw: bool = True) -> Union[float, int]: + def get_value( + self, handle: Optional[str], throw: bool = True + ) -> Optional[AugmentedType]: """Read the value of a readback or setpoint PV. Args: @@ -203,9 +201,7 @@ def get_value(self, handle: str, throw: bool = True) -> Union[float, int]: """ return self._cs.get_single(self.get_pv_name(handle), throw) - def set_value( - self, value: Union[float, int, List[int], List[float]], throw: bool = True - ) -> None: + def set_value(self, value: AugmentedType, throw: bool = True) -> None: """Set the device value. Args: @@ -218,7 +214,7 @@ def set_value( """ self._cs.set_single(self.get_pv_name(pytac.SP), value, throw) - def get_pv_name(self, handle: str) -> str: + def get_pv_name(self, handle: Optional[str]) -> str: """Get the PV name for the specified handle. Args: @@ -252,7 +248,7 @@ class PvEnabler(object): _cs: ControlSystem """The control system object.""" - def __init__(self, pv: str, enabled_value: str, cs: ControlSystem): + def __init__(self, pv: str, enabled_value: str, cs: ControlSystem) -> None: """ Args: pv: The PV name. @@ -270,5 +266,8 @@ def __bool__(self) -> bool: Returns: True if the device should be considered enabled. """ - pv_value = self._cs.get_single(self._pv, True) - return self._enabled_value == str(int(float(pv_value))) + pv_value = self._cs.get_single(self._pv, throw=True) + # pv_value is not None as throw is True. + # This would raise a ControlSystemException rather than returning None. + pv_value_cast = cast(AugmentedType, pv_value) + return self._enabled_value == str(int(float(pv_value_cast))) diff --git a/src/pytac/element.py b/src/pytac/element.py index a4a2f70e..4f9ec78c 100644 --- a/src/pytac/element.py +++ b/src/pytac/element.py @@ -1,8 +1,9 @@ """Module containing the element class.""" -from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Set +from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Sequence, Set, cast import pytac -from pytac.data_source import DataSource, DataSourceManager +from pytac.cs import AugmentedType +from pytac.data_source import DataSourceManager, DeviceDataSource from pytac.device import Device from pytac.exceptions import DataSourceException, FieldException from pytac.units import UnitConv @@ -71,7 +72,12 @@ def s(self) -> Optional[float]: if self._lattice is None: return None else: - return sum([el.length for el in self._lattice[: self.index - 1]]) + # index is not None when self._lattice is not None as per index property. + index = cast(int, self.index) + # Lattice must be iterable and indexable, however mypy doesnt accept only + # having __getitem__ and __len__ instead of __iter__. + typed_lattice = cast(Sequence, self._lattice) + return sum([el.length for el in typed_lattice[: index - 1]]) @property def cell(self) -> Optional[int]: @@ -85,7 +91,9 @@ def cell(self) -> Optional[int]: elif self._lattice.cell_length is None: return None else: - return int(self.s / self._lattice.cell_length) + 1 + # s is not None when self._lattice is not None as per s property. + s = cast(float, self.s) + return int(s / self._lattice.cell_length) + 1 @property def families(self) -> Set[Any]: @@ -128,7 +136,9 @@ def set_default_units(self, units: str) -> None: """ self._data_source_manager.default_units = units - def set_data_source(self, data_source: DataSource, data_source_type: str) -> None: + def set_data_source( + self, data_source: DeviceDataSource, data_source_type: str + ) -> None: """Add a data source to the element. Args: @@ -242,7 +252,7 @@ def get_value( units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, - ) -> float: + ) -> Optional[AugmentedType]: """Get the value for a field. Returns the value of a field on the element. This value is uniquely @@ -277,7 +287,7 @@ def get_value( def set_value( self, field: str, - value: float, + value: AugmentedType, units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 1e4186ce..820ed4f6 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -2,14 +2,15 @@ machine. """ import logging -from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Union +from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Union, cast import numpy +from numpy.typing import DTypeLike, NDArray import pytac -from pytac.cs import ControlSystem -from pytac.data_source import DataSource, DataSourceManager -from pytac.device import Device, EpicsDevice +from pytac.cs import AugmentedType, ControlSystem +from pytac.data_source import DataSourceManager, DeviceDataSource +from pytac.device import Device from pytac.element import Element from pytac.exceptions import DataSourceException, UnitsException from pytac.units import UnitConv @@ -75,7 +76,9 @@ def cell_bounds(self) -> Optional[List[int]]: for cell in range(2, self.symmetry + 1, 1): for elem in self._elements[bounds[-1] :]: if elem.cell == cell: - bounds.append(elem.index) + # index is not None when cell attribute is not None. + index = cast(int, elem.index) + bounds.append(index) break bounds.append(len(self._elements)) return bounds @@ -101,7 +104,9 @@ def __len__(self) -> int: """ return len(self._elements) - def set_data_source(self, data_source: DataSource, data_source_type: str) -> None: + def set_data_source( + self, data_source: DeviceDataSource, data_source_type: str + ) -> None: """Add a data source to the lattice. Args: @@ -187,7 +192,7 @@ def get_value( units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, - ) -> float: + ) -> Optional[AugmentedType]: """Get the value for a field on the lattice. Returns the value of a field on the lattice. This value is uniquely @@ -217,11 +222,11 @@ def get_value( def set_value( self, field: str, - value: float, + value: AugmentedType, units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, - ): + ) -> None: """Set the value for a field. This value can be set on the machine or the simulation. @@ -319,7 +324,7 @@ def get_family_s(self, family: str) -> List[float]: s_positions.append(element.s) return s_positions - def get_element_devices(self, family: str, field: str) -> List[EpicsDevice]: + def get_element_devices(self, family: str, field: str) -> List[Device]: """Get devices for a specific field for elements in the specfied family. @@ -335,7 +340,7 @@ def get_element_devices(self, family: str, field: str) -> List[EpicsDevice]: Devices for specified family and field. """ elements = self.get_elements(family) - devices = [] + devices: List[Device] = [] for element in elements: try: devices.append(element.get_device(field)) @@ -369,8 +374,8 @@ def get_element_values( units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, - dtype: Optional[numpy.dtype] = None, - ) -> Union[List[float], numpy.ndarray]: + dtype: Optional[DTypeLike] = None, + ) -> Union[List[Optional[AugmentedType]], NDArray]: """Get the value of the given field for all elements in the given family in the lattice. @@ -389,19 +394,20 @@ def get_element_values( The requested values. """ elements = self.get_elements(family) - values = [ + values: List[Optional[AugmentedType]] = [ element.get_value(field, handle, units, data_source, throw) for element in elements ] if dtype is not None: - values = numpy.array(values, dtype=dtype) + array_values: NDArray = numpy.array(values, dtype=dtype) + return array_values return values def set_element_values( self, family: str, field: str, - values: Sequence[Any], + values: Sequence[AugmentedType], units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, @@ -498,8 +504,13 @@ def get_default_data_source(self) -> str: return self._data_source_manager.default_data_source def convert_family_values( - self, family: str, field: str, values: Sequence[Any], origin: str, target: str - ) -> List[float]: + self, + family: str, + field: str, + values: Sequence[Optional[AugmentedType]], + origin: str, + target: str, + ) -> List[Optional[AugmentedType]]: """Convert the given values according to the given origin and target units, using the unit conversion objects for the given field on the elements in the given family. @@ -517,7 +528,7 @@ def convert_family_values( f"Number of elements in given sequence({len(values)}) must " f"be equal to the number of elements in the family({len(elements)})." ) - converted_values: List[float] = [] + converted_values = [] for elem, value in zip(elements, values): uc = elem.get_unitconv(field) converted_values.append(uc.convert(value, origin, target)) @@ -606,8 +617,8 @@ def get_element_values( units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, - dtype: Optional[numpy.dtype] = None, - ) -> Union[list, numpy.ndarray]: + dtype: Optional[DTypeLike] = None, + ) -> Union[List[Optional[AugmentedType]], NDArray]: """Get the value of the given field for all elements in the given family in the lattice. @@ -625,6 +636,8 @@ def get_element_values( Returns: The requested values. """ + values: Union[List[Optional[AugmentedType]], NDArray] = [] + if data_source == pytac.DEFAULT: data_source = self.get_default_data_source() if units == pytac.DEFAULT: @@ -641,14 +654,15 @@ def get_element_values( family, field, handle, units, data_source, throw ) if dtype is not None: - values = numpy.array(values, dtype=dtype) + array_values: NDArray = numpy.array(values, dtype=dtype) + return array_values return values def set_element_values( self, family: str, field: str, - values: Sequence[Any], + values: Sequence[AugmentedType], units: str = pytac.DEFAULT, data_source: str = pytac.DEFAULT, throw: bool = True, @@ -675,17 +689,23 @@ def set_element_values( units = self.get_default_units() if data_source == pytac.LIVE: if units == pytac.PHYS: - values = self.convert_family_values( + values_result = self.convert_family_values( family, field, values, pytac.PHYS, pytac.ENG ) + else: + values_result = [ + cast(Optional[AugmentedType], value) for value in values + ] pv_names = self.get_element_pv_names(family, field, pytac.SP) - if len(pv_names) != len(values): + if len(pv_names) != len(values_result): raise IndexError( f"Number of elements in given sequence({len(values)}) " "must be equal to the number of elements in " f"the family({len(pv_names)})." ) - self._cs.set_multiple(pv_names, values, throw) + # There is no reason to ever set a PV to None. + values_result_cast = cast(List[AugmentedType], values_result) + self._cs.set_multiple(pv_names, values_result_cast, throw) else: super(EpicsLattice, self).set_element_values( family, field, values, units, data_source, throw diff --git a/src/pytac/units.py b/src/pytac/units.py index b830edf6..52922d17 100644 --- a/src/pytac/units.py +++ b/src/pytac/units.py @@ -1,7 +1,8 @@ """Classes for use in unit conversion.""" -from typing import Any, Callable, List, Optional +from typing import Any, Callable, List, Optional, Union import numpy +from numpy.typing import NDArray from scipy.interpolate import PchipInterpolator import pytac @@ -33,7 +34,7 @@ class UnitConv: the conversion, the other happens before the conversion back. """ - name: Optional[str] + name: Optional[Union[str, int]] """An identifier for the unit conversion object.""" eng_units: str """The unit type of the post conversion engineering value.""" @@ -51,7 +52,7 @@ def __init__( pre_phys_to_eng: Callable[[float], float] = unit_function, engineering_units: str = "", physics_units: str = "", - name: Optional[str] = None, + name: Optional[Union[str, int]] = None, ) -> None: """Initialise the UnitConv Object. @@ -67,8 +68,8 @@ def __init__( self._pre_phys_to_eng = pre_phys_to_eng self.eng_units = engineering_units self.phys_units = physics_units - self.lower_limit: Optional[float] = None - self.upper_limit: Optional[float] = None + self.lower_limit: Any = None # Optional[float] + self.upper_limit: Any = None # Optional[float] def __str__(self) -> str: string_rep = self.__class__.__name__ @@ -76,36 +77,48 @@ def __str__(self) -> str: string_rep += f" {self.name}" return string_rep - def set_post_eng_to_phys(self, post_eng_to_phys: Callable[[float], float]) -> None: + def set_post_eng_to_phys(self, post_eng_to_phys: Callable[[Any], Any]) -> None: """Set the function to be applied after the initial conversion. + N.B. post_eng_to_phys should be of type: Callable[[float], float], but this + cannot be implimented with the current code structure. + Args: post_eng_to_phys: Function to be applied after the initial conversion. """ self._post_eng_to_phys = post_eng_to_phys - def set_pre_phys_to_eng(self, pre_phys_to_eng: Callable[[float], float]) -> None: + def set_pre_phys_to_eng(self, pre_phys_to_eng: Callable[[Any], Any]) -> None: """Set the function to be applied before the initial conversion. + N.B. pre_phys_to_eng should be of type: Callable[[float], float], but this + cannot be implimented with the current code structure. + Args: pre_phys_to_eng: Function to be applied before the initial conversion. """ self._pre_phys_to_eng = pre_phys_to_eng - def _raw_eng_to_phys(self, value: float): + def _raw_eng_to_phys(self, value: Any) -> Any: """Function to be implemented by child classes. + N.B. value and return should be of type: float, but this cannot be implimented + with the current code structure. + Args: value: The engineering value to be converted to physics units. """ raise NotImplementedError(f"{self}: No eng-to-phys conversion provided") - def eng_to_phys(self, value: float) -> float: + def eng_to_phys(self, value: Any) -> Any: """Function that does the unit conversion. Conversion from engineering to physics units. An additional function may be cast on the initial conversion. + N.B. value and return should be of type: float, but this cannot be implimented + with the current code structure. + Args: value: Value to be converted from engineering to physics units. @@ -140,20 +153,26 @@ def eng_to_phys(self, value: float) -> float: ) return valid_results[0] - def _raw_phys_to_eng(self, value: float): + def _raw_phys_to_eng(self, value: Any) -> Any: """Function to be implemented by child classes. + N.B. value and return should be of type: float, but this cannot be implimented + with the current code structure. + Args: value: The physics value to be converted to engineering units. """ raise NotImplementedError(f"{self}: No phys-to-eng conversion provided") - def phys_to_eng(self, value: float) -> float: + def phys_to_eng(self, value: Any) -> Any: """Function that does the unit conversion. Conversion from physics to engineering units. An additional function may be cast on the initial conversion. + N.B. value and return should be of type: float, but this cannot be implimented + with the current code structure. + Args: value: Value to be converted from physics to engineering units. @@ -185,10 +204,13 @@ def phys_to_eng(self, value: float) -> float: ) return valid_results[0] - def convert(self, value: float, origin: str, target: str) -> float: + def convert(self, value: Any, origin: str, target: str) -> Any: """Convert between two different unit types and check the validity of the result. + N.B. value and return should be of type: float, but this cannot be implimented + with the current code structure. + Args: value: the value to be converted origin: pytac.ENG or pytac.PHYS @@ -212,10 +234,13 @@ def convert(self, value: float, origin: str, target: str) -> float: f"{self}: Conversion from {origin} to {target} not understood." ) - def set_conversion_limits(self, lower_limit: float, upper_limit: float) -> None: + def set_conversion_limits(self, lower_limit: Any, upper_limit: Any) -> None: """Conversion limits to be applied before or after a conversion take place. Limits should be set in in engineering units. + N.B. lower_limit and upper_limit should be of type: Optional[float], but + this cannot be implimented with the current code structure. + Args: lower_limit: the lower conversion limit upper_limit: the upper conversion limit @@ -229,9 +254,12 @@ def set_conversion_limits(self, lower_limit: float, upper_limit: float) -> None: self.lower_limit = lower_limit self.upper_limit = upper_limit - def get_conversion_limits(self, units: str = pytac.ENG) -> List[float]: + def get_conversion_limits(self, units: str = pytac.ENG) -> List[Any]: """Return the current conversion limits in the specified unit type. + N.B. return should be type: List[Optional[float]], but this cannot be + implimented with the current code structure. + Args: units: The unit type. @@ -260,12 +288,12 @@ class PolyUnitConv(UnitConv): def __init__( self, - coef: numpy.ndarray[Any, numpy.dtype[numpy.generic]], + coef: NDArray[numpy.generic], post_eng_to_phys: Callable[[float], float] = unit_function, pre_phys_to_eng: Callable[[float], float] = unit_function, engineering_units: str = "", physics_units: str = "", - name: Optional[str] = None, + name: Optional[Union[str, int]] = None, ): """Initialise the PolyUnitConv Object. @@ -334,7 +362,7 @@ def __init__( pre_phys_to_eng: Callable[[float], float] = unit_function, engineering_units: str = "", physics_units: str = "", - name: Optional[str] = None, + name: Optional[Union[str, int]] = None, ) -> None: """ Args: diff --git a/src/pytac/utils.py b/src/pytac/utils.py index 331c3be3..462681fa 100644 --- a/src/pytac/utils.py +++ b/src/pytac/utils.py @@ -1,6 +1,6 @@ """Utility functions.""" import math -from typing import Callable +from typing import Any, Callable import scipy.constants @@ -8,9 +8,12 @@ electron_mass_mev, _, _ = scipy.constants.physical_constants[electron_mass_name] -def get_rigidity(energy_mev: float) -> float: +def get_rigidity(energy_mev: Any) -> float: """Get rigidity function. + N.B. energy_mev should be of type: float, but this cannot be implimented with + the current code structure. + Args: energy_mev: the energy of the lattice. @@ -24,9 +27,12 @@ def get_rigidity(energy_mev: float) -> float: return p / scipy.constants.e -def get_div_rigidity(energy: float) -> Callable[[float], float]: +def get_div_rigidity(energy: Any) -> Callable[[float], float]: """Return the function div_rigidity. + N.B. energy should be of type: float, but this cannot be implimented with + the current code structure. + Args: energy: the energy of the lattice. @@ -35,15 +41,18 @@ def get_div_rigidity(energy: float) -> Callable[[float], float]: """ rigidity = get_rigidity(energy) - def div_rigidity(value): + def div_rigidity(value: float) -> float: return value / rigidity return div_rigidity -def get_mult_rigidity(energy: float) -> Callable[[float], float]: +def get_mult_rigidity(energy: Any) -> Callable[[float], float]: """Return the function mult_rigidity. + N.B. energy should be of type: float, but this cannot be implimented with + the current code structure. + Args: energy: the energy of the lattice. @@ -52,7 +61,7 @@ def get_mult_rigidity(energy: float) -> Callable[[float], float]: """ rigidity = get_rigidity(energy) - def mult_rigidity(value): + def mult_rigidity(value: float) -> float: return value * rigidity return mult_rigidity From 742586d1a63b626c0b696295da39669d20b15088 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Wed, 30 Aug 2023 15:36:17 +0000 Subject: [PATCH 6/8] Update coefficient type --- src/pytac/device.py | 2 +- src/pytac/units.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pytac/device.py b/src/pytac/device.py index 52ca648d..a664d15f 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -5,7 +5,7 @@ DLS is a sextupole magnet that contains also horizontal and vertical corrector magnets and a skew quadrupole. """ -from typing import Optional, Union, cast +from typing import List, Optional, Union, cast import pytac from pytac.cs import AugmentedType, ControlSystem diff --git a/src/pytac/units.py b/src/pytac/units.py index 52922d17..9d680388 100644 --- a/src/pytac/units.py +++ b/src/pytac/units.py @@ -288,7 +288,7 @@ class PolyUnitConv(UnitConv): def __init__( self, - coef: NDArray[numpy.generic], + coef: Union[List[float], NDArray[numpy.generic]], post_eng_to_phys: Callable[[float], float] = unit_function, pre_phys_to_eng: Callable[[float], float] = unit_function, engineering_units: str = "", From ff4895535c45c7d00ed5595a73bf91cd2cdc2b66 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Thu, 31 Aug 2023 09:42:58 +0000 Subject: [PATCH 7/8] Fix mypy issues with cast and explicit typing. Also update docstring. --- src/pytac/device.py | 2 +- src/pytac/element.py | 12 +++++++----- src/pytac/lattice.py | 39 +++++++++++++++++++-------------------- src/pytac/load_csv.py | 22 +++++++++++++++++----- 4 files changed, 44 insertions(+), 31 deletions(-) diff --git a/src/pytac/device.py b/src/pytac/device.py index a664d15f..52ca648d 100644 --- a/src/pytac/device.py +++ b/src/pytac/device.py @@ -5,7 +5,7 @@ DLS is a sextupole magnet that contains also horizontal and vertical corrector magnets and a skew quadrupole. """ -from typing import List, Optional, Union, cast +from typing import Optional, Union, cast import pytac from pytac.cs import AugmentedType, ControlSystem diff --git a/src/pytac/element.py b/src/pytac/element.py index 4f9ec78c..9a6e71c3 100644 --- a/src/pytac/element.py +++ b/src/pytac/element.py @@ -4,7 +4,7 @@ import pytac from pytac.cs import AugmentedType from pytac.data_source import DataSourceManager, DeviceDataSource -from pytac.device import Device +from pytac.device import Device, EpicsDevice from pytac.exceptions import DataSourceException, FieldException from pytac.units import UnitConv @@ -346,11 +346,13 @@ def get_pv_name(self, field: str, handle: str) -> str: FieldException: if the specified field doesn't exist. """ try: - return ( - self._data_source_manager.get_data_source(pytac.LIVE) - .get_device(field) - .get_pv_name(handle) + device = self._data_source_manager.get_data_source(pytac.LIVE).get_device( + field ) + # EpicsDevice is the only Device class to have the get_pv_name method. + # If the device is not EpicsDevice then an exception would be raised. + epics_device = cast(EpicsDevice, device) + return epics_device.get_pv_name(handle) except DataSourceException as e: raise DataSourceException(f"{self}: {e}") except AttributeError: diff --git a/src/pytac/lattice.py b/src/pytac/lattice.py index 820ed4f6..721e51dd 100644 --- a/src/pytac/lattice.py +++ b/src/pytac/lattice.py @@ -10,8 +10,8 @@ import pytac from pytac.cs import AugmentedType, ControlSystem from pytac.data_source import DataSourceManager, DeviceDataSource -from pytac.device import Device -from pytac.element import Element +from pytac.device import Device, EpicsDevice +from pytac.element import Element, EpicsElement from pytac.exceptions import DataSourceException, UnitsException from pytac.units import UnitConv @@ -364,7 +364,10 @@ def get_element_device_names(self, family: str, field: str) -> List[str]: list: device names for specified family and field. """ devices = self.get_element_devices(family, field) - return [device.name for device in devices] + # EpicsDevice is the only Device class to have the .name attribute. + # If the device is not EpicsDevice then an exception would be raised. + epics_devices = [cast(EpicsDevice, device) for device in devices] + return [device.name for device in epics_devices] def get_element_values( self, @@ -435,7 +438,8 @@ def set_element_values( f"equal to the number of elements in the family({len(elements)})." ) for element, value in zip(elements, values): - status = element.set_value( + # Python implicitly returns None, but mypy does not recognise this. + status = element.set_value( # type: ignore field, value, units=units, @@ -540,20 +544,11 @@ class EpicsLattice(Lattice): Allows efficient get_element_values() and set_element_values() methods, and adds get_pv_names() method. - - Attributes: - name: The name of the lattice. - symmetry: The symmetry of the lattice (the number of cells). - - .. Private Attributes: - _elements: The list of all the element objects in the lattice - _cs: The control system to use for the more efficient batch getting and - setting of PVs. - _data_source_manager: A class that manages the data sources associated - with this lattice. """ _cs: ControlSystem + """The control system to use for the more efficient batch + getting and setting of PVs.""" def __init__( self, name: str, epics_cs: ControlSystem, symmetry: Optional[int] = None @@ -578,11 +573,13 @@ def get_pv_name(self, field: str, handle: str) -> str: The readback or setpoint PV for the specified field. """ try: - return ( - self._data_source_manager.get_data_source(pytac.LIVE) - .get_device(field) - .get_pv_name(handle) + device = self._data_source_manager.get_data_source(pytac.LIVE).get_device( + field ) + # EpicsDevice is the only Device class to have the get_pv_name method. + # If the device is not EpicsDevice then an exception would be raised. + epics_device = cast(EpicsDevice, device) + return epics_device.get_pv_name(handle) except AttributeError: raise DataSourceException( f"Cannot get PV for field {field} on lattice " @@ -606,7 +603,9 @@ def get_element_pv_names(self, family: str, field: str, handle: str) -> List[str elements = self.get_elements(family) pv_names = [] for element in elements: - pv_names.append(element.get_pv_name(field, handle)) + # EpicsElement is the only Element class to have the get_pv_name method. + epics_element = cast(EpicsElement, element) + pv_names.append(epics_element.get_pv_name(field, handle)) return pv_names def get_element_values( diff --git a/src/pytac/load_csv.py b/src/pytac/load_csv.py index 80b46f7b..853d23ea 100644 --- a/src/pytac/load_csv.py +++ b/src/pytac/load_csv.py @@ -14,12 +14,13 @@ import copy import csv from pathlib import Path -from typing import Dict, Iterator, Optional +from typing import Dict, Iterator, Optional, Sequence, Union, cast import pytac from pytac import data_source, element, utils from pytac.cs import ControlSystem from pytac.device import EpicsDevice, SimpleDevice +from pytac.element import Element from pytac.exceptions import ControlSystemException from pytac.lattice import EpicsLattice, Lattice from pytac.units import NullUnitConv, PchipUnitConv, PolyUnitConv, UnitConv @@ -202,13 +203,21 @@ def load( pve = True d = EpicsDevice(name, control_system, pve, get_pv, set_pv) # Devices on index 0 are attached to the lattice not elements. - target = lat if index == 0 else lat[index - 1] + # Explicitly type target as the base classes to validate add_device method. + target: Union[Lattice, Element] = lat if index == 0 else lat[index - 1] target.add_device(item["field"], d, DEFAULT_UC) # Add basic devices to the lattice. positions = [] - for elem in lat: + # Lattice must be iterable and indexable, however mypy doesnt accept only + # having __getitem__ and __len__ instead of __iter__. + type_lattice = cast(Sequence, lat) + for elem in type_lattice: positions.append(elem.s) - lat.add_device("s_position", SimpleDevice(positions, readonly=True), True) + # AugmentedType can be a list. Until cothread is typed correctly + # this is impractical to fix. + # The add_device method on the lattice accepts a boolean as a UnitConv object. + # This should not be typed as such and DEFAULT_UC should be used instead. + lat.add_device("s_position", SimpleDevice(positions, readonly=True), True) # type: ignore # noqa: E501 simple_devices_file = mode_dir / SIMPLE_DEVICES_FILENAME if simple_devices_file.exists(): with csv_loader(simple_devices_file) as csv_reader: @@ -219,7 +228,10 @@ def load( readonly = item["readonly"].lower() == "true" # Devices on index 0 are attached to the lattice not elements. target = lat if index == 0 else lat[index - 1] - target.add_device(field, SimpleDevice(value, readonly=readonly), True) + # The add_device method on the lattice accepts a boolean as a + # UnitConv object. # This should not be typed as such and DEFAULT_UC + # should be used instead. + target.add_device(field, SimpleDevice(value, readonly=readonly), True) # type: ignore # noqa: E501 with csv_loader(mode_dir / FAMILIES_FILENAME) as csv_reader: for item in csv_reader: lat[int(item["el_id"]) - 1].add_to_family(item["family"]) From 2637548549f0e746caa1db72e41e546db9da0761 Mon Sep 17 00:00:00 2001 From: Joshua Appleby Date: Thu, 31 Aug 2023 09:48:25 +0000 Subject: [PATCH 8/8] Fix import of SupportsIndex for py3.7 --- src/pytac/cs.py | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/src/pytac/cs.py b/src/pytac/cs.py index 5b89eda1..18a0f299 100644 --- a/src/pytac/cs.py +++ b/src/pytac/cs.py @@ -1,15 +1,11 @@ """Class representing an abstract control system.""" +import sys +from typing import List, Optional, Sequence, Sized, SupportsFloat, SupportsInt - -from typing import ( - List, - Optional, - Sequence, - Sized, - SupportsFloat, - SupportsIndex, - SupportsInt, -) +if sys.version_info < (3, 8): + from typing_extensions import SupportsIndex # noqa +else: + from typing import SupportsIndex # noqa class AugmentedType(SupportsFloat, SupportsInt, SupportsIndex, Sized):