NAS-132991 / 25.04 / make hardware.cpu plugin private (#15168)

* make private hardware.cpu plugin

* format and add enum

* address review

src/middlewared/middlewared/plugins/hardware/cpu.py

@@ -1,55 +1,67 @@
-from pathlib import Path
+import os
 
 from middlewared.service import Service
 from middlewared.service_exception import ValidationError
-from middlewared.schema import accepts, returns, Dict, Str
 from middlewared.utils.functools_ import cache
 
 
-class HardwareCpuService(Service):
+class SysfsCpuObj:
+    BASE = "/sys/devices/system/cpu/"
+    GOVERNOR_BASE = os.path.join(BASE, "cpu0/cpufreq")
+    CURRENT_GOVERNOR = os.path.join(GOVERNOR_BASE, "scaling_governor")
+    GOVERNORS_AVAIL = os.path.join(GOVERNOR_BASE, "scaling_available_governors")
+
 
+class HardwareCpuService(Service):
     class Config:
-        namespace = 'hardware.cpu'
-        cli_namespace = 'system.hardware.cpu'
+        private = True
+        namespace = "hardware.cpu"
+        cli_namespace = "system.hardware.cpu"
 
-    @accepts()
-    @returns(Dict('governor', additional_attrs=True))
     @cache
-    def available_governors(self):
+    def available_governors(self) -> dict[str, str] | dict:
         """Return available cpu governors"""
         try:
-            with open('/sys/devices/system/cpu/cpu0/cpufreq/scaling_available_governors') as f:
+            with open(SysfsCpuObj.GOVERNORS_AVAIL) as f:
                 return {i: i for i in f.read().split()}
         except FileNotFoundError:
             # doesn't support changing governor
             return dict()
 
-    @accepts()
-    @returns(Str('governor'))
-    def current_governor(self):
+    def current_governor(self) -> str | None:
         """Returns currently set cpu governor"""
         try:
-            with open('/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor') as f:
+            with open(SysfsCpuObj.CURRENT_GOVERNOR) as f:
                 return f.read().strip()
         except FileNotFoundError:
             # doesn't support changing governor
             return
 
-    @accepts(Str('governor', required=True))
-    @returns()
-    def set_governor(self, governor):
-        """Set the cpu governor to `governor` on all cpus"""
+    def set_governor(self, governor: str) -> None:
+        """Set the cpu governor to `governor` on all cpus.
+        Available governors may be determined by calling hardware.cpu.available_governors.
+        """
         curr_gov = self.current_governor()
         if curr_gov is None:
-            raise ValidationError('hardware.cpu.governor', 'Changing cpu governor is not supported')
+            raise ValidationError(
+                "hardware.cpu.governor", "Changing cpu governor is not supported"
+            )
         elif curr_gov == governor:
             # current governor is already set to what is being requested
             return
         elif governor not in self.available_governors():
-            raise ValidationError('hardware.cpu.governor', f'{governor} is not available')
+            raise ValidationError(
+                "hardware.cpu.governor", f"{governor} is not available"
+            )
 
-        for i in Path('/sys/devices/system/cpu').iterdir():
-            if i.is_dir() and i.name.startswith('cpu'):
-                cpug = (i / 'cpufreq/scaling_governor')
-                if cpug.exists():
-                    cpug.write_text(governor)
+        try:
+            with os.scandir(SysfsCpuObj.BASE) as sdir:
+                for i in filter(
+                    lambda x: x.is_dir() and x.name.startswith("cpu"), sdir
+                ):
+                    with open(
+                        os.path.join(i.path, "cpufreq/scaling_governonr"), "w"
+                    ) as f:
+                        f.write(governor)
+        except FileNotFoundError:
+            pass