diff --git a/examples/lcls/server.py b/examples/lcls/server.py index 7155e5a..46b3c53 100644 --- a/examples/lcls/server.py +++ b/examples/lcls/server.py @@ -28,7 +28,9 @@ def _evaluate(self, input_dict: dict) -> dict: parser = argparse.ArgumentParser() parser.add_argument('--standalone', action='store_true', help='Spins up an internal EPICS PVA server to simulate the PVs used by this test') args = parser.parse_args() - + + sim_server = None + # Spin up P4P server if requested if args.standalone: with open(Path(__file__).parent / "sim_config.yml") as fp: diff --git a/lume_epics/epics_pva_server.py b/lume_epics/epics_pva_server.py index 163fbd9..826069b 100644 --- a/lume_epics/epics_pva_server.py +++ b/lume_epics/epics_pva_server.py @@ -4,7 +4,7 @@ from multiprocessing.managers import DictProxy from multiprocessing.sharedctypes import Synchronized from queue import Full, Empty -from lume_epics import model +from lume_epics import model, types import numpy as np import time import signal @@ -20,6 +20,7 @@ from p4p.nt.ndarray import ntndarray as NTNDArrayData from p4p.server.raw import ServOpWrap from p4p import Value, Type +from lume_epics.types import type_handler p4p_logger = logging.getLogger("p4p") p4p_logger.setLevel("DEBUG") @@ -119,7 +120,8 @@ def update_pv(self, pvname: str, value: Union[np.ndarray, float]) -> None: value = value.raw.value varname = self._pvname_to_varname_map[pvname] - model_variable = self._get_default_value(self._input_variables[varname]) + var = self._input_variables[varname] + model_variable = type_handler(var).default_value(var) # check for already cached variable model_variable = self._cached_values.get(varname, model_variable) @@ -136,7 +138,8 @@ def _monitor_callback(self, pvname, V) -> None: """Callback function used for updating read_only process variables.""" value = V.raw.value varname = self._pvname_to_varname_map[pvname] - model_variable = self._get_default_value(self._input_variables[varname]) + var = self._input_variables[varname] + model_variable = type_handler(var).default_value(var) if not model_variable: model_variable = self._output_variables[varname] @@ -152,40 +155,6 @@ def _monitor_callback(self, pvname, V) -> None: self._in_queue.put({"protocol": self.protocol, "vars": self._cached_values, "vals": self._input_values}) self._cached_values = {} - def _get_default_value(self, variable: Variable) -> Any: - """ Returns the default value for the variable """ - if isinstance(variable, ScalarVariable): - return variable.default_value - else: - return None - - def _build_scalar_type(self, initial, config: dict) -> Tuple[Value, NTScalar]: - """ - Builds a scalar type based on the spec. - - Parameters - ---------- - initial : Any - Initial value - config : dict - Configuration - - Returns - ------- - tuple[Value, NTScalar] - Tuple containing the wrapped initia value and the NTScalar type - """ - nt = NTScalar("d", display=True, control=True) - initial_value = nt.wrap(initial) if initial else nt.wrap(0.0) - - # Set display parameters - initial_value['display']['description'] = config.get('description', '') - - # Start with sensible default for the timestamp - self._update_timestamp(initial_value) - - return (initial_value, nt) - def _make_timestamp(self, ts: float) -> Tuple[int, int]: """ Converts a timestamp into a tuple that can be fed to EPICS @@ -203,9 +172,9 @@ def _make_timestamp(self, ts: float) -> Tuple[int, int]: f, i = math.modf(ts) return (i, int(f * 1e9)) - def _update_timestamp(self, pv: Value, ts: float = time.time()) -> None: + def _update_timestamp(self, pv: Value, ts: float = 0) -> None: """ - Updates the timestamp on a PV structure + Updates the timestamp on a PV structure, if it has one Parameters ---------- @@ -214,7 +183,10 @@ def _update_timestamp(self, pv: Value, ts: float = time.time()) -> None: ts : float Timestamp in seconds since UNIX epoch """ - sec, nsec = self._make_timestamp(ts) + if 'timeStamp' not in pv: + return + + sec, nsec = self._make_timestamp(ts if ts > 0 else time.time()) pv['timeStamp']['secondsPastEpoch'] = sec pv['timeStamp']['nanoseconds'] = nsec @@ -225,6 +197,149 @@ def _initialize_model(self): self._in_queue.put(rep) + def _create_monitor(self, variable: Variable, config: dict) -> None: + """ + Creates a new monitored remote PV + + Parameters + ---------- + variable : Variable + LUME variable + config : dict + Configuration + """ + pvname = config.get("pvname") + + if variable.name in self._input_variables: + self._monitors[pvname] = self._context.monitor( + pvname, partial(self._monitor_callback, pvname) + ) + # in this case, externally hosted output variable + else: + self._providers[pvname] = None + + def _create_summary(self): + """Creates the summary PV, describing the model""" + pvname = self._epics_config["summary"].get("pvname") + owner = self._epics_config["summary"].get("owner") + date_published = self._epics_config["summary"].get("date_published") + description = self._epics_config["summary"].get("description") + id = self._epics_config["summary"].get("id") + + spec = [ + ("id", "s"), + ("owner", "s"), + ("date_published", "s"), + ("description", "s"), + ("input_variables", "as"), + ("output_variables", "as"), + ] + values = { + "id": id, + "date_published": date_published, + "description": description, + "owner": owner, + "input_variables": [ + self._epics_config[var]["pvname"] + for var in self._input_variables + ], + "output_variables": [ + self._epics_config[var]["pvname"] + for var in self._input_variables + ], + } + + pv_type = Type(id="summary", spec=spec) + value = Value(pv_type, values) + pv = SharedPV(initial=value) + self._providers[pvname] = pv + + def _create_struct(self, config: dict, variable_name: str, variables: Dict[str, Variable]) -> None: + """ + Create a new structure + + Parameters + ---------- + config : dict + Configuration for this structure/variable, from the YAML file + variable_name : str + Name of the variable + variables : Dict[str, Variable] + List of variables described already + """ + spec = [] + structure = {} + + fields = config.get("fields") + pvname = config.get("pvname") + + for field in fields: + # track fields in dict + self._field_to_parent_map[field] = variable_name + variable = variables[field] + initial = variable.default_value + + if variable is None: + raise ValueError( + f"Field {field} for {variable_name} not found in variable list" + ) + + handler = type_handler(variable) + if handler is None: + raise ValueError(f"Unsupported variable type provided: {type(variable)}") + + initial = handler.initial_value(config, variable) + spec.append((field, 'v')) # Using variant here because we can't extract tuple struct desc from the NT types in p4p... + + structure[field] = initial + + # Set default output var value + self._output_values[variable.name] = initial + + # Assemble type and value + struct_type = Type(id=variable_name, spec=spec) + struct_value = Value(struct_type, structure) + + # Store off type and current value + self._structures[variable_name] = structure + self._structure_types[variable_name] = struct_type + pv = SharedPV(initial=struct_value) + self._providers[pvname] = pv + + def _create_variable(self, config: dict, variable: Variable) -> None: + """ + Create a new variable + + Parameters + ---------- + config : dict + Configuration for this variable + variable : Variable + LUME variable + """ + pvname = config.get("pvname") + + handler = type_handler(variable) + if handler is None: + raise ValueError(f"Unsupported variable type provided: {type(variable)}") + + # Create an initial Value() + initial = handler.initial_value(config, variable) + + if variable.name in self._input_variables: + handler = PVAccessInputHandler( + pvname=pvname, + is_constant=variable.is_constant, + server=self, + ) + pv = SharedPV(handler=handler, initial=initial) + else: + pv = SharedPV(initial=initial) + + # Set default output var value + self._output_values[variable.name] = initial + self._providers[pvname] = pv + def setup_server(self) -> None: """Configure and start server.""" @@ -262,150 +377,45 @@ def setup_server(self) -> None: except Empty: pass + # No need to do this if we're being shutdown if self.shutdown_event.is_set(): - pass - - # if startup hasn't failed - else: - model_output_vars = model_outputs.get("output_variables", {}) - self._output_variables.update(model_output_vars) + return + + model_output_vars = model_outputs.get("output_variables", {}) + self._output_variables.update(model_output_vars) + + variables = copy.deepcopy(self._input_variables) + variables.update(self._output_variables) + + # ignore interrupt in subprocess + signal.signal(signal.SIGINT, signal.SIG_IGN) + logger.info("Initializing pvAccess server") + + # initialize global inputs + self._structures = {} + self._structure_types: Dict[str, Type] = {} + + # Initialize all of the variables specified in our config + for variable_name, config in self._epics_config.items(): + # Not served, create a monitor + if not config["serve"]: + self._create_monitor(variables[variable_name], config) + continue - variables = copy.deepcopy(self._input_variables) - variables.update(self._output_variables) + # Handle structures + if "fields" in config: + self._create_struct(config, variable_name, variables) + else: + self._create_variable(config, variables[variable_name]) - # ignore interrupt in subprocess - signal.signal(signal.SIGINT, signal.SIG_IGN) + # Create a summary PV, if requested. + if "summary" in self._epics_config: + self._create_summary() - logger.info("Initializing pvAccess server") + # initialize pva server + self.pva_server = P4PServer(providers=[self._providers]) - # initialize global inputs - self._structures = {} - self._structure_types: Dict[str, Type] = {} - for variable_name, config in self._epics_config.items(): - if config["serve"]: - fields = config.get("fields") - pvname = config.get("pvname") - - if fields is not None: - spec = [] - structure = {} - - for field in fields: - # track fields in dict - self._field_to_parent_map[field] = variable_name - - variable = variables[field] - initial = variable.default_value - - if variable is None: - raise ValueError( - f"Field {field} for {variable_name} not found in variable list" - ) - - if isinstance(variable, ScalarVariable): - initial, nt = self._build_scalar_type(initial, config) - spec.append((field, 'v')) # Using variant here because we can't extract tuple struct desc from the NT types in p4p... - - structure[field] = initial - - # Set default output var value - self._output_values[variable.name] = initial - - # Assemble type and value - struct_type = Type(id=variable_name, spec=spec) - struct_value = Value(struct_type, structure) - - # Store off type and current value - self._structures[variable_name] = structure - self._structure_types[variable_name] = struct_type - - pv = SharedPV(initial=struct_value) - self._providers[pvname] = pv - - else: - variable = variables[variable_name] - - initial = variable.default_value - - # prepare scalar variable types - if isinstance(variable, ScalarVariable): - initial, nt = self._build_scalar_type(initial, config) - else: - raise ValueError( - "Unsupported variable type provided: %s", - variable.variable_type, - ) - - if variable.name in self._input_variables: - handler = PVAccessInputHandler( - pvname=pvname, - is_constant=variable.is_constant, - server=self, - ) - - pv = SharedPV(handler=handler, nt=nt, initial=initial) - - else: - pv = SharedPV(nt=nt, initial=initial) - - # Set default output var value - self._output_values[variable.name] = initial - - self._providers[pvname] = pv - - # if not serving pv, set up monitor - else: - variable = variables[variable_name] - pvname = config.get("pvname") - - if variable.name in self._input_variables: - self._monitors[pvname] = self._context.monitor( - pvname, partial(self._monitor_callback, pvname) - ) - - # in this case, externally hosted output variable - else: - self._providers[pvname] = None - - if "summary" in self._epics_config: - pvname = self._epics_config["summary"].get("pvname") - owner = self._epics_config["summary"].get("owner") - date_published = self._epics_config["summary"].get("date_published") - description = self._epics_config["summary"].get("description") - id = self._epics_config["summary"].get("id") - - spec = [ - ("id", "s"), - ("owner", "s"), - ("date_published", "s"), - ("description", "s"), - ("input_variables", "as"), - ("output_variables", "as"), - ] - values = { - "id": id, - "date_published": date_published, - "description": description, - "owner": owner, - "input_variables": [ - self._epics_config[var]["pvname"] - for var in self._input_variables - ], - "output_variables": [ - self._epics_config[var]["pvname"] - for var in self._input_variables - ], - } - - pv_type = Type(id="summary", spec=spec) - value = Value(pv_type, values) - pv = SharedPV(initial=value) - self._providers[pvname] = pv - - # initialize pva server - self.pva_server = P4PServer(providers=[self._providers]) - - logger.info("pvAccess server started") + logger.info("pvAccess server started") def update_pvs( self, @@ -439,6 +449,8 @@ def update_pvs( ) value = output_values[variable.name] + handler = type_handler(variable) + # update structure or pv if parent: self._structures[parent][variable.name]['value'] = value @@ -453,10 +465,12 @@ def update_pvs( output_provider = self._providers[pvname] if output_provider: - if isinstance(value, Value): - output_provider.post(value) - else: - output_provider.post(value, timestamp=time.time()) + # Convert to value if it hasn't been already + if not isinstance(value, Value): + value = handler.to_value(value) + + self._update_timestamp(value) + output_provider.post(value) # in this case externally hosted else: @@ -504,6 +518,7 @@ class PVAccessInputHandler: """ Handler object that defines the callbacks to execute on put operations to input process variables. + This will proxy PUT operations into the internal cache. """ def __init__(self, pvname: str, is_constant: bool, server: PVAServer): diff --git a/lume_epics/types.py b/lume_epics/types.py new file mode 100644 index 0000000..dfcd30b --- /dev/null +++ b/lume_epics/types.py @@ -0,0 +1,156 @@ + +from p4p.server.thread import SharedPV +from p4p import Type, Value +from p4p.nt import NTScalar, NTBase +from typing import Any, Dict +from lume_model.variables import Variable, ScalarVariable +import typing + +class VariableTypeHandler: + """ + Base class for all variable types + Implements type-specific operations in a portable manner. + + Specializations of this class for additional types should be added to the _TYPE_HANDLERS dict, keyed + by the LUME variable type they're specialized for (i.e. ScalarVariable). + + Type handlers for a specific variable instance may be obtained using the type_handler() function in this module. + type_handlers() returns the full list. + """ + + def default_value(self, variable: Variable) -> Any: + """ + Returns the default value of the variable + + Parameters + ---------- + variable : Variable + The variable instance to get the default value for + + Returns + ------- + Any : + The default value + """ + raise NotImplementedError() + + def pva_typedef(self) -> Type: + """ + Returns the p4p typedef for the variable. + """ + raise NotImplementedError() + + def ca_typedef(self) -> dict: + """ + Returns the CA typedef, passed to pcaspy to create the structure. + """ + raise NotImplementedError() + + def to_value(self, pyvalue: Any) -> Value: + """ + Convert the variable's value to a p4p Value + + Parameters + ---------- + pyvalue : Any + Convert the Python value to a P4P Value + + Returns + ------- + Value : + The P4P value + """ + raise NotImplementedError() + + def from_value(self, value: Value) -> Any: + """ + Convert the p4p Value to a value that can be used in Python + + Parameters + ---------- + value : Value + P4P value to be converted to a Python value + + Returns + ------- + Any : + The value converted to Python + """ + raise NotImplementedError() + + def initial_value(self, config: dict, variable: Variable) -> Value: + """ + Create an initial value for p4p to use + This should generally include important metadata (description, units, etc.) that are not + otherwise going to change. + + Parameters + ---------- + config : dict + Configuration for this variable. + variable : Variable + The variable to obtain the default from, and convert it to p4p.Value + + Returns + ------- + Value : + P4P value to be passed to SharedPV() + """ + +class ScalarTypeHandler(VariableTypeHandler): + """Type handler for ScalarVariable""" + def __init__(self): + self._nt = NTScalar('d', control=True, display=True) + + def default_value(self, variable: Variable) -> Any: + d = variable.default_value + return d if d is not None else 0.0 + + def pva_typedef(self): + return NTScalar.buildType('d', control=True, display=True) + + def ca_typedef(self): + raise NotImplementedError() + + def to_value(self, pyvalue: Any) -> Value: + return self._nt.wrap(pyvalue) + + def from_value(self, value: Value) -> Any: + return self._nt.unwrap(value) + + def initial_value(self, config: dict, variable: Variable) -> Value: + v = self.to_value(self.default_value(variable)) + + # Set initial display parameters + v['display']['description'] = config.get('description', '') + return v + + +_TYPE_HANDLERS = { + ScalarVariable: ScalarTypeHandler() +} + +def type_handlers() -> Dict[typing.Type, VariableTypeHandler]: + """ + Returns a map of LUME variable type -> handler. + """ + return _TYPE_HANDLERS + +def type_handler(var: Variable) -> VariableTypeHandler | None: + """ + Returns the type handler for the variable + + Parameters + ---------- + var : Variable + The variable to get the type handler for. + + Returns + ------- + VariableTypeHandler | None : + The type handler, or None if one is not registered for this variable type. + """ + try: + return _TYPE_HANDLERS[type(var)] + except: + return None