engine.py

# SPDX-License-Identifier: GPL-2.0-only
# Copyright (C) 2026 Cyber Defense Institute, Inc.
"""
engine.py  --  SERINUS OT  core logic (v4)

records = history of all detected MACs (single source of truth). MAC-unique, newest first.
State is either Pending (要確認) or Registered (登録済み).

Changes (v4):
  - Added ts_first (first-seen timestamp) to Record. ts = last-seen timestamp.
  - Added volume setting (High/Mid/Low/Mute) to EngineState.
  - On re-detection of the same MAC: only ts (last-seen) is updated; ts_first is unchanged.
  - New MAC detected in GUARD mode: ts_first = ts = now.

Modes:
  LEARNING = bulk registration / initial setup. No alert.
             Communicating devices are automatically promoted to Registered.
             Time limit is controlled by the app layer.
  GUARD    = New MACs are flagged as Pending (alert sounds).

Operations:
  toggle(mac)  ... toggle between Pending <-> Registered.
  silence()    ... stop the current alarm only. New detections re-trigger automatically.

★ Inside the HAL boundary (independent of OS, GUI, and scapy). Translatable to C for hardware.
"""

from dataclasses import dataclass, field
from typing import Callable, List, Optional

from parser import FrameEvent

MODE_LEARNING = "LEARNING"
MODE_GUARD    = "GUARD"

ST_PENDING    = "要確認"
ST_REGISTERED = "登録済み"

VOLUME_HIGH   = "大"
VOLUME_MID    = "中"
VOLUME_LOW    = "小"
VOLUME_MUTE   = "ミュート"
VOLUME_LEVELS = [VOLUME_HIGH, VOLUME_MID, VOLUME_LOW, VOLUME_MUTE]


@dataclass
class Record:
    mac:      str
    ip:       str
    proto:    str
    ts:       float          # last-seen timestamp
    state:    str = ST_PENDING
    ts_first: float = 0.0    # first-seen timestamp (0.0 = record loaded from previous session)


@dataclass
class EngineState:
    mode:        str   = MODE_LEARNING
    learn_start: float = 0.0
    records:     List[Record] = field(default_factory=list)
    alarming:    bool  = False
    muted:       bool  = False
    volume:      str   = VOLUME_MID


class Engine:
    """
    Callbacks:
      on_alert()  called when a new unregistered MAC triggers an alarm (buzzer / ALARM screen)
      on_clear()  called to stop the buzzer
      on_state()  called on any state change (triggers redraw)
    """

    def __init__(self, state: Optional[EngineState] = None,
                 on_alert: Callable[[], None] = None,
                 on_clear: Callable[[], None] = None,
                 on_state: Callable[[], None] = None):
        self.s = state or EngineState()
        self.on_alert = on_alert or (lambda: None)
        self.on_clear = on_clear or (lambda: None)
        self.on_state = on_state or (lambda: None)

    # -- counts ---------------------------------------------------------
    def registered_count(self) -> int:
        return sum(1 for r in self.s.records if r.state == ST_REGISTERED)

    def pending_count(self) -> int:
        return sum(1 for r in self.s.records if r.state == ST_PENDING)

    def is_registered(self, mac) -> bool:
        r = self._find(mac)
        return bool(r and r.state == ST_REGISTERED)

    def _find(self, mac) -> Optional[Record]:
        for r in self.s.records:
            if r.mac == mac:
                return r
        return None

    # -- frame event handler (core) ------------------------------------
    def handle_event(self, ev: FrameEvent, now: float):
        mac = ev.src_mac
        rec = self._find(mac)

        if rec is not None:
            # existing MAC: update ts (last-seen) only; ts_first stays unchanged.
            rec.ip    = ev.ip_guess or rec.ip
            rec.proto = ev.proto
            rec.ts    = now
            # re-communication during LEARNING -> promote to Registered
            if self.s.mode == MODE_LEARNING and rec.state == ST_PENDING:
                rec.state = ST_REGISTERED
            self.on_state()
            return

        # new MAC
        new_rec = Record(
            mac=mac,
            ip=ev.ip_guess or "?",
            proto=ev.proto,
            ts=now,
            ts_first=now,   # set first-seen timestamp
        )
        if self.s.mode == MODE_LEARNING:
            new_rec.state = ST_REGISTERED
            self.s.records.insert(0, new_rec)
            self.s.records = self.s.records[:255]
            self.on_state()
            return

        # new MAC in GUARD mode -> Pending + alert
        new_rec.state = ST_PENDING
        self.s.records.insert(0, new_rec)
        self.s.records = self.s.records[:255]
        was_alarming   = self.s.alarming
        self.s.alarming = True
        self.s.muted    = False
        if not was_alarming:
            self.on_alert()
        self.on_state()

    # -- mode control --------------------------------------------------
    def start_learning(self, now: float):
        self.s.mode        = MODE_LEARNING
        self.s.learn_start = now
        self.s.alarming    = False
        self.s.muted       = False
        self.on_clear()
        self.on_state()

    def to_guard(self):
        self.s.mode = MODE_GUARD
        self.on_state()

    # -- state toggle --------------------------------------------------
    def toggle(self, mac: str) -> Optional[str]:
        rec = self._find(mac)
        if not rec:
            return None
        rec.state = ST_PENDING if rec.state == ST_REGISTERED else ST_REGISTERED
        self.on_state()
        return rec.state

    # -- alarm control -------------------------------------------------
    def silence(self) -> bool:
        """Stop the current alarm only. New detections will re-trigger automatically."""
        if self.s.alarming:
            self.s.alarming = False
            self.s.muted    = True
            self.on_clear()
            self.on_state()
            return True
        return False

    def reset_all(self, now: float):
        self.s.records.clear()
        self.s.alarming    = False
        self.s.muted       = False
        self.s.learn_start = now
        self.on_clear()
        self.on_state()

    def registered_list(self):
        return [(r.mac, r.ip) for r in self.s.records
                if r.state == ST_REGISTERED]

    # -- latest pending index (used to focus history cursor after silence) --
    def latest_pending_index(self) -> int:
        """Return the index of the most recently detected Pending record. Returns 0 if none."""
        for i, r in enumerate(self.s.records):
            if r.state == ST_PENDING:
                return i
        return 0


def remaining_mmss(start: float, window: float, now: float) -> str:
    rem = max(0, int(start + window - now))
    return f"{rem // 60:02d}:{rem % 60:02d}"