""" Main signature matching engine Updated to use unified device identifier from iteration 5/5. Provides backward compatibility with legacy MatchResult format. """ from dataclasses import dataclass from typing import List, Optional, Dict, Any from loguru import logger from ..parser.metadata import SignalMetadata from .device_identifier import get_device_identifier, IdentificationResult @dataclass class MatchResult: """Result of a signature match""" device_id: int device_name: str manufacturer: str confidence: float # 0.0 to 1.0 match_method: str # 'exact', 'partial', 'pattern', 'timing' match_details: Dict[str, Any] def to_dict(self) -> Dict: """Convert to dictionary""" return { 'device_id': self.device_id, 'device_name': self.device_name, 'manufacturer': self.manufacturer, 'confidence': self.confidence, 'match_method': self.match_method, 'match_details': self.match_details } class SignatureMatcher: """ Main signature matching engine Coordinates multiple matching strategies to identify devices from RF signal metadata """ def __init__(self, database=None): """ Initialize matcher with database connection Args: database: Database connection for signature queries (legacy, optional) """ self.db = database self.strategies = [] # New unified identifier (iteration 5/5) self.identifier = get_device_identifier() def add_strategy(self, strategy): """Add a matching strategy (legacy)""" self.strategies.append(strategy) def match(self, metadata: SignalMetadata, max_results: int = 10) -> List[MatchResult]: """ Match signal metadata against signature database Updated to use unified device identifier (iteration 5/5). Maintains backward compatibility with legacy MatchResult format. Args: metadata: Parsed signal metadata max_results: Maximum number of results to return Returns: List of MatchResult objects sorted by confidence """ # Use new unified identifier try: result = self.identifier.identify(metadata, top_k=max_results) # Convert to legacy MatchResult format match_results = [] for i, device_match in enumerate(result.matches): match_results.append(MatchResult( device_id=i + 1, # Sequential ID device_name=device_match.name, manufacturer=device_match.manufacturer or "Unknown", confidence=device_match.confidence, match_method=device_match.match_method, # Preserve each device's own catalog category so the API can # surface it per-device (e.g. an Acurite sensor stays # "Weather Sensor"); the ML overall call remains in the # match_details' predicted_category. match_details={ **device_match.details, "device_category": device_match.category, } )) return match_results except Exception as e: logger.error(f"Unified identifier failed: {e}") # Fallback to legacy strategy-based matching if self.strategies: all_matches = [] for strategy in self.strategies: try: matches = strategy.match(metadata, self.db) all_matches.extend(matches) except Exception as e2: logger.error(f"Strategy {strategy.__class__.__name__} failed: {e2}") # Deduplicate and sort unique_matches = self._deduplicate_matches(all_matches) sorted_matches = sorted(unique_matches, key=lambda x: x.confidence, reverse=True) return sorted_matches[:max_results] else: return [] def _deduplicate_matches(self, matches: List[MatchResult]) -> List[MatchResult]: """ Deduplicate matches, keeping highest confidence for each device Args: matches: List of match results Returns: Deduplicated list """ device_map = {} for match in matches: if match.device_id not in device_map: device_map[match.device_id] = match else: # Keep higher confidence match if match.confidence > device_map[match.device_id].confidence: device_map[match.device_id] = match return list(device_map.values()) class MatchStrategy: """Base class for matching strategies""" def match(self, metadata: SignalMetadata, db) -> List[MatchResult]: """ Match metadata against database Args: metadata: Signal metadata db: Database connection Returns: List of MatchResult objects """ raise NotImplementedError