Sql interface lfs

.gitattributes

@@ -1,4 +1,4 @@
-# slac_db/yaml/* linguist-generated
-# slac_db/package_data/* linguist-generated
-# tests/test_data/* linguist-generated
-# tests/test_data/lcls-tools-yaml/* linguist-generated
+slac_db/yaml/* linguist-generated
+slac_db/package_data/* linguist-generated
+tests/test_data/* linguist-generated
+*.sqlite3 filter=lfs diff=lfs merge=lfs -text

.github/workflows/tests.yml

@@ -22,6 +22,8 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
+      with:
+        lfs: true
     - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v5
       with:
@@ -36,4 +38,4 @@ jobs:
       run: |
         echo -e '## Test results\n\n```' >> "$GITHUB_STEP_SUMMARY"
         python -m unittest discover -s tests -v 2>&1 | tee -a "$GITHUB_STEP_SUMMARY"
-        echo '```' >> "$GITHUB_STEP_SUMMARY"
+        echo '```' >> "$GITHUB_STEP_SUMMARY"

pyproject.toml

@@ -15,8 +15,9 @@ classifiers = [
 ]
 dependencies = [
     "numpy",
-    "pyyaml",
     "pyepics",
+    "pykern",
+    "pyyaml",
 ]
 description = "Tools to support high level application development at LCLS using Python"
 dynamic = ["version"]

slac_db/__init__.py

@@ -18,7 +18,7 @@ def _flatten(nested_list):
             return _flatten(nested_list[0]) + _flatten(nested_list[1:])
         return nested_list[:1] + _flatten(nested_list[1:])
     beampath_definition_file = os.path.join(
-        slac_db.package_data(), "beampaths.yaml"
+        slac_db.config.package_data(), "beampaths.yaml"
     )
     with open(beampath_definition_file, "r") as file:
         beampath_definitions = yaml.safe_load(file)

slac_db/create/__init__.py

@@ -0,0 +1,8 @@
+import slac_db.create.meme_names
+import slac_db.create.lcls_elements_csv
+
+def oracle_db(csv_source=None):
+    slac_db.create.lcls_elements_csv.to_oracle_db(csv_source)
+
+def aida_db():
+    slac_db.create.meme_names.to_aida_db()

slac_db/create/lcls_elements_csv.py

@@ -0,0 +1,26 @@
+import csv
+import slac_db.config
+import slac_db.oracle
+
+def to_oracle_db(csv_source=None):
+    p = _Parser(csv_source=csv_source)
+    return slac_db.oracle.recreate(p)
+
+class _Parser():
+    def __init__(self, csv_source=None):
+        if not csv_source:
+            csv_source = (
+                slac_db.config.package_data() / "lcls_elements.csv"
+            )
+        self.rows = {}
+        with open(csv_source, "r") as c:
+            reader = csv.reader(c)
+            self._parse_csv(reader)
+
+    def _parse_csv(self, reader):
+        names = [r.lower() for r in next(reader)]
+        i = 0
+        for row in reader:
+            values = [None if v == '' else v for v in row]
+            self.rows[i] =  dict(zip(names, values))
+            i += 1

slac_db/create/meme_names.py

@@ -0,0 +1,15 @@
+import slac_db.directory_service
+
+def to_directory_service_db():
+    return slac_db.directory_service.recreate(_Parser())
+
+class _Parser:
+    def __init__(self):
+        self.addresses = set()
+        self._get_from_meme()
+
+    def _get_from_meme(self):
+        import meme.names
+        address_list = meme.names.list_pvs("%", timeout=600)
+        for a in address_list:
+            self.addresses.add(a)

slac_db/directory_service.py

@@ -0,0 +1,100 @@
+import os
+import os.path
+import slac_db.config
+import sqlalchemy
+import pykern.sql_db
+import slac_db.oracle
+
+_meta = None
+
+def get_addresses(device=None):
+    """Get all addresses per device.
+
+    Args:
+        device (str): MAD name of the device as found in Oracle.
+
+    Returns:
+        tuple: Sorted address values.
+    """
+    head = slac_db.oracle.get_address_header(device=device)
+    with _session() as s:
+        cs_address = s.t.addresses.c["address"]
+        return tuple(sorted(
+            r["address"] for r in s.select(
+                sqlalchemy.select(
+                    cs_address
+                ).where(
+                    cs_address.like(f"{head}%")
+                )
+            )
+        ))
+
+def recreate(parser):
+    """Rebuild the local directory_service sqlite3 database
+    only if it is not already loaded.
+
+    Args:
+        parser: Container for column data.
+    """
+    assert not _meta
+    assert parser.addresses
+    if os.path.exists(_directory_service_uri()):
+        os.remove(_directory_service_uri())
+    _Inserter(parser)
+
+class _Inserter:
+    """Creates a session and commits rows to the db.
+
+    Functions:
+        _addresses: Inserts all addresses in parser.addresses.
+    """
+    def __init__(self, parser):
+        self.counts = {"addresses": 0}
+        with _session() as s:
+            self._addresses(parser.addresses, s)
+
+    def _addresses(self, addresses, session):
+        # We have to do it this way unfortunately.
+        # Bulk insert is not faster.
+        n = len(addresses)
+        i = 0
+        for a in addresses:
+            session.insert("addresses", address=a)
+            i += 1
+            print("{i} / {n}", end='\r')
+
+def _db_type_prefix(uri):
+    if not uri.startswith("sqlite"):
+        uri = 'sqlite:///' + uri
+    return uri
+
+def _init_db(uri=None):
+    """Initializes pykern sqlalchemy wrapper. Initialization
+    occurs when a session is first created.
+
+       _meta: wrapper that holds sqlalchemy metadata.
+    """
+    global _meta
+    if uri is None:
+        uri = _directory_service_uri()
+    uri = _db_type_prefix(uri)
+    schema = {
+        "addresses": {
+            "address": "str 64 primary_key",
+        }
+    }
+    _meta = pykern.sql_db.Meta(
+        uri=uri,
+        schema=schema
+    )
+
+def _directory_service_uri():
+    uri = (
+        slac_db.config.package_data() / 'directory_service_pvs.sqlite3'
+    )
+    return str(uri)
+
+def _session():
+    if _meta is None:
+        _init_db()
+    return _meta.session()

slac_db/oracle.py

@@ -0,0 +1,191 @@
+import slac_db.config
+import sqlalchemy
+import pykern.sql_db
+import os.path
+import os
+
+
+_meta = None
+
+def get_address_header(device=None):
+    """Get address header of a device.
+
+    Args:
+        device (str): MAD name of the device as found in Oracle.
+
+    Returns:
+        tuple: The address header.
+    """
+    with _session() as s:
+        return s.select_one(
+            sqlalchemy.select(
+                s.t.elements.c["control system name"]
+            ).where(
+                s.t.elements.c["element"] == device
+            )
+        )["control system name"]
+
+def get_devices(area=None, device_type=None):
+    """Get devices of one type from an area.
+
+    Args:
+        area (str): Name of the accelerator area.
+        device_type (str): Type of device as listed in Oracle.
+
+    Returns:
+        tuple: Device names in Z order.
+    """
+    if device_type is None:
+        device_type = "%"
+    with _session() as s:
+        return tuple(
+            r.element for r in s.select(
+                sqlalchemy.select(
+                    s.t.elements.c["element"]
+                ).where(
+                    s.t.elements.c["keyword"] == device_type
+                ).where(
+                    s.t.elements.c["area"] == area
+                ).order_by(s.t.elements.c["suml (m)"])
+            )
+        )
+
+def get_device_row(element=None):
+    """Get the full row for an element.
+
+    Args:
+        element: name of the element to get.
+    Returns:
+        sql alchemy row: Row object with each column.
+    """
+    with _session() as s:
+        return s.select_one(
+            sqlalchemy.select(
+                s.t.elements
+            ).where(
+                s.t.elements.c["element"] == element
+            )
+        )
+
+def get_beampaths():
+    """Get all beampaths from Oracle.
+
+    Returns:
+        List of beampaths sorted alphabetically.
+    """
+    beampaths = set()
+    def parse_beampaths(beampath_csv):
+        if beampath_csv is None:
+            return
+        c = beampath_csv.replace(' ', '').split(',')
+        c = filter(None, c)
+        beampaths.update(c)
+
+    with _session() as s:
+        query = sqlalchemy.select(s.t.elements.c.beampath).distinct()
+        for r in s.select(query):
+            parse_beampaths(r.beampath)
+    return sorted(list(beampaths))
+
+
+def get_areas():
+    """Get all areas from Oracle.
+
+    Returns:
+        List of areas sorted alphabetically.
+    """
+    def exclude_bad_patterns(column):
+        bad_patterns = ['\t- NO AREA -', '*%']
+        filters = [None] * len(bad_patterns)
+        for i in range(0, len(bad_patterns)):
+            filters[i] = column.not_like(bad_patterns[i])
+        return sqlalchemy.and_(*filters)
+
+    with _session() as s:
+        return list(
+            r.area for r in s.select(
+                sqlalchemy.select(
+                    s.t.elements.c.area
+                ).where(
+                    exclude_bad_patterns(s.t.elements.c.area)
+                ).distinct()
+            )
+        )
+
+def recreate(parser):
+    """Rebuilds the sqlite copy of Oracle.
+    Fails if a connection has already been made.
+
+    Args:
+        Parser object with attribute 'rows' for row data.
+    """
+    if _meta:
+        raise AssertionError(
+            "Database connnection already initialized. "
+            + "Restart Python interpreter."
+        )
+    if not hasattr(parser, "rows"):
+        raise AssertionError(
+            "Parser is missing attribute 'rows'. "
+        )        
+    if os.path.exists(_oracle_uri()):
+        os.remove(_oracle_uri())
+    _Inserter(parser)
+
+
+class _Inserter:
+    """Inserts rows into sqllite database.
+    """
+    def __init__(self, parser):
+        with _session() as s:
+            self._rows(parser.rows, s)
+    def _rows(self, rows, session):
+        for r in rows.values():
+            ins = {}
+            for c in session.t.elements.c:
+                ins[c.name] = r[c.name]
+            session.insert("elements", **ins)
+
+
+def _db_type_prefix(uri):
+    if not uri.startswith("sqlite"):
+        uri = 'sqlite:///' + uri
+    return uri
+
+def _init_db(uri=None):
+    """Initializes pykern sqlalchemy wrapper. Initialization
+    occurs when a session is first created.
+
+       _meta: wrapper that holds sqlalchemy metadata.
+    """
+    global _meta
+    if uri is None:
+        uri = _oracle_uri()
+    uri = _db_type_prefix(uri)
+    schema = {
+        "elements": {
+            "Area": "str 64 nullable",
+            "Element": "str 64 primary_key",
+            "Control System Name": "str 64 nullable",
+            "Keyword": "str 64 nullable",
+            "Beampath": "str 64 nullable",
+            "SumL (m)": "float 64 nullable",
+            "Effective Length (m)": "float 64 nullable",
+            "Rf Frequency (MHz)": "float 64 nullable"
+        }
+    }
+    _meta = pykern.sql_db.Meta(
+        uri=uri,
+        schema=schema
+    )
+
+def _oracle_uri():
+    uri = (
+        slac_db.config.package_data() / 'lcls_elements.sqlite3'
+    )
+    return str(uri)
+
+def _session():
+    if _meta is None:
+        _init_db()
+    return _meta.session()