Tester Present & Keep-Alive Strategies

#!/usr/bin/env python3
# Thread-safe TesterPresent keep-alive for UDS diagnostic sessions
import threading, time, udsoncan
from udsoncan.client import Client

class SessionKeepAlive:
    def __init__(self, client: Client, interval_s: float = 2.0):
        self._client   = client
        self._interval = interval_s
        self._active   = False
        self._thread   = None
        self._lock     = threading.Lock()

    def start(self):
        self._active = True
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self):
        self._active = False
        if self._thread:
            self._thread.join(timeout=self._interval + 1)

    def _run(self):
        while self._active:
            try:
                with self._lock:
                    # Suppress positive response (0x3E 0x80) — no CAN frame from ECU
                    self._client.tester_present(suppress_positive_response=True)
            except Exception:
                pass  # log but do not crash keep-alive thread
            time.sleep(self._interval)

# Usage:
# keepalive = SessionKeepAlive(client, interval_s=2.0)
# keepalive.start()
# try:
#     long_running_operation(client)  # takes 30s; S3 maintained by keepalive
# finally:
#     keepalive.stop()