update

src/qoa4ml/utils/jetson/engine.py

@@ -0,0 +1,119 @@
+# Logging
+import logging
+import os
+from typing import ClassVar
+
+# from .exceptions import JtopException
+# Create logger
+logger = logging.getLogger(__name__)
+
+
+def read_engine(path):
+    # Read status online
+    engine = {}
+    # Check if access to this file
+    if os.access(path + "/clk_enable_count", os.R_OK):
+        with open(path + "/clk_enable_count") as f:
+            # Write online engine
+            engine["online"] = int(f.read()) == 1
+    # Check if access to this file
+    if os.access(path + "/clk_rate", os.R_OK):
+        with open(path + "/clk_rate") as f:
+            # Write current engine
+            engine["cur"] = int(f.read()) // 1000
+    # Decode clock rate
+    max_value = False
+    if os.access(path + "/clk_max_rate", os.R_OK):
+        with open(path + "/clk_max_rate") as f:
+            # Write status engine
+            value = int(f.read())
+            # 18446744073709551615 = FFFF FFFF FFFF FFFF = 2 ^ 16
+            if value != 18446744073709551615:
+                engine["max"] = value // 1000
+                max_value = True
+    if os.access(path + "/clk_min_rate", os.R_OK) and max_value:
+        with open(path + "/clk_min_rate") as f:
+            # Write status engine
+            engine["min"] = int(f.read()) // 1000
+    return engine
+
+
+class EngineService:
+    ENGINES: ClassVar[list[str]] = [
+        "ape",
+        "dla",
+        "pva",
+        "vic",
+        "nvjpg",
+        "nvenc",
+        "nvdec",
+        "se.",
+        "cvnas",
+        "msenc",
+        "ofa",
+    ]
+
+    def __init__(self):
+        # Sort list before start
+        EngineService.ENGINES.sort()
+        self.engines_path = {}
+        # List all engines available
+        engine_path = "/sys/kernel/debug/clk"
+        if os.getenv("JTOP_TESTING", False):
+            engine_path = "/fake_sys/kernel/debug/clk"
+            logger.warning(f"Running in JTOP_TESTING folder={engine_path}")
+        list_all_engines = [x[0] for x in os.walk(engine_path)]
+        # Search all available engines
+        for name in EngineService.ENGINES:
+            if name.endswith("."):
+                name = name[:-1]
+                local_path = f"{engine_path}/{name}"
+                if os.path.isdir(local_path):
+                    self.engines_path[name.upper()] = [local_path]
+            else:
+                # https://stackoverflow.com/questions/4843158/how-to-check-if-a-string-is-a-substring-of-items-in-a-list-of-strings
+                local_path = f"{engine_path}/{name}"
+                # In this search are removed all engines that have a '.' on their name
+                # like ape.buffer or nvdec.buf
+                matching = [
+                    s for s in list_all_engines if local_path in s and "." not in s
+                ]
+                # Add in list all engines
+                if matching:
+                    # Check if name end with a number, if true collect by number
+                    # dla0 dla1 ...
+                    if os.path.basename(matching[0]).split("_")[0] == f"{name}0":
+                        logger.info(f"Special Engine group found: [{name}X]")
+                        for num in range(10):
+                            name_engine = f"{name}{num}"
+                            new_match = [
+                                match for match in matching if name_engine in match
+                            ]
+                            if new_match:
+                                self.engines_path[name_engine.upper()] = sorted(
+                                    new_match
+                                )
+                            else:
+                                break
+                    else:
+                        self.engines_path[name.upper()] = sorted(matching)
+        # Print all engines found
+        if self.engines_path:
+            engines_string = " ".join(name for name in self.engines_path)
+            logger.info(f"Engines found: [{engines_string}]")
+        else:
+            logger.warn("Not engines found!")
+
+    def get_status(self):
+        status = {}
+        # Read status from all engines
+        for engine in self.engines_path:
+            status[engine] = {}
+            for local_path in self.engines_path[engine]:
+                name_engine = os.path.basename(local_path).upper()
+                logger.debug(f"Status [{name_engine}] in {local_path}")
+                status[engine][name_engine] = read_engine(local_path)
+        return status
+
+
+# EOF

src/qoa4ml/utils/jetson/memory.py

@@ -291,8 +291,7 @@ def swap_deactivate(self, path=""):
 
 
 class MemoryService:
-    def __init__(self, config):
-        self._config = config
+    def __init__(self):
         # Extract memory page size
         self._page_size = os.sysconf("SC_PAGE_SIZE")
         # board type
@@ -309,11 +308,11 @@ def __init__(self, config):
         # Initialization memory
         logger.info("Memory service started")
 
-    def swap_path(self):
-        config = self._config.get("swap", {})
-        directory = config.get("directory", CONFIG_DEFAULT_SWAP_DIRECTORY)
-        swap_name = config.get("name", CONFIG_DEFAULT_SWAP_NAME)
-        return f"{directory}/{swap_name}"
+    # def swap_path(self):
+    #     config = self._config.get("swap", {})
+    #     directory = config.get("directory", CONFIG_DEFAULT_SWAP_DIRECTORY)
+    #     swap_name = config.get("name", CONFIG_DEFAULT_SWAP_NAME)
+    #     return f"{directory}/{swap_name}"
 
     def clear_cache(self):
         """

tests/gpu_test.py

@@ -1,6 +1,12 @@
 from devtools import debug
 from qoa4ml.utils.jetson.jetson_gpu import GPUService
+from qoa4ml.utils.jetson.memory import MemoryService
+from qoa4ml.utils.jetson.processes import ProcessService
 
 gpu_service = GPUService()
+proceses_service = ProcessService()
+total, table = proceses_service.get_status()
+memory_service = MemoryService()
 
-debug(gpu_service.get_status())
+# debug(gpu_service.get_status())
+debug(memory_service.get_status(total))