Files
serv_benchmark/backend/app/utils/pci_classifier.py
2026-01-11 23:41:30 +01:00

253 lines
8.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
PCI Device Classifier
Classifies PCI devices based on lspci output and device class information.
"""
import re
from typing import Tuple, Optional, Dict, Any
class PCIClassifier:
"""
Classifier for PCI devices based on device class and characteristics.
"""
# PCI device class mappings to type_principal and sous_type
CLASS_MAPPINGS = {
# Storage devices
"SATA controller": ("PCI", "Contrôleur SATA"),
"NVMe": ("PCI", "SSD NVMe"),
"Non-Volatile memory controller": ("PCI", "SSD NVMe"),
"RAID bus controller": ("PCI", "Contrôleur RAID"),
"IDE interface": ("PCI", "Contrôleur IDE"),
"SCSI storage controller": ("PCI", "Contrôleur SCSI"),
# Network devices
"Ethernet controller": ("PCI", "Carte réseau Ethernet"),
"Network controller": ("PCI", "Carte réseau"),
"Wireless controller": ("PCI", "Carte WiFi"),
# Graphics
"VGA compatible controller": ("PCI", "Carte graphique"),
"3D controller": ("PCI", "Carte graphique"),
"Display controller": ("PCI", "Carte graphique"),
# Audio
"Audio device": ("PCI", "Carte son"),
"Multimedia audio controller": ("PCI", "Carte son"),
# USB
"USB controller": ("PCI", "Contrôleur USB"),
# System infrastructure
"Host bridge": ("PCI", "Pont système"),
"PCI bridge": ("PCI", "Pont PCI"),
"ISA bridge": ("PCI", "Pont ISA"),
"SMBus": ("PCI", "Contrôleur SMBus"),
"IOMMU": ("PCI", "Contrôleur IOMMU"),
# Security
"Encryption controller": ("PCI", "Contrôleur de chiffrement"),
# Other
"Serial controller": ("PCI", "Contrôleur série"),
"Communication controller": ("PCI", "Contrôleur de communication"),
"Signal processing controller": ("PCI", "Contrôleur de traitement du signal"),
}
@staticmethod
def classify_device(
device_section: str,
device_info: Optional[Dict[str, Any]] = None
) -> Tuple[str, str]:
"""
Classify a PCI device based on lspci output.
Args:
device_section: Full lspci -v output for a single device
device_info: Optional pre-parsed device information
Returns:
Tuple of (type_principal, sous_type)
"""
if not device_info:
from app.utils.lspci_parser import parse_device_info
device_info = parse_device_info(device_section)
device_class = device_info.get("device_class", "")
description = device_info.get("device_name", "")
vendor_name = device_info.get("vendor_name", "")
# Strategy 1: Direct class mapping
for class_key, (type_principal, sous_type) in PCIClassifier.CLASS_MAPPINGS.items():
if class_key.lower() in device_class.lower():
# Refine network devices
if sous_type == "Carte réseau":
refined = PCIClassifier.refine_network_type(device_section, description)
if refined:
return ("PCI", refined)
return (type_principal, sous_type)
# Strategy 2: Keyword detection in description
keyword_result = PCIClassifier.detect_from_keywords(device_section, description)
if keyword_result:
return ("PCI", keyword_result)
# Strategy 3: Vendor-specific detection
vendor_result = PCIClassifier.detect_from_vendor(vendor_name, description)
if vendor_result:
return ("PCI", vendor_result)
# Default: Generic PCI device
return ("PCI", "Autre")
@staticmethod
def refine_network_type(content: str, description: str) -> Optional[str]:
"""
Refine network device classification (WiFi vs Ethernet).
Args:
content: Full device section
description: Device description
Returns:
Refined sous_type or None
"""
normalized = content.lower() + " " + description.lower()
# WiFi patterns
wifi_patterns = [
r"wi[-]?fi", r"wireless", r"802\.11[a-z]", r"wlan",
r"wireless\s+adapter", r"wireless\s+network",
r"atheros", r"qualcomm.*wireless", r"broadcom.*wireless",
r"intel.*wireless", r"realtek.*wireless"
]
for pattern in wifi_patterns:
if re.search(pattern, normalized, re.IGNORECASE):
return "Carte WiFi"
# Ethernet patterns
ethernet_patterns = [
r"ethernet", r"gigabit", r"10/100", r"1000base",
r"rtl81\d+", r"e1000", r"bnx2", r"tg3"
]
for pattern in ethernet_patterns:
if re.search(pattern, normalized, re.IGNORECASE):
return "Carte réseau Ethernet"
return None
@staticmethod
def detect_from_keywords(content: str, description: str) -> Optional[str]:
"""
Detect device type from keywords in content and description.
Args:
content: Full device section
description: Device description
Returns:
Detected sous_type or None
"""
normalized = content.lower() + " " + description.lower()
keyword_mappings = [
# Storage
(r"nvme|ssd.*pcie|non-volatile.*memory", "SSD NVMe"),
(r"sata|ahci", "Contrôleur SATA"),
# Network
(r"wi[-]?fi|wireless|802\.11", "Carte WiFi"),
(r"ethernet|gigabit|network", "Carte réseau Ethernet"),
# Graphics
(r"nvidia|geforce|quadro|rtx|gtx", "Carte graphique"),
(r"amd.*radeon|rx\s*\d+", "Carte graphique"),
(r"intel.*graphics|intel.*hd", "Carte graphique"),
(r"vga|display|graphics", "Carte graphique"),
# Audio
(r"audio|sound|hda|ac97", "Carte son"),
# USB
(r"xhci|ehci|ohci|uhci|usb.*host", "Contrôleur USB"),
]
for pattern, sous_type in keyword_mappings:
if re.search(pattern, normalized, re.IGNORECASE):
return sous_type
return None
@staticmethod
def detect_from_vendor(vendor_name: str, description: str) -> Optional[str]:
"""
Detect device type from vendor name and description.
Args:
vendor_name: Vendor name
description: Device description
Returns:
Detected sous_type or None
"""
if not vendor_name:
return None
vendor_lower = vendor_name.lower()
# GPU vendors
if any(v in vendor_lower for v in ["nvidia", "amd", "intel", "ati"]):
if any(k in description.lower() for k in ["geforce", "radeon", "quadro", "graphics", "vga"]):
return "Carte graphique"
# Network vendors
if any(v in vendor_lower for v in ["realtek", "intel", "broadcom", "qualcomm", "atheros"]):
if any(k in description.lower() for k in ["ethernet", "network", "wireless", "wifi", "802.11"]):
if any(k in description.lower() for k in ["wireless", "wifi", "802.11"]):
return "Carte WiFi"
return "Carte réseau Ethernet"
# Storage vendors
if any(v in vendor_lower for v in ["samsung", "crucial", "micron", "western digital", "seagate"]):
if "nvme" in description.lower():
return "SSD NVMe"
return None
@staticmethod
def extract_technical_specs(device_info: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract technical specifications for caracteristiques_specifiques field.
Args:
device_info: Parsed device information
Returns:
Dictionary with technical specifications
"""
specs = {
"slot": device_info.get("slot"),
"device_class": device_info.get("device_class"),
"vendor_name": device_info.get("vendor_name"),
"subsystem": device_info.get("subsystem"),
"driver": device_info.get("driver"),
"iommu_group": device_info.get("iommu_group"),
}
# Add vendor:device ID if available
if device_info.get("vendor_device_id"):
specs["pci_device_id"] = device_info.get("vendor_device_id")
# Add revision if available
if device_info.get("revision"):
specs["revision"] = device_info.get("revision")
# Add modules if available
if device_info.get("modules"):
specs["modules"] = ", ".join(device_info.get("modules", []))
# Clean None values
return {k: v for k, v in specs.items() if v is not None}