267 lines
10 KiB
Python
267 lines
10 KiB
Python
"""
|
|
Plugin pour LibreHardwareMonitor
|
|
Utilise l'API REST intégrée de LHM (port 8085 par défaut)
|
|
"""
|
|
import requests
|
|
from typing import Dict, List, Optional, Any
|
|
from .base import BasePlugin
|
|
|
|
|
|
class LibreHardwareMonitorPlugin(BasePlugin):
|
|
"""Plugin pour LibreHardwareMonitor"""
|
|
|
|
def __init__(self, config: dict):
|
|
super().__init__(config)
|
|
self.history = {}
|
|
self.max_history = 120 # 2 minutes à 1 sample/sec
|
|
|
|
def get_id(self) -> str:
|
|
return 'librehardwaremonitor'
|
|
|
|
def get_name(self) -> str:
|
|
return 'LibreHardwareMonitor'
|
|
|
|
def get_default_port(self) -> int:
|
|
return 8085
|
|
|
|
def test_connection(self) -> Dict[str, Any]:
|
|
"""Teste la connexion à LibreHardwareMonitor"""
|
|
try:
|
|
url = f"{self.get_base_url()}/data.json"
|
|
response = requests.get(url, timeout=5)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
|
|
# Extraire la version si disponible
|
|
version = "Inconnue"
|
|
if data and "Text" in data:
|
|
version = data.get("Text", "LibreHardwareMonitor")
|
|
|
|
# Compter les capteurs
|
|
sensors = self.parse_sensors(data)
|
|
|
|
return {
|
|
'success': True,
|
|
'message': f'Connecté - {len(sensors)} capteurs détectés',
|
|
'version': version,
|
|
'sensor_count': len(sensors)
|
|
}
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
return {
|
|
'success': False,
|
|
'message': f'Impossible de se connecter à {self.get_base_url()}. Vérifiez que LibreHardwareMonitor est lancé avec le serveur web activé.'
|
|
}
|
|
except requests.exceptions.Timeout:
|
|
return {
|
|
'success': False,
|
|
'message': 'Timeout - Le serveur ne répond pas'
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'message': f'Erreur: {str(e)}'
|
|
}
|
|
|
|
def get_data(self) -> Optional[Dict]:
|
|
"""Récupère les données JSON depuis LHM Remote Server"""
|
|
try:
|
|
url = f"{self.get_base_url()}/data.json"
|
|
response = requests.get(url, timeout=5)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Erreur LHM get_data: {e}")
|
|
return None
|
|
|
|
def get_hierarchy(self) -> List[dict]:
|
|
"""Récupère la hiérarchie des capteurs pour l'admin"""
|
|
data = self.get_data()
|
|
if not data:
|
|
return []
|
|
|
|
return self._build_hierarchy(data)
|
|
|
|
def _build_hierarchy(self, data: Dict) -> List[dict]:
|
|
"""Construit la hiérarchie des capteurs"""
|
|
if not data or "Children" not in data:
|
|
return []
|
|
|
|
def build_tree(node, level=0):
|
|
node_text = node.get("Text", "")
|
|
node_value = node.get("Value", "")
|
|
sensor_id = node.get("SensorId", "")
|
|
|
|
result = {
|
|
"name": node_text,
|
|
"value": node_value,
|
|
"id": sensor_id,
|
|
"level": level,
|
|
"type": self._guess_sensor_type(node_text, node_value) if node_value else "group",
|
|
"children": []
|
|
}
|
|
|
|
if "Children" in node and node["Children"]:
|
|
for child in node["Children"]:
|
|
result["children"].append(build_tree(child, level + 1))
|
|
|
|
return result
|
|
|
|
tree = []
|
|
for child in data.get("Children", []):
|
|
tree.append(build_tree(child))
|
|
|
|
return tree
|
|
|
|
def parse_sensors(self, data: Dict) -> List[Dict]:
|
|
"""Parse les données et retourne une liste plate de capteurs"""
|
|
sensors = []
|
|
|
|
if not data or "Children" not in data:
|
|
return sensors
|
|
|
|
def extract_sensors(node, parent_name="", hardware_name="", hardware_type=""):
|
|
node_text = node.get("Text", "")
|
|
current_path = f"{parent_name}/{node_text}" if parent_name else node_text
|
|
|
|
# Détecter le type de hardware au niveau 1
|
|
if not hardware_name and "Children" in node:
|
|
hw_type = self._guess_hardware_type(node_text)
|
|
hardware_name = node_text
|
|
hardware_type = hw_type
|
|
|
|
# Si le nœud a une valeur, c'est un capteur
|
|
if "Value" in node and node["Value"]:
|
|
sensor_id = node.get("SensorId", "")
|
|
if not sensor_id:
|
|
sensor_id = f"{current_path}".replace(" ", "-").lower()
|
|
|
|
value_str = node.get("Value", "")
|
|
raw_value = self._extract_numeric_value(value_str)
|
|
unit = self._extract_unit(value_str)
|
|
sensor_type = self._guess_sensor_type(node_text, value_str)
|
|
|
|
sensor = {
|
|
"id": sensor_id,
|
|
"name": node_text,
|
|
"value": value_str,
|
|
"raw_value": raw_value,
|
|
"type": sensor_type,
|
|
"unit": unit,
|
|
"hardware": hardware_name,
|
|
"hardware_type": hardware_type,
|
|
"category": parent_name.split("/")[0] if "/" in parent_name else parent_name,
|
|
"path": current_path,
|
|
"min": node.get("Min", ""),
|
|
"max": node.get("Max", ""),
|
|
}
|
|
sensors.append(sensor)
|
|
|
|
# Parcourir les enfants récursivement
|
|
if "Children" in node:
|
|
for child in node["Children"]:
|
|
extract_sensors(child, current_path, hardware_name, hardware_type)
|
|
|
|
for child in data.get("Children", []):
|
|
extract_sensors(child)
|
|
|
|
return sensors
|
|
|
|
def _guess_hardware_type(self, name: str) -> str:
|
|
"""Devine le type de hardware basé sur le nom"""
|
|
name_lower = name.lower()
|
|
|
|
if 'cpu' in name_lower or 'processor' in name_lower:
|
|
return 'CPU'
|
|
elif 'gpu' in name_lower or 'graphics' in name_lower or 'nvidia' in name_lower or 'amd' in name_lower or 'radeon' in name_lower or 'geforce' in name_lower:
|
|
return 'GPU'
|
|
elif 'memory' in name_lower or 'ram' in name_lower:
|
|
return 'RAM'
|
|
elif 'motherboard' in name_lower or 'mainboard' in name_lower:
|
|
return 'Motherboard'
|
|
elif 'storage' in name_lower or 'disk' in name_lower or 'ssd' in name_lower or 'hdd' in name_lower or 'nvme' in name_lower:
|
|
return 'Storage'
|
|
elif 'network' in name_lower or 'ethernet' in name_lower or 'wifi' in name_lower:
|
|
return 'Network'
|
|
elif 'battery' in name_lower:
|
|
return 'Battery'
|
|
else:
|
|
return 'Other'
|
|
|
|
def _guess_sensor_type(self, name: str, value: str) -> str:
|
|
"""Devine le type de capteur basé sur le nom et la valeur"""
|
|
name_lower = name.lower()
|
|
value_lower = value.lower()
|
|
|
|
if "°c" in value_lower or "temperature" in name_lower or "temp" in name_lower:
|
|
return "temperature"
|
|
elif "%" in value or "load" in name_lower or "usage" in name_lower:
|
|
return "percentage"
|
|
elif "rpm" in value_lower or "fan" in name_lower:
|
|
return "fan"
|
|
elif "mhz" in value_lower or "ghz" in value_lower or "clock" in name_lower:
|
|
return "frequency"
|
|
elif "w" in value_lower and "wh" not in value_lower:
|
|
return "power"
|
|
elif "v" in value_lower:
|
|
return "voltage"
|
|
elif "gb" in value_lower or "mb" in value_lower or "memory" in name_lower:
|
|
return "memory"
|
|
else:
|
|
return "generic"
|
|
|
|
def _extract_numeric_value(self, value_str: str) -> float:
|
|
"""Extrait la valeur numérique d'une chaîne"""
|
|
try:
|
|
# Enlever tous les caractères non numériques sauf le point et la virgule
|
|
clean = value_str.replace("°C", "").replace("%", "").replace("V", "").replace("W", "")
|
|
clean = clean.replace("MHz", "").replace("GHz", "").replace("GB", "").replace("MB", "")
|
|
clean = clean.replace("RPM", "").replace(" ", "").replace(",", ".").strip()
|
|
|
|
if clean:
|
|
return float(clean)
|
|
except (ValueError, AttributeError):
|
|
pass
|
|
return 0.0
|
|
|
|
def _extract_unit(self, value_str: str) -> str:
|
|
"""Extrait l'unité d'une valeur"""
|
|
value_str = value_str.strip()
|
|
|
|
units = ['°C', '°F', '%', 'MHz', 'GHz', 'GB', 'MB', 'KB', 'RPM', 'W', 'V', 'A', 'GB/s', 'MB/s']
|
|
for unit in units:
|
|
if unit in value_str:
|
|
return unit
|
|
return ''
|
|
|
|
# === Gestion de l'historique ===
|
|
|
|
def update_history(self, sensors: List[Dict]):
|
|
"""Met à jour l'historique des capteurs pour les graphiques"""
|
|
for sensor in sensors:
|
|
sensor_id = sensor.get("id", "")
|
|
if not sensor_id:
|
|
continue
|
|
|
|
raw_value = sensor.get("raw_value")
|
|
if raw_value is None:
|
|
raw_value = self._extract_numeric_value(sensor.get("value", ""))
|
|
|
|
if raw_value == 0.0 and "0" not in sensor.get("value", ""):
|
|
continue
|
|
|
|
if sensor_id not in self.history:
|
|
self.history[sensor_id] = []
|
|
|
|
self.history[sensor_id].append(raw_value)
|
|
|
|
if len(self.history[sensor_id]) > self.max_history:
|
|
self.history[sensor_id] = self.history[sensor_id][-self.max_history:]
|
|
|
|
def get_history(self, sensor_id: str, count: int = 60) -> List[float]:
|
|
"""Récupère l'historique d'un capteur"""
|
|
if sensor_id not in self.history:
|
|
return []
|
|
return self.history[sensor_id][-count:]
|