SQL == CSV For YAML Generation

slac_db/controls_information.py

@@ -25,15 +25,7 @@ def get_screen_controls_information(screen_information: Dict = None):
     # }
 
     # Stuff like Device-Position mappings for motor/ladder-based screens
-    controls_information = {}
-    for k, v in screen_information.items():
-        pv_cache = {}
-        pvs = v["controls_information"]["PVs"]
-        if "orient_x" in pvs and "orient_y" in pvs:
-            pv_cache["orient_x"] = caget(pvs["orient_x"], as_string=True)
-            pv_cache["orient_y"] = caget(pvs["orient_y"], as_string=True)
-        controls_information[k] = {"pv_cache": pv_cache}
-    return controls_information
+    return {}
 
 
 def get_wire_controls_information(wire_names: List[str] = None):

slac_db/create/combined.py

@@ -3,7 +3,7 @@
 import slac_db.io
 import slac_db.oracle
 import slac_db.device
-from slac_db.metadata import get_wire_metadata
+from slac_db.metadata import get_wire_metadata, get_pmt_metadata
 from pykern.pkcollections import PKDict
 import yaml
 

slac_db/db_to_yaml.py

@@ -43,8 +43,8 @@ def _round_values(meta):
 
     beampath_csv = slac_db.oracle.get_device_row(device_name)["beampath"]
     rv = {
-        "area": slac_db.device.get_attribute(device_name, "area"),
         "beam_path": list(parse_beampaths(beampath_csv)),
+        "area": slac_db.device.get_attribute(device_name, "area"),
         "type": slac_db.device.get_attribute(device_name, "device_type"),
     }
     rv.update(_round_values(slac_db.device.get_all_meta(device_name)))
@@ -66,9 +66,10 @@ def _build_controls_information(device_name):
     Args:
         device_name (str): MAD Device Name
     """
+    pvs = slac_db.device.get_all_accessors(device_name)
     return {
-        "PVs": slac_db.device.get_all_accessors(device_name),
         "control_name": slac_db.device.get_attribute(device_name, "cs_name"),
+        "PVs": pvs,
     }
 
 
@@ -164,10 +165,10 @@ def build():
     def _parse_areas():
         areas = slac_db.device.get_all_areas()
         for a in areas:
-            if " " in a or "*" in a:
+            if "NO AREA" in a or "*" in a:
                 continue
             yield a
 
     areas = list(_parse_areas())
-    out = {a: d for a, d in _build_areas(areas)}
+    out = {a: d for a, d in _build_areas(areas) if d != {}}
     return out

slac_db/directory_service.py

@@ -7,6 +7,27 @@
 
 _meta = None
 
+def verify_address(address):
+    """Verify that an address exists.
+
+    Args:
+        address (str): EPICS PV address
+
+    Returns:
+        Address if address exists.
+        None if it does not.
+    """
+    with _session() as s:
+        return list(sorted(
+            r["address"] for r in s.select(
+                sqlalchemy.select(
+                    s.t.addresses.c["address"]
+                ).where(
+                    s.t.addresses.c["address"] == address
+                )
+            )
+        ))
+
 def get_addresses(device=None):
     """Get all addresses per device.
 

slac_db/generate.py

@@ -4,6 +4,7 @@
 from typing import Any, Union, List, Dict, Optional
 import numpy as np
 import slac_db.config
+import slac_db.directory_service
 from slac_db.metadata import (
     get_magnet_metadata,
     get_screen_metadata,
@@ -31,6 +32,8 @@ def __init__(
         self,
         csv_location=slac_db.config.package_data() / "lcls_elements.csv",
         filter_location=slac_db.config.package_data() / "filter.yaml",
+        use_meme=True,
+        suppress_warnings=False,
     ):
         self.csv_location = csv_location
         self.filter_location = filter_location
@@ -45,6 +48,8 @@ def __init__(
             "SumL (m)",
         ]
         self.elements = self._filter_elements_by_fields(self._required_fields)
+        self.use_meme = use_meme
+        self.suppress_warnings = suppress_warnings
         self._areas = self.extract_areas()
         self._beam_paths = self.extract_beampaths()
 
@@ -142,7 +147,7 @@ def _construct_information_from_element(
             },
             "metadata": {
                 "beam_path": [
-                    item.strip() for item in element["Beampath"].split(",") if item
+                    item.strip() for item in element["Beampath"].split(",") if item.strip()
                 ],
                 "area": element["Area"],
                 "type": element["Keyword"],
@@ -153,6 +158,7 @@ def _construct_information_from_element(
                 ),
             },
         }
+        print(device_information["metadata"]["beam_path"])
         [
             device_information["metadata"].update({field_name: field_value})
             for field_name, field_value in additional_metadata_fields.items()
@@ -168,7 +174,12 @@ def _construct_information_from_element(
     def _construct_pv_list_from_control_system_name(
         self, name, search_with_handles: Optional[Dict[str, str]]
     ) -> Dict[str, str]:
-        from meme import names
+
+        if self.use_meme:
+            from meme import names
+            query_pv = names.list_pvs
+        else:
+            query_pv = slac_db.directory_service.verify_address
 
         if name == "":
             raise RuntimeError("No control system name provided for meme search.")
@@ -180,7 +191,7 @@ def _construct_pv_list_from_control_system_name(
                 search_term, field = search_term.split(".")
             # End of the PV name is implied in search_term
             try:
-                pv_list = names.list_pvs(name + ":" + search_term, sort_by="z")
+                pv_list = query_pv(name + ":" + search_term)
                 # We expect to have ZERO or ONE result returned from meme
                 if pv_list != list():
                     if len(pv_list) == 1:
@@ -239,7 +250,8 @@ def extract_devices(
             ]
         # Must have passed an area that does not exist or we don't have that device in this area!
         if len(device_elements) < 1:
-            print(f"No devices of types {required_types} found in area {area}")
+            if not self.suppress_warnings:
+                print(f"No devices of types {required_types} found in area {area}")
             return
         # Fill in the dict that will become the yaml file
         for device in device_elements:
@@ -274,7 +286,8 @@ def add_to_device_metadata(
             try:
                 device_data[device]["metadata"].update(additional_metadata[device])
             except KeyError:
-                print("No additional metadata found for ", device)
+                if not self.suppress_warnings:
+                    print("No additional metadata found for ", device)
 
         return device_data
 
@@ -289,7 +302,8 @@ def add_to_device_controls_information(
                     additional_controls_information[device]
                 )
             except KeyError:
-                print("No additional controls information found for ", device)
+                if not self.suppress_warnings:
+                    print("No additional controls information found for ", device)
         return device_data
 
     def add_extra_data_to_device(

slac_db/metadata.py

@@ -42,25 +42,7 @@ def get_screen_metadata(basic_screen_data: dict):
     #  scr-name-2 : {metadata-field-1 : value-1, metadata-field-2 : value-2},
     #  ...
     # }
-    from meme.names import list_pvs
-
-    metadata = {}
-    for mad_name, info in basic_screen_data.items():
-        metadata[mad_name] = {}
-        ctrl_name = info["controls_information"]["control_name"]
-        flags = list_pvs(ctrl_name + "%INSTALLED")
-        hardware = {}
-        for i in flags:
-            name = re.search("(?<=^" + ctrl_name + ":).*(?=INSTALLED)", i)
-            if name is None:
-                continue
-            name = name.group(0)
-            status = caget(i)
-            if status is not None:
-                hardware[name] = status
-
-        metadata[mad_name]["hardware"] = hardware
-    return metadata
+    return {}
 
 
 def get_wire_metadata(basic_wire_data: dict) -> Dict[str, Dict[str, Any]]:
@@ -163,15 +145,18 @@ def get_pmt_metadata(pmt_names: List[str] = []):
     #  pmt-name-2 : {metadata-field-1 : value-1, metadata-field-2 : value-2},
     #  ...
     # }
-    pmt_metadata = {}
+    return {}
 
-    here = slac_db.config.package_data()
-    yaml_path = os.path.join(here, "pmt_metadata.yaml")
+    # EY 08/18/2026 - Removed because of Oracle update.
+    # pmt_metadata = {}
+
+    # here = slac_db.config.package_data()
+    # yaml_path = os.path.join(here, "pmt_metadata.yaml")
 
-    with open(yaml_path, "r") as f:
-        pmt_metadata = yaml.safe_load(f)
+    # with open(yaml_path, "r") as f:
+    #     pmt_metadata = yaml.safe_load(f)
 
-    return pmt_metadata
+    # return pmt_metadata
 
 
 def get_toroid_metadata(toroid_names: List[str] = []):