Label top CPU frequencies by core type

tools/pc-metrics/push-metrics-ble.py

@@ -125,6 +125,7 @@ class CpuFrequencyReader:
         self.total_counter = ctypes.c_void_p()
         self.core_counters: list[tuple[int, ctypes.c_void_p]] = []
         self.base_mhz = self._base_mhz()
+        self.core_type_by_index: dict[int, str] = {}
         self.available = False
 
         if sys.platform != "win32":
@@ -168,6 +169,7 @@ class CpuFrequencyReader:
                 self.total_counter = ctypes.c_void_p()
 
             logical_count = psutil.cpu_count(logical=True) or 0
+            self.core_type_by_index = self._read_core_type_map(logical_count)
             for index in range(logical_count):
                 counter = ctypes.c_void_p()
                 path = rf"\Processor Information(0,{index})\% Processor Performance"
@@ -192,9 +194,90 @@ class CpuFrequencyReader:
         return 0
 
     @staticmethod
-    def _format_core_mhz(index: int, mhz: int) -> str:
+    def _fallback_core_type_map(logical_count: int) -> dict[int, str]:
+        if logical_count == 16:
+            return {
+                index: ("P" if index < 4 else "E" if index < 12 else "L")
+                for index in range(logical_count)
+            }
+        return {}
+
+    @staticmethod
+    def _read_core_type_map(logical_count: int) -> dict[int, str]:
+        fallback = CpuFrequencyReader._fallback_core_type_map(logical_count)
+        if sys.platform != "win32":
+            return fallback
+
+        try:
+            kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+            get_cpu_sets = kernel32.GetSystemCpuSetInformation
+            get_cpu_sets.restype = wintypes.BOOL
+            get_cpu_sets.argtypes = [
+                ctypes.c_void_p,
+                wintypes.ULONG,
+                ctypes.POINTER(wintypes.ULONG),
+                ctypes.c_void_p,
+                wintypes.ULONG,
+            ]
+
+            returned = wintypes.ULONG()
+            get_cpu_sets(None, 0, ctypes.byref(returned), None, 0)
+            if returned.value <= 0:
+                return fallback
+
+            buffer = ctypes.create_string_buffer(returned.value)
+            if not get_cpu_sets(buffer, returned, ctypes.byref(returned), None, 0):
+                return fallback
+
+            base = ctypes.addressof(buffer)
+            offset = 0
+            infos: list[dict[str, int]] = []
+            while offset + 20 <= returned.value:
+                ptr = base + offset
+                size = ctypes.c_uint32.from_address(ptr).value
+                info_type = ctypes.c_uint32.from_address(ptr + 4).value
+                if size <= 0:
+                    break
+                if info_type == 0:
+                    infos.append(
+                        {
+                            "logical": ctypes.c_ubyte.from_address(ptr + 14).value,
+                            "llc": ctypes.c_ubyte.from_address(ptr + 16).value,
+                            "efficiency": ctypes.c_ubyte.from_address(ptr + 18).value,
+                        }
+                    )
+                offset += size
+
+            if not infos:
+                return fallback
+
+            min_efficiency = min(item["efficiency"] for item in infos)
+            max_efficiency = max(item["efficiency"] for item in infos)
+            if max_efficiency == min_efficiency:
+                return fallback
+
+            p_llc = {
+                item["llc"]
+                for item in infos
+                if item["efficiency"] == max_efficiency
+            }
+            type_map: dict[int, str] = {}
+            for item in infos:
+                logical = item["logical"]
+                if item["efficiency"] == max_efficiency:
+                    type_map[logical] = "P"
+                elif item["llc"] not in p_llc:
+                    type_map[logical] = "L"
+                else:
+                    type_map[logical] = "E"
+            return type_map or fallback
+        except (AttributeError, OSError, ValueError):
+            return fallback
+
+    def _format_core_mhz(self, index: int, mhz: int) -> str:
         ghz10 = max(0, int(round(mhz / 100)))
-        return f"{index:02d}:{ghz10 // 10}.{ghz10 % 10}"
+        core_type = self.core_type_by_index.get(index, "C")
+        return f"{core_type}:{ghz10 // 10}.{ghz10 % 10}"
 
     def _read_counter(self, counter: ctypes.c_void_p) -> float:
         if self.pdh is None or not counter: