#!/usr/bin/env python3
"""SHDR adapter feeding the OtOpcUa MTConnect integration-test fixture live data.

The mtconnect/agent image ships *only* the agent binary plus its schemas and styles --
there is no bundled simulator. Without an adapter the Agent answers /probe correctly but
reports every observation UNAVAILABLE forever, which cannot prove the two things this
fixture exists for: that ReadAsync returns real coerced values, and that the driver's
/sample long poll delivers OnDataChange from a genuinely moving stream.

So this is the data source. It speaks SHDR (the Agent's plain-text adapter protocol) over
TCP: the Agent dials IN as the client (see agent.cfg's Adapters block), and every line we
write is `<timestamp>|<key>|<value>[|<key>|<value>...]`, where <key> matches a DataItem's
`name` attribute in Devices.xml (falling back to its `id` for the unnamed ones).

Deliberate behaviours -- each one is asserted on by the integration suite:

  * Xact / Xload / Cspeed move on EVERY tick. "The value changed between two reads" is the
    only assertion that distinguishes a live stream from a cached /current snapshot.
  * Pcount (PART_COUNT) increments on a slower cadence, so it is both a *changing* value
    and an integer -- the live half of the UPPER_SNAKE PART_COUNT -> Int64 inference.
  * Pblock is NEVER written. The Agent therefore reports it UNAVAILABLE for the life of
    the fixture, which is what exercises UNAVAILABLE -> BadNoCommunication.
  * Conditions are re-asserted periodically rather than once at connect, so a driver that
    attaches mid-run still sees them.

Pure standard library on purpose: the container is a stock `python:*-alpine` with this
file bind-mounted, so the fixture needs no image build step at all (unlike the Modbus /
S7 fixtures, whose simulators do).
"""

from __future__ import annotations

import argparse
import datetime
import math
import selectors
import socket
import sys
import threading
import time

# How often the periodic feed writes a batch of observations. Comfortably faster than the
# driver's default SampleIntervalMs (1000) so a subscription sees several distinct chunks
# inside a short test timeout.
TICK_SECONDS = 0.25

# The Agent's PING/PONG watchdog: it sends `* PING` and expects `* PONG <ms>` back. The
# number is how long the Agent should wait before declaring us dead.
PONG_TIMEOUT_MS = 10000

EXECUTION_STATES = ("ACTIVE", "READY", "INTERRUPTED", "ACTIVE", "STOPPED")
CONTROLLER_MODES = ("AUTOMATIC", "MANUAL", "AUTOMATIC", "SEMI_AUTOMATIC")


def timestamp() -> str:
    """UTC in the ISO-8601 form the Agent expects (milliseconds, trailing Z)."""
    now = datetime.datetime.now(datetime.timezone.utc)

    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"


class Feed:
    """One connected Agent. Owns the socket for the life of that connection."""

    def __init__(self, conn: socket.socket, peer: str) -> None:
        self._conn = conn
        self._peer = peer
        self._started = time.monotonic()
        self._ticks = 0

    # ---- wire ----

    def send(self, line: str) -> None:
        self._conn.sendall((line + "\n").encode("utf-8"))

    def send_observations(self, pairs: list[tuple[str, str]]) -> None:
        """One SHDR line carrying several key/value pairs, all sharing one timestamp."""
        if not pairs:
            return

        body = "|".join(f"{key}|{value}" for key, value in pairs)
        self.send(f"{timestamp()}|{body}")

    # ---- content ----

    def send_initial_state(self) -> None:
        """Everything that is not periodic, sent once as soon as the Agent attaches."""
        self.send_observations(
            [
                ("Favail", "AVAILABLE"),
                ("Pprogram", "OTOPCUA-FIXTURE.NC"),
                # DATA_SET representation: the Agent renders these as <Entry key=...> children,
                # which is precisely the wire shape the driver demotes to String / BadNotSupported.
                ("Pvars", "toolNumber=7 offsetX=1.25 offsetZ=-0.5"),
            ]
        )
        self.send_conditions()

    def send_conditions(self) -> None:
        """SHDR condition form: <key>|<level>|<nativeCode>|<nativeSeverity>|<qualifier>|<text>."""
        stamp = timestamp()
        self.send(f"{stamp}|Xtravel|normal||||")
        self.send(f"{stamp}|Plogic|normal||||")

    def tick(self) -> None:
        """One periodic batch. Called every TICK_SECONDS."""
        self._ticks += 1
        elapsed = time.monotonic() - self._started

        # Continuously-moving SAMPLEs. Rounded to 4 dp so the value is a clean double on
        # the wire and still differs between any two consecutive ticks.
        position = round(120.0 + 40.0 * math.sin(elapsed / 2.0), 4)
        load = round(45.0 + 15.0 * math.sin(elapsed / 3.7), 4)
        speed = round(1500.0 + 250.0 * math.sin(elapsed / 5.0), 4)

        pairs: list[tuple[str, str]] = [
            ("Xact", f"{position}"),
            ("Xload", f"{load}"),
            ("Cspeed", f"{speed}"),
        ]

        # PART_COUNT: an integer EVENT that also moves. Slower than the samples so it reads
        # like a real counter rather than a signal.
        pairs.append(("Pcount", str(100 + int(elapsed // 5))))
        pairs.append(("Pline", str(10 + (self._ticks % 90))))

        self.send_observations(pairs)

        # TIME_SERIES form: <key>|<sampleCount>|<sampleRate>|<v1> <v2> ...
        series = " ".join(
            f"{round(21.5 + 0.5 * math.sin(elapsed + i / 4.0), 3)}" for i in range(8)
        )
        self.send(f"{timestamp()}|Ctemps|8|100|{series}")

        # Controlled-vocabulary EVENTs, on their own slow cadences.
        if self._ticks % 28 == 1:
            self.send_observations(
                [("Pexec", EXECUTION_STATES[(self._ticks // 28) % len(EXECUTION_STATES)])]
            )
        if self._ticks % 44 == 1:
            self.send_observations(
                [("Cmode", CONTROLLER_MODES[(self._ticks // 44) % len(CONTROLLER_MODES)])]
            )

        # Re-assert conditions occasionally so a late-attaching driver still sees them.
        if self._ticks % 60 == 0:
            self.send_conditions()

        # NOTE: Pblock is never written, on purpose. See the module docstring.


def handle(conn: socket.socket, peer: str) -> None:
    feed = Feed(conn, peer)
    print(f"[adapter] agent connected from {peer}", flush=True)

    selector = selectors.DefaultSelector()
    selector.register(conn, selectors.EVENT_READ)

    pending = b""
    next_tick = time.monotonic()

    try:
        feed.send_initial_state()

        while True:
            now = time.monotonic()
            if now >= next_tick:
                feed.tick()
                # Absolute schedule, not `now + TICK`: drift-free, and a slow tick cannot
                # compound into an ever-widening gap the Agent reads as a stall.
                next_tick += TICK_SECONDS
                if next_tick < now:
                    next_tick = now + TICK_SECONDS

            for _ in selector.select(timeout=max(0.0, next_tick - time.monotonic())):
                chunk = conn.recv(4096)
                if not chunk:
                    print(f"[adapter] agent {peer} closed the connection", flush=True)

                    return

                pending += chunk
                while b"\n" in pending:
                    line, pending = pending.split(b"\n", 1)
                    text = line.decode("utf-8", errors="replace").strip()
                    # The Agent's heartbeat. Answering is mandatory: an Agent that gets no
                    # PONG tears the connection down and every observation goes UNAVAILABLE.
                    if text.startswith("* PING"):
                        feed.send(f"* PONG {PONG_TIMEOUT_MS}")
    except (BrokenPipeError, ConnectionResetError, OSError) as ex:
        print(f"[adapter] connection to {peer} ended: {ex}", flush=True)
    finally:
        selector.close()
        try:
            conn.close()
        except OSError:
            pass


def main() -> int:
    parser = argparse.ArgumentParser(description="SHDR adapter for the OtOpcUa MTConnect fixture.")
    parser.add_argument("--host", default="0.0.0.0", help="Bind address (default: all interfaces).")
    parser.add_argument("--port", type=int, default=7878, help="SHDR listen port (default: 7878).")
    args = parser.parse_args()

    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind((args.host, args.port))
    listener.listen(8)
    print(f"[adapter] listening on {args.host}:{args.port}", flush=True)

    try:
        while True:
            conn, addr = listener.accept()
            conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            # One thread per Agent connection. The Agent reconnects after any restart, and
            # a serialized accept loop would leave the new connection unserved while the
            # dead one's socket was still being reaped.
            threading.Thread(
                target=handle, args=(conn, f"{addr[0]}:{addr[1]}"), daemon=True
            ).start()
    except KeyboardInterrupt:
        return 0
    finally:
        listener.close()


if __name__ == "__main__":
    sys.exit(main())