Convert CubeMX and Headers into Knowledge Graphs using Kuzu

.github/workflows/test-kg.yml

@@ -0,0 +1,88 @@
+name: Test Generate Knowledge Graphs
+
+on:
+  workflow_dispatch:
+  pull_request:
+
+jobs:
+  test-kg-stmicro:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        family_group:
+          - name: 'L1_4_U0'
+            families: [STM32L1, STM32L4, STM32U0]
+          - name: 'F0_4_H5_WL'
+            families: [STM32F0, STM32F4, STM32H5, STM32WL]
+          - name: 'G4_F7'
+            families: [STM32G4, STM32F7]
+          - name: 'G0_L0_U5'
+            families: [STM32G0, STM32L0, STM32U5]
+          - name: 'C0_F1_2_3_L5_U3_WB'
+            families: [STM32C0, STM32F1, STM32F2, STM32F3, STM32L5, STM32U3, STM32WB]
+          - name: 'H7'
+            families: [STM32H7]
+
+    name: Update KG ${{ matrix.family_group.name }}
+
+    steps:
+      - name: Check out Repository
+        uses: actions/checkout@v4
+
+      - name: Setup Python 3.12
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.12"
+
+      - name: Install Python Dependencies
+        run: |
+          pip3 install -r tools/requirements.txt -e .
+
+      - name: Clone STMicro CubeMX Archive
+        uses: actions/checkout@v4
+        with:
+          repository: modm-ext/archive-stmicro-cubemx
+          path: ext/stmicro/cubemx
+          ssh-key: ${{ secrets.SSH_KEY_STMICRO_CUBEMX }}
+
+      - name: Clone STMicro Sources
+        run: |
+          make ext/arm/cmsis/ ext/stmicro/cubehal/ ext/stmicro/header/
+
+      - name: Generate Database for ${{ matrix.family_group.name }}
+        run: |
+          FAMILY_LIST='${{ toJson(matrix.family_group.families) }}'
+          OVERALL_STATUS=0
+          set +e
+
+          for FAMILY in $(echo "$FAMILY_LIST" | jq -r '.[]')
+          do
+            echo "::group::Generating database for $FAMILY"
+
+            python3 -m modm_data.cube2kg --family $FAMILY
+            if [ $? -ne 0 ]; then
+                echo "::error::Database generation for $FAMILY FAILED."
+                OVERALL_STATUS=1
+            fi
+
+            echo "::endgroup::"
+          done
+
+          exit $OVERALL_STATUS
+
+      - name: Upload KG Logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: logs-${{ matrix.family_group.name }}
+          path: log/stmicro/kg
+          retention-days: 7
+
+      - name: Upload KGs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: kg-${{ matrix.family_group.name }}
+          path: ext/stmicro/kg-archive
+          retention-days: 7

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
     "CppHeaderParser>=2.7.4,<3",
     "jinja2>=3.1.3,<4",
     "lxml>=5.2.0,<6",
-    "owlready2==0.45",
+    "kuzu>=0.11.3,<0.12",
     "pypdfium2==4.30.0",
     "tqdm>=4.66.2,<5",
 ]

src/modm_data/__init__.py

@@ -21,7 +21,7 @@
     "html",
     "html2owl",
     "html2svd",
-    "owl",
+    "kg",
     "pdf",
     "pdf2html",
     "svd",

src/modm_data/cube2owl/__main__.py → src/modm_data/cube2kg/__main__.py

@@ -8,10 +8,7 @@
 from multiprocessing.pool import ThreadPool
 
 from modm_data.cubemx import devices_from_prefix, devices_from_partname
-from modm_data.header2svd.stmicro import Header
-from modm_data.html.stmicro import datasheet_for_device, reference_manual_for_device
-from modm_data.owl.stmicro import create_ontology
-from modm_data.py2owl.stmicro import owl_from_did, owl_from_cubemx, owl_from_header, owl_from_doc
+from modm_data.kg.stmicro import kg_create, kg_from_cubemx
 
 
 def main():
@@ -22,13 +19,13 @@ def main():
 
     if args.family:
         partnames = devices_from_prefix(args.family.lower())
-        print(partnames)
         # cluster the parallel jobs only around family+name
         partnames = sorted(list(set(p[:9] for p in partnames)))
+        print(" ".join(partnames))
 
-        Path("log/stmicro/owl").mkdir(exist_ok=True, parents=True)
+        Path("log/stmicro/kg").mkdir(exist_ok=True, parents=True)
         calls = [
-            f"python3 -m modm_data.cube2owl --prefix {partname} > log/stmicro/owl/cubemx_{partname}.txt 2>&1"
+            f"python3 -m modm_data.cube2kg --prefix {partname} > log/stmicro/kg/cube2kg_{partname}.txt 2>&1"
             for partname in partnames
         ]
         with ThreadPool() as pool:
@@ -39,29 +36,17 @@ def main():
         return all(r.returncode == 0 for r in retvals)
 
     for partname in devices_from_prefix(args.prefix.lower()):
-        ds, rm = None, None
         for device in devices_from_partname(partname):
             did = device["id"]
-            # Only change the documentation object if necessary to preserve caching
-            if ds != (nds := datasheet_for_device(did)):
-                ds = nds
-            if rm != (nrm := reference_manual_for_device(did)):
-                rm = nrm
-            print(did, ds, rm)
-            if ds is None or rm is None:
-                print(f"Ignoring {did} due to lack of documents")
-                continue
 
-            cmsis_header = Header(did)
-            onto = create_ontology(did)
+            db = kg_create(did)
+            kg_from_cubemx(db, device)
 
-            owl_from_did(onto)
-            owl_from_cubemx(onto, device)
-            owl_from_header(onto, cmsis_header)
-            owl_from_doc(onto, ds)
-            owl_from_doc(onto, rm)
+            # header = read_header(did)
+            # memory_map_from_header(header)
+            # kg_from_header(db, header)
 
-            onto.store.save(did.string)
+            db.save()
 
     return True
 

src/modm_data/cubehal/__init__.py

@@ -10,6 +10,6 @@
 - Determine the map of register bit field values to names.
 """
 
-from .dmamux_requests import read_request_map
+from .dmamux_requests import read_request_map, read_bdma_request_map
 
-__all__ = ["read_request_map"]
+__all__ = ["read_request_map", "read_bdma_request_map"]

src/modm_data/cubehal/dmamux_requests.py

@@ -4,11 +4,12 @@
 import re
 from pathlib import Path
 from ..utils import ext_path
-from ..owl import DeviceIdentifier
+from ..kg import DeviceIdentifier
 
 _CUBE_PATH = ext_path("stmicro/cubehal")
 _DMAMUX_PATTERN = re.compile(r"^\s*#define\s+(?P<name>(LL_DMAMUX_REQ_\w+))\s+(?P<id>(0x[0-9A-Fa-f]+))U")
 _REQUEST_PATTERN = re.compile(r"^\s*#define\s+(?P<name>(DMA_REQUEST_\w+))\s+(?P<id>([0-9]+))U")
+_BDMA_REQUEST_PATTERN = re.compile(r"^\s*#define\s+(?P<name>(BDMA_REQUEST_\w+))\s+(?P<id>([0-9]+))U")
 
 
 def read_request_map(did: DeviceIdentifier) -> dict[str, int]:
@@ -18,27 +19,34 @@ def read_request_map(did: DeviceIdentifier) -> dict[str, int]:
     :param did: Device to query for.
     :return: A dictionary of DMA trigger name to trigger position.
     """
-    dma_header = _get_hal_dma_header_path(did.family)
-    dmamux_header = _get_ll_dmamux_header_path(did.family)
+    dma_header = _get_hal_dma_header_path(did)
+    dmamux_header = _get_ll_dmamux_header_path(did)
     request_map = None
-    if did.family in ["g4", "h7", "l5"]:
-        request_map = _read_requests(dma_header)
-    elif did.family in ["g0", "wb", "wl"]:
+    if did.family in ["c0", "g4", "h7", "l5"]:
+        request_map = _read_requests(dma_header, _REQUEST_PATTERN)
+    elif did.family in ["g0", "u0", "wb", "wl"]:
         request_map = _read_requests_from_ll_dmamux(dma_header, dmamux_header)
     elif did.family == "l4" and did.name[0] in ["p", "q", "r", "s"]:
-        request_map = _read_requests_l4(did.name in ["p5", "q5"])
+        request_map = _read_requests_l4(did)
     else:
-        raise RuntimeError(f"No DMAMUX request data available for {did}")
-    _fix_request_data(request_map)
+        raise RuntimeError("No DMAMUX request data available for {}".format(did))
+    _fix_request_data(request_map, "DMA")
     return request_map
 
 
-def _fix_request_data(request_map):
+def read_bdma_request_map(did):
+    dma_header = _get_hal_dma_header_path(did)
+    request_map = _read_requests(dma_header, _BDMA_REQUEST_PATTERN)
+    _fix_request_data(request_map, "BDMA")
+    return request_map
+
+
+def _fix_request_data(request_map, prefix):
     fix_requests = {}
     dac_pattern = re.compile(r"(?P<dac>(DAC[0-9]))_CHANNEL(?P<ch>[0-9])")
     for name, number in request_map.items():
         if name.startswith("GENERATOR"):
-            fix_requests["DMA_" + name] = number
+            fix_requests[prefix + "_" + name] = number
         elif name == "FMAC_READ":
             fix_requests["FMAC_RD"] = number
         elif name == "FMAC_WRITE":
@@ -61,6 +69,16 @@ def _fix_request_data(request_map):
             fix_requests["SUBGHZ_RX"] = number
         elif name == "SUBGHZSPI_TX":
             fix_requests["SUBGHZ_TX"] = number
+        elif name == "DAC":
+            fix_requests["DAC_OUT1"] = number
+        elif name == "USART_TX":
+            fix_requests["USART1_TX"] = number
+        elif name == "USART_RX":
+            fix_requests["USART1_RX"] = number
+        elif name == "LPUART_TX":
+            fix_requests["LPUART1_TX"] = number
+        elif name == "LPUART_RX":
+            fix_requests["LPUART1_RX"] = number
         else:
             m = dac_pattern.match(name)
             if m:
@@ -69,48 +87,67 @@ def _fix_request_data(request_map):
     request_map.update(fix_requests)
 
 
+def _get_family(did):
+    family = f"{did.family}xx"
+    if did.string[5:8] in ["h7r", "h7s"]:
+        family = "h7rsxx"
+    elif did.string[5:8] == "wb0":
+        family = "wb0x"
+    elif did.string[5:8] == "wba":
+        family = "wbax"
+    elif did.string[5:8] == "wl3":
+        family = "wl3x"
+    return family
+
+
 def _get_include_path(family):
-    return _CUBE_PATH / Path(f"stm32{family}xx/Inc")
+    return _CUBE_PATH / Path("stm32{}/Inc".format(family))
 
 
-def _get_hal_dma_header_path(family):
-    return _get_include_path(family) / Path(f"stm32{family}xx_hal_dma.h")
+def _get_hal_dma_header_path(did):
+    family = _get_family(did)
+    return _get_include_path(family) / Path("stm32{}_hal_dma.h".format(family))
 
 
-def _get_ll_dmamux_header_path(family):
-    return _get_include_path(family) / Path(f"stm32{family}xx_ll_dmamux.h")
+def _get_ll_dmamux_header_path(did):
+    family = _get_family(did)
+    return _get_include_path(family) / Path("stm32{}_ll_dmamux.h".format(family))
 
 
 # For G4, H7 and L5
-def _read_requests(hal_dma_file):
-    requests_map = _read_map(hal_dma_file, _REQUEST_PATTERN)
+def _read_requests(hal_dma_file, request_pattern):
+    requests_map = _read_map(hal_dma_file, request_pattern)
     out_map = {}
     for r in requests_map.keys():
-        out_map[r.replace("DMA_REQUEST_", "", 1)] = int(requests_map[r])
+        prefix = "BDMA" if "BDMA" in r else "DMA"
+        out_map[r.replace(prefix + "_REQUEST_", "", 1)] = int(requests_map[r])
     return out_map
 
 
 # For G0, WB and WL
 def _read_requests_from_ll_dmamux(hal_dma_file, ll_dmamux_file):
     dmamux_map = _read_map(ll_dmamux_file, _DMAMUX_PATTERN)
-    request_pattern = re.compile(r"^\s*#define\s+(?P<name>(DMA_REQUEST_\w+))\s+(?P<id>(LL_DMAMUX?_REQ_\w+))\s*")
+    request_pattern = re.compile(
+        r"^\s*#define\s+(?P<name>(DMAM?U?X?_REQU?E?S?T?_\w+))\s+(?P<id>(LL_DMAMUX?_REQ_\w+))\s*"
+    )
     requests_map = _read_map(hal_dma_file, request_pattern)
     out_map = {}
     for r in requests_map.keys():
-        out_map[r.replace("DMA_REQUEST_", "", 1)] = int(dmamux_map[requests_map[r]], 16)
+        out_map[r.replace("DMA_REQUEST_", "", 1).replace("DMAMUX_REQ_", "", 1)] = int(dmamux_map[requests_map[r]], 16)
     return out_map
 
 
 # For L4+
-def _read_requests_l4(read_for_p5_q5):
+def _read_requests_l4(did):
+    read_for_p5_q5 = did.name in ["p5", "q5"]
     out_map = {}
     p5_q5_if = "#if defined (STM32L4P5xx) || defined (STM32L4Q5xx)"
     if_pattern = re.compile(r"^\s*#\s*if\s+")
     else_pattern = re.compile(r"^\s*#\s*else")
     endif_pattern = re.compile(r"^\s*#\s*endif")
     in_p5_q5_section = False
     ignore = False
-    with open(_get_hal_dma_header_path("l4"), "r") as header_file:
+    with open(_get_hal_dma_header_path(did), "r") as header_file:
         if_counter = 0
         for line in header_file.readlines():
             if p5_q5_if in line: