Merge pull request #84 from jacquelinegarrahan/structure-pvs

Introduce structured pvs and summary variable

docs/EPICS.md

@@ -33,6 +33,70 @@ The optional field `serve` for each variable accepts a boolean defaulting to tru
 
 The client controller `lume_epics.client.controller.Controller` is also initialized using the EPICS configuration dictionary and a common file may be used for a project, though the serve field is unimportant to the controller.
 
+Over pvAccess, you also have the option to host a summary process variable:
+
+```yaml
+input_variables:
+  input1:
+    pvname: test:input1
+    protocol: ca
+    serve: false
+
+  input2:
+    pvname: test:input2
+    protocol: pva
+
+output_variables:
+  output1:
+    pvname: test:output1
+    protocol: pva
+
+  output2:
+    pvname: test:output2
+    protocol: pva
+
+  output3:
+    pvname: test:output3
+    protocol: pva
+
+summary:
+  pvname: test:summary_variable
+  owner: Jacqueline Garrahan
+  date_published: 1/27/22
+  description: A basic epics configuration
+  id: model1
+```
+
+
+You can also serve output variables as a pvAccess structure:
+
+```yaml
+input_variables:
+  input1:
+    pvname: test:input1
+    protocol: ca
+    serve: false
+
+  input2:
+    pvname: test:input2
+    protocol: pva
+
+output_variables:
+  pvname: test:output
+  protocol: pva
+  fields:
+    - output1
+    - output2
+    - output3
+
+summary:
+  pvname: test:summary_variable
+  owner: Jacqueline Garrahan
+  date_published: 1/27/22
+  description: A basic epics configuration
+  id: model1
+```
+
 
 ## EPICS environment configuration
 

examples/files/epics_config_struct.yml

@@ -0,0 +1,17 @@
+input_variables:
+  input1:
+    pvname: test:input1
+    protocol: ca
+
+  input2:
+    pvname: test:input2
+    protocol: pva
+
+output_variables:
+  output_summary:
+    pvname: test:output_summary
+    protocol: pva
+    fields:
+      - output1
+      - output2
+      - output3

examples/files/iris_epics_config.yml

@@ -19,3 +19,10 @@ output_variables:
   Species:
     pvname: test:species
     protocol: pva
+
+summary:
+  pvname: test:iris:model
+  description: Iris example for lume-epics
+  owner: Jacqueline Garrahan
+  date_published: 1/13/2022
+  id: model1

examples/keras/iris_server.py

@@ -1,16 +1,15 @@
-
 from lume_epics.epics_server import Server
 from lume_model.utils import model_from_yaml
+from lume_epics.utils import config_from_yaml
+
+if __name__ == "__main__":
+    with open("examples/files/iris_config.yml", "r") as f:
+        model_class, model_kwargs = model_from_yaml(f, load_model=False)
 
-with open("examples/files/iris_config.yml", "r") as f:
-    model_class, model_kwargs = model_from_yaml(f, load_model=False)
+    with open("examples/files/iris_epics_config.yml", "r") as f:
+        epics_config = config_from_yaml(f)
 
-prefix = "test"
-server = Server(
-    model_class,
-    prefix,
-    model_kwargs=model_kwargs
-)
+    server = Server(model_class, epics_config, model_kwargs=model_kwargs)
 
-# monitor = False does not loop in main thread
-server.start(monitor=True)
+    # monitor = False does not loop in main thread
+    server.start(monitor=True)

examples/struct/server.py

@@ -0,0 +1,26 @@
+from lume_epics.epics_server import Server
+from lume_model.utils import variables_from_yaml
+from lume_epics.utils import config_from_yaml
+import os
+from examples.model import DemoModel
+
+import logging
+
+
+if __name__ == "__main__":
+    with open("examples/files/demo_config.yml", "r") as f:
+        input_variables, output_variables = variables_from_yaml(f)
+
+    with open("examples/files/epics_config_struct.yml", "r") as f:
+        epics_config_struct = config_from_yaml(f)
+
+    server = Server(
+        DemoModel,
+        epics_config_struct,
+        model_kwargs={
+            "input_variables": input_variables,
+            "output_variables": output_variables,
+        },
+    )
+    # monitor = False does not loop in main thread
+    server.start(monitor=True)

lume_epics/epics_pva_server.py

@@ -11,11 +11,12 @@
 from functools import partial
 from lume_model.variables import InputVariable, OutputVariable
 from p4p.client.thread import Context
-from p4p.nt import NTScalar, NTNDArray
+from p4p.nt import NTScalar, NTNDArray, NTTable
 from p4p.server.thread import SharedPV
 from p4p.server import Server as P4PServer
 from p4p.nt.ndarray import ntndarray as NTNDArrayData
 from p4p.server.raw import ServOpWrap
+from p4p import Value, Type
 
 p4p_logger = logging.getLogger("p4p")
 p4p_logger.setLevel("DEBUG")
@@ -188,6 +189,7 @@ def setup_server(self) -> None:
                 try:
                     val = self._context.get(self._varname_to_pvname_map[var_name])
                     val = val.raw.value
+
                 except:
                     self.exit_event.set()
                     raise ValueError(
@@ -224,55 +226,136 @@ def setup_server(self) -> None:
             logger.info("Initializing pvAccess server")
 
             # initialize global inputs
-            for variable in variables.values():
-                pvname = self._varname_to_pvname_map[variable.name]
-
-                if self._epics_config[variable.name]["serve"]:
-
-                    # prepare scalar variable types
-                    if variable.variable_type == "scalar":
-                        nt = NTScalar("d")
-                        initial = variable.value
-
-                    # prepare image variable types
-                    elif variable.variable_type == "image":
-                        nd_array = variable.value.view(NTNDArrayData)
-                        nd_array.attrib = {
-                            "x_min": variable.x_min,
-                            "y_min": variable.y_min,
-                            "x_max": variable.x_max,
-                            "y_max": variable.y_max,
-                        }
-                        nt = NTNDArray()
-                        initial = nd_array
-
-                    elif variable.variable_type == "array":
-                        if variable.value_type == "str":
-                            nt = NTScalar("as")
+            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:
+
+                            variable = variables[field]
+
+                            if variable is None:
+                                raise ValueError(
+                                    f"Field {field} for {variable_name} not found in variable list"
+                                )
+
+                            if variable.variable_type == "scalar":
+                                spec.append((field, "d"))
+                                nt = NTScalar("d")
+                                initial = variable.value
+
+                            if variable.variable_type == "table":
+                                spec.append((field, "v"))
+                                table_rep = ()
+                                for col in variable.columns:
+
+                                    # here we assume double type in tables...
+                                    table_rep += (col, "ad")
+
+                                nt = NTTable(table_rep)
+                                initial = nt.wrap(variable.value)
+
+                            if variable.variable_type == "array":
+                                spec.append((field, "v"))
+
+                                if variable.value_type == "str":
+                                    nt = NTScalar("as")
+                                    initial = nt.wrap(variable.value)
+
+                                else:
+                                    nd_array = variable.value.view(NTNDArrayData)
+                                    nt = NTNDArray()
+                                    initial = nt.wrap(nd_array)
+
+                            if variable.variable_type == "image":
+                                spec.append((field, "v"))
+
+                                nd_array = variable.value.view(NTNDArrayData)
+                                nd_array.attrib = {
+                                    "x_min": variable.x_min,
+                                    "y_min": variable.y_min,
+                                    "x_max": variable.x_max,
+                                    "y_max": variable.y_max,
+                                }
+
+                                nt = NTNDArray()
+                                initial = nt.wrap(nd_array)
+
+                            structure[field] = initial
+
+                        # assemble pv
+                        struct_type = Type(id=variable_name, spec=spec)
+
+                        struct_value = Value(struct_type, structure)
+                        pv = SharedPV(initial=struct_value)
+                        self._providers[pvname] = pv
+
+                    else:
+                        variable = variables[variable_name]
+                        # prepare scalar variable types
+                        if variable.variable_type == "scalar":
+                            nt = NTScalar("d")
                             initial = variable.value
 
-                        else:
+                        # prepare image variable types
+                        elif variable.variable_type == "image":
                             nd_array = variable.value.view(NTNDArrayData)
+                            nd_array.attrib = {
+                                "x_min": variable.x_min,
+                                "y_min": variable.y_min,
+                                "x_max": variable.x_max,
+                                "y_max": variable.y_max,
+                            }
                             nt = NTNDArray()
                             initial = nd_array
 
-                    else:
-                        raise ValueError(
-                            "Unsupported variable type provided: %s",
-                            variable.variable_type,
-                        )
+                        elif variable.variable_type == "table":
+                            table_rep = ()
+                            for col in variable.columns:
 
-                    if variable.name in self._input_variables:
-                        handler = PVAccessInputHandler(
-                            pvname=pvname, is_constant=variable.is_constant, server=self
-                        )
+                                # here we assume double type in tables...
+                                table_rep += (col, "ad")
 
-                        pv = SharedPV(handler=handler, nt=nt, initial=initial)
+                            nt = NTTable(table_rep)
+                            initial = nt.wrap(variable.value)
 
-                    else:
-                        pv = SharedPV(nt=nt, initial=initial)
+                        elif variable.variable_type == "array":
+                            if variable.value_type == "str":
+                                nt = NTScalar("as")
+                                initial = variable.value
+
+                            else:
+                                nd_array = variable.value.view(NTNDArrayData)
+                                nt = NTNDArray()
+                                initial = nd_array
+
+                        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,
+                            )
 
-                    self._providers[pvname] = pv
+                            pv = SharedPV(handler=handler, nt=nt, initial=initial)
+
+                        else:
+                            pv = SharedPV(nt=nt, initial=initial)
+
+                        self._providers[pvname] = pv
 
                 # if not serving pv, set up monitor
                 else:
@@ -285,6 +368,41 @@ def setup_server(self) -> None:
                     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])