HLA-825 Make BPM records from csv

Created new func in create_csv which adds BBA pvs to the new bba.csv file, previously some of these PVs were defined in feedback.csv and some at runtime in atip_server/_create_feedback_records. Removed BBA special case record creation from atip_server/_create_feedback_records and instead load them from csv in _create_bba_records. Reran creation of csv files for DIAD and I04

Author: ptsOSL

src/virtac/atip_ioc_entry.py

@@ -62,6 +62,7 @@ def main():
     server = atip_server.ATIPServer(
         ring_mode,
         DATADIR / ring_mode / "limits.csv",
+        DATADIR / ring_mode / "bba.csv",
         DATADIR / ring_mode / "feedback.csv",
         DATADIR / ring_mode / "mirrored.csv",
         DATADIR / ring_mode / "tunefb.csv",

src/virtac/atip_server.py

@@ -1,4 +1,3 @@
-import ast
 import csv
 from warnings import warn
 
@@ -59,6 +58,7 @@ def __init__(
         self,
         ring_mode,
         limits_csv=None,
+        bba_csv=None,
         feedback_csv=None,
         mirror_csv=None,
         tune_csv=None,
@@ -70,7 +70,10 @@ def __init__(
             limits_csv (str): The filepath to the .csv file from which to
                                     load the pv limits, for more information
                                     see create_csv.py.
-            feedback_csv (str): The filepath to the .csv file from which to
+            bba_csv (str): The filepath to the .csv file from which to
+                                    load the bba records, for more
+                                    information see create_csv.py.
+            feedback_csv (string): The filepath to the .csv file from which to
                                     load the feedback records, for more
                                     information see create_csv.py.
             mirror_csv (str): The filepath to the .csv file from which to
@@ -88,12 +91,15 @@ def __init__(
         self._in_records = {}
         self._out_records = {}
         self._rb_only_records = []
+        self._bba_records = {}
         self._feedback_records = {}
         self._mirrored_records = {}
         self._monitored_pvs = {}
         self._offset_pvs = {}
         print("Starting record creation.")
         self._create_records(limits_csv, disable_emittance)
+        if bba_csv is not None:
+            self._create_bba_records(bba_csv)
         if feedback_csv is not None:
             self._create_feedback_records(feedback_csv, disable_emittance)
         if mirror_csv is not None:
@@ -114,6 +120,7 @@ def all_record_names(self):
             list(self._in_records.keys())
             + list(self._out_records.keys())
             + list(self._feedback_records.values())
+            + list(self._bba_records.values())
             + mirrored
         )
         return {record.name: record for record in all_records}
@@ -291,10 +298,47 @@ def _on_update(self, value, name):
                 field, value, units=pytac.ENG, data_source=pytac.SIM
             )
 
+    def _create_bba_records(self, bba_csv):
+        """Create all the beam-based-alignment records from the .csv file at the
+        location passed, see create_csv.py for more information.
+
+        Args:
+            bba_csv (str): The filepath to the .csv file to load the
+                                    records in accordance with.
+        """
+        # We don't set limits or precision but this shouldn't be an issue as these
+        # records aren't really intended to be set to by a user.
+        csv_reader = csv.DictReader(open(bba_csv))
+        for line in csv_reader:
+            prefix, suffix = line["pv"].split(":", 1)
+            builder.SetDeviceName(prefix)
+            if line["record_type"] == "ai":
+                record = builder.aIn(
+                    suffix, initial_value=int(line["value"]), MDEL="-1"
+                )
+                self._bba_records[(int(line["index"]), line["field"])] = record
+            elif line["record_type"] == "ao":
+                record = builder.aOut(
+                    suffix, initial_value=int(line["value"]), always_update=True
+                )
+                self._bba_records[(int(line["index"]), line["field"])] = record
+            elif line["record_type"] == "wfm":
+                record = builder.WaveformOut(
+                    suffix,
+                    # We remove the [] around the string
+                    initial_value=numpy.fromstring(
+                        (line["value"])[1:-1], dtype=int, sep=" "
+                    ),
+                    always_update=True,
+                )
+                self._bba_records[(int(line["index"]), line["field"])] = record
+            else:
+                raise ValueError
+
     def _create_feedback_records(self, feedback_csv, disable_emittance):
         """Create all the feedback records from the .csv file at the location
-        passed, see create_csv.py for more information; records for two edge
-        cases are also created.
+        passed, see create_csv.py for more information; records for one edge
+        case are also created.
 
         Args:
             feedback_csv (str): The filepath to the .csv file to load the
@@ -306,25 +350,25 @@ def _create_feedback_records(self, feedback_csv, disable_emittance):
         # records aren't really intended to be set to by a user.
         csv_reader = csv.DictReader(open(feedback_csv))
         for line in csv_reader:
-            try:
-                readonly = ast.literal_eval(line["read-only"])
-                assert isinstance(readonly, bool)
-            except (ValueError, AssertionError) as exc:
-                raise ValueError(
-                    f"Unable to evaluate {line['read-only']} as a boolean."
-                ) from exc
             prefix, suffix = line["pv"].split(":", 1)
             builder.SetDeviceName(prefix)
-            if readonly:
-                in_record = builder.aIn(
+            if line["record_type"] == "ai":
+                record = builder.aIn(
                     suffix, initial_value=int(line["value"]), MDEL="-1"
                 )
-                self._feedback_records[(int(line["index"]), line["field"])] = in_record
-            else:
-                out_record = builder.aOut(
+                self._feedback_records[(int(line["index"]), line["field"])] = record
+            elif line["record_type"] == "ao":
+                record = builder.aOut(
+                    suffix, initial_value=int(line["value"]), always_update=True
+                )
+                self._feedback_records[(int(line["index"]), line["field"])] = record
+            elif line["record_type"] == "wfm":
+                record = builder.WaveformOut(
                     suffix, initial_value=int(line["value"]), always_update=True
                 )
-                self._feedback_records[(int(line["index"]), line["field"])] = out_record
+                self._feedback_records[(int(line["index"]), line["field"])] = record
+            else:
+                raise ValueError
         # Special case: BPM ID for the x axis of beam position plot, since we
         # cannot currently create Waveform records via CSV.
         bpm_ids = [
@@ -336,28 +380,7 @@ def _create_feedback_records(self, feedback_csv, disable_emittance):
             "BPMID", NELM=len(bpm_ids), initial_value=bpm_ids
         )
         self._feedback_records[(0, "bpm_id")] = bpm_id_record
-        # Special case: Fast BBA oscillation PVs - should also be moved to CSV in future
-        for cell in range(1, self.lattice.symmetry + 1):
-            cell = str(cell).zfill(2)
-            builder.SetDeviceName(f"SR{cell}A-CS-FOFB-01")
-            start_times = builder.WaveformOut(
-                "EXCITE:START_TIMES", initial_value=numpy.zeros(18)
-            )
-            self._feedback_records[(0, f"cell_{cell}_excite_start_times")] = start_times
-            excite_amps = builder.WaveformOut(
-                "EXCITE:AMPS", initial_value=numpy.zeros(18)
-            )
-            self._feedback_records[(0, f"cell_{cell}_excite_amps")] = excite_amps
-            excite_deltas = builder.WaveformOut(
-                "EXCITE:DELTAS", initial_value=numpy.zeros(18)
-            )
-            self._feedback_records[(0, f"cell_{cell}_excite_deltas")] = excite_deltas
-            excite_ticks = builder.WaveformOut(
-                "EXCITE:TICKS", initial_value=numpy.zeros(18)
-            )
-            self._feedback_records[(0, f"cell_{cell}_excite_ticks")] = excite_ticks
-            excite_prime = builder.aOut("EXCITE:PRIME", initial_value=1)
-            self._feedback_records[(0, f"cell_{cell}_excite_prime")] = excite_prime
+
         # We can choose to not calculate emittance as it is not always required,
         # which decreases computation time.
         if not disable_emittance:
@@ -546,13 +569,14 @@ def set_feedback_record(self, index, field, value):
         """
         try:
             self._feedback_records[(index, field)].set(value)
+            self._bba_records[(index, field)].set(value)
         except KeyError as exc:
             if index == 0:
                 raise FieldException(
                     f"Simulated lattice {self.lattice} does not have field {field}."
                 ) from exc
             else:
                 raise FieldException(
-                    f"Simulated element {self.lattice[index]} does not have "
-                    f"field {field}."
+                    f"Simulated element {self.lattice[index]} does not have \
+                    field {field}."
                 ) from exc

src/virtac/create_csv.py

@@ -6,6 +6,7 @@
 import csv
 import os
 
+import numpy as np
 import pytac
 from cothread.catools import FORMAT_CTRL, caget
 
@@ -23,54 +24,104 @@ def generate_feedback_pvs(all_elements):
         + all_elements.q1b
     )
     # Data to be written is stored as a list of tuples each with structure:
-    #     element index (int), field (str), pv (str), value (int).
+    #     element index (int), field (str), pv (str), value (int), record_type (str).
     # We have special cases for four lattice fields that feedback systems read from.
     data = [
-        ("index", "field", "pv", "value", "read-only"),
-        (0, "beam_current", "SR-DI-DCCT-01:SIGNAL", 300, True),
-        (0, "feedback_status", "CS-CS-MSTAT-01:FBSTAT", 2, True),
-        (0, "fofb_status", "SR01A-CS-FOFB-01:RUN", 0, False),
-        (0, "feedback_heart", "CS-CS-MSTAT-01:FBHEART", 10, False),
+        ("index", "field", "pv", "value", "record_type"),
+        (0, "beam_current", "SR-DI-DCCT-01:SIGNAL", 300, "ai"),
+        (0, "feedback_status", "CS-CS-MSTAT-01:FBSTAT", 2, "ai"),
+        (0, "fofb_status", "SR01A-CS-FOFB-01:RUN", 0, "ao"),
+        (0, "feedback_heart", "CS-CS-MSTAT-01:FBHEART", 10, "ao"),
     ]
     # Iterate over our elements to get the PV names.
     for elem in all_elements.hstr:
         pv_stem = elem.get_device("x_kick").name
-        data.append((elem.index, "error_sum", pv_stem + ":ERCSUM", 0, True))
-        data.append((elem.index, "state", pv_stem + ":STATE", 2, True))
+        data.append((elem.index, "error_sum", pv_stem + ":ERCSUM", 0, "ai"))
+        data.append((elem.index, "state", pv_stem + ":STATE", 2, "ai"))
     for elem in all_elements.vstr:
         pv_stem = elem.get_device("y_kick").name
-        data.append((elem.index, "error_sum", pv_stem + ":ERCSUM", 0, True))
-        data.append((elem.index, "state", pv_stem + ":STATE", 2, True))
+        data.append((elem.index, "error_sum", pv_stem + ":ERCSUM", 0, "ai"))
+        data.append((elem.index, "state", pv_stem + ":STATE", 2, "ai"))
     for elem in all_elements.bpm:
         data.append(
-            (elem.index, "enabled", elem.get_pv_name("enabled", pytac.RB), 1, True)
+            (elem.index, "enabled", elem.get_pv_name("enabled", pytac.RB), 1, "ai")
         )
     # Add elements for Tune Feedback
     for elem in tune_quad_elements:
         data.append(
-            (elem.index, "offset", elem.get_device("b1").name + ":OFFSET1", 0, True)
+            (elem.index, "offset", elem.get_device("b1").name + ":OFFSET1", 0, "ai")
         )
     return data
 
 
-def generate_bba_pvs(all_elements):
+def generate_bba_pvs(all_elements, symmetry):
     """Data to be written is stored as a list of tuples each with structure:
-    element index (int), field (str), pv (str), value (int).
+    element index (int), field (str), pv (str), value (int), record_type (str).
     """
-    data = [("index", "field", "pv", "value", "read-only")]
+    data = [("index", "field", "pv", "value", "record_type")]
     # Iterate over the BPMs to construct the PV names.
     for elem in all_elements.bpm:
         pv_stem = elem.get_device("enabled").name
         data.append(
-            (elem.index, "golden_offset_x", pv_stem + ":CF:GOLDEN_X_S", 0, False)
+            (elem.index, "golden_offset_x", pv_stem + ":CF:GOLDEN_X_S", 0, "ao")
         )
         data.append(
-            (elem.index, "golden_offset_y", pv_stem + ":CF:GOLDEN_Y_S", 0, False)
+            (elem.index, "golden_offset_y", pv_stem + ":CF:GOLDEN_Y_S", 0, "ao")
+        )
+        data.append((elem.index, "bcd_offset_x", pv_stem + ":CF:BCD_X_S", 0, "ao"))
+        data.append((elem.index, "bcd_offset_y", pv_stem + ":CF:BCD_Y_S", 0, "ao"))
+        data.append((elem.index, "bba_offset_x", pv_stem + ":CF:BBA_X_S", 0, "ao"))
+        data.append((elem.index, "bba_offset_y", pv_stem + ":CF:BBA_Y_S", 0, "ao"))
+    for cell in range(1, symmetry + 1):
+        cell = str(cell).zfill(2)
+        pv_stem = f"SR{cell}A-CS-FOFB-01"
+        # Waveform records
+        data.append(
+            (
+                cell,
+                f"cell_{cell}_excite_start_times",
+                f"{pv_stem}:EXCITE:START_TIMES",
+                np.zeros(18),
+                "wfm",
+            )
+        )
+        data.append(
+            (
+                cell,
+                f"cell_{cell}_excite_amps",
+                f"{pv_stem}:EXCITE:AMPS",
+                np.zeros(18),
+                "wfm",
+            )
+        )
+        data.append(
+            (
+                cell,
+                f"cell_{cell}_excite_deltas",
+                f"{pv_stem}:EXCITE:DELTAS",
+                np.zeros(18),
+                "wfm",
+            )
+        )
+        data.append(
+            (
+                cell,
+                f"cell_{cell}_excite_ticks",
+                f"{pv_stem}:EXCITE:TICKS",
+                np.zeros(18),
+                "wfm",
+            )
+        )
+        # ao record
+        data.append(
+            (
+                cell,
+                f"cell_{cell}_excite_prime",
+                f"{pv_stem}:EXCITE:PRIME",
+                0,
+                "ao",
+            )
         )
-        data.append((elem.index, "bcd_offset_x", pv_stem + ":CF:BCD_X_S", 0, False))
-        data.append((elem.index, "bcd_offset_y", pv_stem + ":CF:BCD_Y_S", 0, False))
-        data.append((elem.index, "bba_offset_x", pv_stem + ":CF:BBA_X_S", 0, False))
-        data.append((elem.index, "bba_offset_y", pv_stem + ":CF:BBA_Y_S", 0, False))
     return data
 
 
@@ -136,7 +187,9 @@ def generate_mirrored_pvs(lattice):
     value:
         The inital value of the output record.
     """
-    data = [("output type", "mirror type", "in", "out", "value")]
+    data: list[tuple[str, str, str, str, int]] = [
+        ("output type", "mirror type", "in", "out", "value")
+    ]
     # Tune PV aliases.
     tune = [
         lattice.get_value("tune_x", pytac.RB, data_source=pytac.SIM),
@@ -308,6 +361,11 @@ def parse_arguments():
         help="Filename for output tune feedback offset PVs CSV file",
         default="tunefb.csv",
     )
+    parser.add_argument(
+        "--bba",
+        help="Filename for output beam-based-alignment PVs CSV file",
+        default="bba.csv",
+    )
     return parser.parse_args()
 
 
@@ -316,8 +374,9 @@ def parse_arguments():
     lattice = atip.utils.loader(args.ring_mode)
     all_elements = atip.utils.preload(lattice)
     data = generate_feedback_pvs(all_elements)
-    data.extend(generate_bba_pvs(all_elements)[1:])
     write_data_to_file(data, args.feedback, args.ring_mode)
+    data = generate_bba_pvs(all_elements, lattice.symmetry)
+    write_data_to_file(data, args.bba, args.ring_mode)
     data = generate_pv_limits(lattice)
     write_data_to_file(data, args.limits, args.ring_mode)
     data = generate_mirrored_pvs(lattice)