feat/add-basic-telemetry #10

Merged
pufereq merged 8 commits from feat/add-basic-telemetry into develop 2026-03-08 19:42:40 +00:00
4 changed files with 199 additions and 4 deletions
Showing only changes of commit 07b9ba6fdf - Show all commits

View File

@@ -6,7 +6,6 @@ from __future__ import annotations
import logging as lg import logging as lg
import os import os
import platform import platform
import re
import subprocess import subprocess
import uuid import uuid
from typing import Any from typing import Any
@@ -53,7 +52,6 @@ class Client:
on_message=self.handle_message, on_message=self.handle_message,
) )
self.logger.debug("Gathering initial telemetry data...")
self.initial_telemetry: dict[str, Any] = ( self.initial_telemetry: dict[str, Any] = (
self._gather_initial_telemetry() self._gather_initial_telemetry()
) )
@@ -73,22 +71,50 @@ class Client:
return mac_address return mac_address
def _get_cpu_model(self) -> str: def _get_cpu_model(self) -> str:
"""Get the CPU model name.
if platform.system() == "Windows": Returns:
return platform.processor() str: The CPU model name, or "Unknown" if it cannot be determined.
elif platform.system() == "Darwin": """
os.environ["PATH"] = os.environ["PATH"] + os.pathsep + "/usr/sbin"
command = "sysctl -n machdep.cpu.brand_string" system = platform.system()
return str(subprocess.check_output(command).strip())
elif platform.system() == "Linux": if system == "Windows":
command = "cat /proc/cpuinfo" try:
all_info = ( import winreg
subprocess.check_output(command, shell=True).decode().strip()
) key = winreg.OpenKey(
for line in all_info.split("\n"): winreg.HKEY_LOCAL_MACHINE,
if "model name" in line: r"HARDWARE\DESCRIPTION\System\CentralProcessor\0",
return re.sub(".*model name.*:", "", line, 1).strip() )
return "Unknown" model, _ = winreg.QueryValueEx(key, "ProcessorNameString")
winreg.CloseKey(key)
return model.strip()
except Exception:
return platform.processor()
elif system == "Linux":
try:
with open("/proc/cpuinfo", "r") as f:
for line in f:
if "model name" in line:
return line.split(":")[1].strip()
except Exception:
return platform.processor()
elif system == "Darwin":
try:
return (
subprocess.check_output(
["sysctl", "-n", "machdep.cpu.brand_string"]
)
.decode()
.strip()
)
except Exception:
return platform.processor()
return platform.processor()
def _gather_initial_telemetry(self) -> dict[str, Any]: def _gather_initial_telemetry(self) -> dict[str, Any]:
"""Gather initial telemetry data.""" """Gather initial telemetry data."""
@@ -110,10 +136,10 @@ class Client:
"model": self._get_cpu_model(), "model": self._get_cpu_model(),
"physical_cores": psutil.cpu_count(logical=False), "physical_cores": psutil.cpu_count(logical=False),
"threads": psutil.cpu_count(logical=True), "threads": psutil.cpu_count(logical=True),
"max_frequency_mhz": cpu_frequency.max "max_frequency_mhz": round(cpu_frequency.max, 2)
if cpu_frequency if cpu_frequency
else "Unknown", else "Unknown",
"min_frequency_mhz": cpu_frequency.min "min_frequency_mhz": round(cpu_frequency.min, 2)
if cpu_frequency if cpu_frequency
else "Unknown", else "Unknown",
} }