Add artifact_2/src/artifact_2/core.py

artifact_2/src/artifact_2/core.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone, timedelta
+from typing import List
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class LogEntry:
+    """Repräsentiert einen einzelnen Log-Eintrag mit Zeitinformationen."""
+
+    epoch_ms: int
+    monotonic_ns: int
+    tz_offset_minutes: int
+    run_id: str
+    step_id: str
+
+    def __post_init__(self) -> None:
+        # Eingabevalidierung (CI-Ready & Input Validation Required)
+        assert isinstance(self.epoch_ms, int) and self.epoch_ms >= 0, 'epoch_ms muss eine positive Ganzzahl sein.'
+        assert isinstance(self.monotonic_ns, int) and self.monotonic_ns >= 0, 'monotonic_ns muss eine positive Ganzzahl sein.'
+        assert isinstance(self.tz_offset_minutes, int), 'tz_offset_minutes muss eine Ganzzahl sein.'
+        assert isinstance(self.run_id, str) and self.run_id.strip(), 'run_id muss ein nicht-leerer String sein.'
+        assert isinstance(self.step_id, str) and self.step_id.strip(), 'step_id muss ein nicht-leerer String sein.'
+
+    def to_datetime(self) -> datetime:
+        tz = timezone(timedelta(minutes=self.tz_offset_minutes))
+        return datetime.fromtimestamp(self.epoch_ms / 1000.0, tz=tz)
+
+
+def check_timestamp_consistency(log_entries: List[LogEntry]) -> bool:
+    """Prüft die Konsistenz von Zeitstempeln.
+
+    Stellt sicher:
+      - epochale Zeit und monotone Zeit steigen monoton.
+      - höchstens ein Zeitzonenwechsel tritt auf.
+      - keine negativen Deltas.
+
+    Rückgabe:
+        True, wenn konsistent; sonst False.
+    """
+
+    if not isinstance(log_entries, list) or not all(isinstance(e, LogEntry) for e in log_entries):
+        raise TypeError('log_entries muss eine Liste von LogEntry-Instanzen sein.')
+
+    if not log_entries:
+        logger.info('Leere Eingabeliste – gilt als konsistent.')
+        return True
+
+    # Nach Zeit sortieren zur Sicherheit
+    log_entries_sorted = sorted(log_entries, key=lambda e: e.epoch_ms)
+    tz_switches = set()
+
+    prev_entry = log_entries_sorted[0]
+    previous_dt = prev_entry.to_datetime()
+    previous_mono = prev_entry.monotonic_ns
+    tz_switches.add(prev_entry.tz_offset_minutes)
+
+    for entry in log_entries_sorted[1:]:
+        current_dt = entry.to_datetime()
+        current_mono = entry.monotonic_ns
+
+        # Zeitzonenwechsel speichern
+        tz_switches.add(entry.tz_offset_minutes)
+
+        # Zeitdifferenzen prüfen
+        delta_epoch = (current_dt - previous_dt).total_seconds()
+        delta_mono = current_mono - previous_mono
+
+        if delta_epoch < 0 or delta_mono < 0:
+            logger.warning('Negative Zeitdifferenz erkannt zwischen %s und %s', previous_dt, current_dt)
+            return False
+
+        previous_dt = current_dt
+        previous_mono = current_mono
+
+    # Gültig: höchstens ein Zeitzonenwechsel
+    if len(tz_switches) > 2:
+        logger.warning('Mehrere Zeitzonenwechsel erkannt: %s', tz_switches)
+        return False
+
+    return True