Signal-Based & Bus-Based Testing

"""CAN bus monitoring: collect and analyse all signals."""
import can, cantools, time, threading
from collections import defaultdict
from typing import Dict, List

class CanBusMonitor:
    def __init__(self, channel: str, db_path: str):
        self.db = cantools.database.load_file(db_path)
        self.bus = can.interface.Bus(channel, bustype="socketcan")
        self.signal_log: Dict[str, List[tuple]] = defaultdict(list)
        self._running = False

    def start(self):
        self._running = True
        self._thread = threading.Thread(target=self._receive_loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False
        self._thread.join(timeout=2.0)

    def _receive_loop(self):
        while self._running:
            msg = self.bus.recv(timeout=0.01)
            if msg:
                try:
                    decoded = self.db.decode_message(
                        msg.arbitration_id, msg.data)
                    for name, value in decoded.items():
                        self.signal_log[name].append((msg.timestamp, value))
                except Exception:
                    pass

    def check_timing(self, signal: str, expected_period_ms: float,
                      tolerance_pct: float = 10.0) -> dict:
        entries = self.signal_log.get(signal, [])
        if len(entries) < 2:
            return {"ok": False, "reason": "insufficient samples"}
        periods = [(entries[i+1][0]-entries[i][0])*1000
                    for i in range(len(entries)-1)]
        avg_period = sum(periods) / len(periods)
        max_jitter = max(abs(p - expected_period_ms) for p in periods)
        ok = max_jitter <= expected_period_ms * tolerance_pct / 100
        return {"ok": ok, "avg_period_ms": avg_period,
                "max_jitter_ms": max_jitter}

"""Fault injection: corrupt or withhold signals to test ECU responses."""
import can, time

class FaultInjector:
    def __init__(self, bus: can.BusABC, db):
        self.bus = bus
        self.db = db

    def inject_out_of_range(self, signal_name: str,
                            value: float, duration_s: float):
        """Send signal with value outside physical range."""
        msg_def = self.db.get_message_by_signal_name(signal_name)
        raw_data = msg_def.encode({signal_name: value},
                                   scaling=False)  # bypass range check
        deadline = time.monotonic() + duration_s
        while time.monotonic() < deadline:
            self.bus.send(can.Message(
                arbitration_id=msg_def.frame_id,
                data=raw_data, is_extended_id=False))
            time.sleep(0.010)  # 10ms cycle

    def withhold_signal(self, signal_name: str, duration_s: float):
        """Stop sending a signal to trigger timeout behaviour."""
        # Simply do not send -- the ECU should detect timeout
        time.sleep(duration_s)