Add disk_io module

lnxlink/__main__.py

@@ -295,7 +295,6 @@ def temp_connection_callback(self, status):
                         topic,
                         payload=message,
                     )
-                    print(topic, message)
             else:
                 logger.info("Power Up detected.")
                 if self.kill:
@@ -455,9 +454,11 @@ def setup_discovery_entities(self, addon, service, exp_name, options):
             retain=True,
         )
 
-    def setup_discovery(self):
+    def setup_discovery(self, filter_name=None):
         """First time setup of discovery for Home Assistant"""
         for service, addon in self.addons.items():
+            if filter_name is not None and filter_name != service:
+                continue
             if hasattr(addon, "exposed_controls"):
                 for exp_name, options in addon.exposed_controls().items():
                     try:

lnxlink/modules/disk_io.py

@@ -0,0 +1,92 @@
+"""Get Disk IO usage"""
+import glob
+import asyncio
+from timeit import default_timer as timer
+
+
+class Addon():
+    """Addon module"""
+
+    def __init__(self, lnxlink):
+        """Setup addon"""
+        self.name = "DiskIO"
+        self.disks = self._get_disks()
+
+        self.stat_items = [
+            "read IOs",
+            "read merges",
+            "read sectors",
+            "read ticks",
+            "write IOs",
+            "write merges",
+            "write sectors",
+            "write ticks",
+            "in_flight",
+            "io_ticks",
+            "time_in_queue",
+            "discard IOs",
+            "discard merges",
+            "discard sectors",
+            "discard ticks",
+            "flush IOs",
+            "flush ticks",
+        ]
+
+    def exposed_controls(self):
+        """Exposes to home assistant"""
+        discovery_info = {}
+        for disk in self.disks:
+            discovery_info[f"Disk IO {disk}"] = {
+                "type": "sensor",
+                "icon": "mdi:barcode-scan",
+                "unit": "%",
+                "entity_category": "diagnostic",
+                "state_class": "measurement",
+                "value_template": f"{{{{ value_json.get('{disk}') }}}}",
+            }
+        return discovery_info
+
+    def get_info(self):
+        """Gather information from the system"""
+        disks = self._get_disks()
+        if self.disks != disks:
+            self.disks = disks
+            self.lnxlink.setup_discovery("disk_io")
+        results = asyncio.run(self._get_info_async())
+        return results
+
+    async def _get_info_async(self):
+        features = []
+        for disk in self.disks:
+            features.append(self._run_check(disk))
+        gathers = await asyncio.gather(*features)
+        results = {}
+        for disk, utilization in gathers:
+            results[disk] = utilization
+        return results
+
+    def _get_disks(self):
+        disks = []
+        for disk in glob.glob("/sys/block/*", recursive=False):
+            disk_name = disk.removeprefix("/sys/block/")
+            if "loop" in disk_name:
+                continue
+            disks.append(disk_name)
+        return disks
+
+    async def _run_check(self, disk):
+        with open(f"/sys/block/{disk}/stat") as file:
+            pout1 = file.read()
+        start = timer()
+        await asyncio.sleep(0.1)
+        with open(f"/sys/block/{disk}/stat") as file:
+            pout2 = file.read()
+        totaltime = timer() - start
+
+        stats1 = dict(zip(self.stat_items, map(int, pout1.split())))
+        stats2 = dict(zip(self.stat_items, map(int, pout2.split())))
+
+        utilization = (stats2["io_ticks"] - stats1["io_ticks"]) / totaltime / 10
+        utilization = min(utilization, 100)
+        utilization = int(round(utilization, 0))
+        return disk, utilization