""" Visit Logger Module Logs detection results and scores to the database. Records include: plate detection, character recognition, and scoring information. """ from datetime import datetime import hashlib from .connection import Database class VisitLogger: """ Logs visit records to the database. Records are written to multiple tables: - visits: Main visit record - plate_scores: License plate prestige score - segment_scores: Vehicle segment/class score - purchasing_power: Final purchasing power metric """ def __init__(self, database: Database): self.db = database def log(self, plate_digits: str, plate_letters: str, plate_score: float, segment: str, segment_score: float, purchasing_power: float, car_brand: str = None, car_model: str = None, timestamp: datetime = None, camera_id: str = None) -> str: if not plate_digits: raise ValueError("plate_digits cannot be empty") if not camera_id: raise ValueError( "camera_id is required — pass a valid camera UUID from the cameras table." ) if not (0.0 <= plate_score <= 1.0): raise ValueError(f"plate_score must be in [0.0, 1.0], got {plate_score}") if not (0.0 <= segment_score <= 1.0): raise ValueError(f"segment_score must be in [0.0, 1.0], got {segment_score}") if not (0.0 <= purchasing_power <= 1.0): raise ValueError(f"purchasing_power must be in [0.0, 1.0], got {purchasing_power}") timestamp = timestamp or datetime.now() try: visit_id = self._insert_visit(timestamp, plate_digits, plate_letters, camera_id) self._insert_plate_score(visit_id, plate_score) self._insert_segment_score(visit_id, segment, segment_score) self._insert_purchasing_power_score(visit_id, purchasing_power) return visit_id except Exception as e: raise RuntimeError(f"Failed to log visit: {e}") def _insert_visit(self, timestamp: datetime, plate_digits: str, plate_letters: str, camera_id: str) -> str: plate_hash = hashlib.sha256( f"{plate_letters}{plate_digits}".encode("utf-8") ).hexdigest() data = { "camera_id": camera_id, "plate_hash": plate_hash, "plate_digits": plate_digits, "plate_letters": plate_letters, "visited_at": timestamp.isoformat() } response = self.db.insert("visits", data) if not response or len(response) == 0: raise RuntimeError("Failed to insert visit: no data returned") return response[0]["id"] def _insert_plate_score(self, visit_id: str, score: float) -> None: data = { "visit_id": visit_id, "digit_length_score": score, "repeat_score": 0, "sequential_score": 0, "low_number_score": 0, "total_score": score } response = self.db.insert("plate_scores", data) if not response or len(response) == 0: raise RuntimeError("Failed to insert plate score") def _insert_segment_score(self, visit_id: str, segment: str, score: float) -> None: data = { "visit_id": visit_id, "segment": segment, "confidence": score, "segment_score": score } response = self.db.insert("segment_scores", data) if not response or len(response) == 0: raise RuntimeError("Failed to insert segment score") def _insert_purchasing_power_score(self, visit_id: str, score: float) -> None: data = { "visit_id": visit_id, "plate_weight": 0, "segment_weight": 0, "final_score": score } response = self.db.insert("purchasing_power", data) if not response or len(response) == 0: raise RuntimeError("Failed to insert purchasing power score")