Add log_handler/src/log_handler/core.py

log_handler/src/log_handler/core.py

@@ -0,0 +1,85 @@
+import re
+import json
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import List
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class TemperatureLog:
+    timestamp: str
+    temperature: float
+    sensor_temp: float
+
+
+class LogParseError(Exception):
+    """Custom exception raised when a log line cannot be parsed."""
+
+
+_TIMESTAMP_PATTERN = re.compile(r"(?P<timestamp>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2})")
+_TEMPERATURE_PATTERN = re.compile(
+    r"temperature:\s*(?P<temperature>-?\d+(?:\.\d+)?)C.*sensor_temp:\s*(?P<sensor_temp>-?\d+(?:\.\d+)?)C"
+)
+
+
+def _validate_log_file_path(log_file_path: str) -> Path:
+    path = Path(log_file_path)
+    if not path.exists() or not path.is_file():
+        raise FileNotFoundError(f"Logfile '{log_file_path}' not found or not a file.")
+    return path
+
+
+def _parse_line(line: str) -> TemperatureLog | None:
+    timestamp_match = _TIMESTAMP_PATTERN.search(line)
+    temp_match = _TEMPERATURE_PATTERN.search(line)
+
+    if timestamp_match and temp_match:
+        timestamp = timestamp_match.group("timestamp")
+        try:
+            # Validate timestamp format
+            datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S")
+        except ValueError:
+            logger.warning("Invalid timestamp format in line: %s", line.strip())
+            return None
+
+        try:
+            temperature = float(temp_match.group("temperature"))
+            sensor_temp = float(temp_match.group("sensor_temp"))
+        except (ValueError, TypeError):
+            logger.warning("Invalid temperature values in line: %s", line.strip())
+            return None
+
+        return TemperatureLog(timestamp=timestamp, temperature=temperature, sensor_temp=sensor_temp)
+    return None
+
+
+def parse_log_file(log_file_path: str) -> List[TemperatureLog]:
+    """
+    Parse a raw camera log file and extract structured temperature and sensor information.
+
+    Args:
+        log_file_path (str): Path to the logfile.
+
+    Returns:
+        List[TemperatureLog]: List of structured temperature logs.
+    """
+    logger.debug("Starting to parse log file: %s", log_file_path)
+    path = _validate_log_file_path(log_file_path)
+    results: List[TemperatureLog] = []
+
+    with path.open("r", encoding="utf-8") as f:
+        for line_no, line in enumerate(f, start=1):
+            parsed = _parse_line(line)
+            if parsed:
+                results.append(parsed)
+            else:
+                logger.debug("Line %d skipped: no valid data.", line_no)
+
+    assert all(isinstance(t, TemperatureLog) for t in results), "Parsed data validation failed."
+    logger.info("Parsed %d valid temperature entries from %s", len(results), log_file_path)
+    return results