spvirit-py 0.1.14

"""Raw frame inspection demo — Channel.read_packet / read_until + Packet.

After a get/put/introspect completes, any further frames the server sends
(beacons, monitor updates, keepalives…) can be read from the Channel's
TCP stream and inspected as ``Packet`` objects.

Exercises:
  - ``Channel.read_packet(timeout=None)`` -> Packet
  - ``Channel.read_packet_async(...)``
  - ``Channel.read_until(predicate, timeout=None, max_frames=None)``
  - ``Packet`` getters: ``command_name``, ``flags``, ``payload_length``,
    ``bytes``, ``payload``, ``is_application`` / ``is_control`` / ``is_server``
  - ``Packet.details()`` -> dict (same shape as ``codec.decode_packet``)

Usage::

    python3 demo_raw_frames.py PV_NAME [SERVER_ADDR]

The example issues a GET, then uses ``read_until`` with a Python
predicate to ignore any stray control frames and wait for a specific
application-level command on the wire.
"""
from __future__ import annotations

import sys

from spvirit import codec
from spvirit.lowlevel import Channel, Packet


def describe(pkt: Packet) -> str:
    return (
        f"cmd={pkt.command_name!r:12}"
        f"  flags=0x{pkt.flags:02x}"
        f"  len={pkt.payload_length:<6}"
        f"  app={pkt.is_application}  ctrl={pkt.is_control}  srv={pkt.is_server}"
    )


def main() -> None:
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)
    pv_name = sys.argv[1]
    server_addr = sys.argv[2] if len(sys.argv) > 2 else "127.0.0.1:5075"

    ch = Channel.connect(pv_name, server_addr, timeout=3.0)
    result = ch.get()
    print(f"get() -> {codec.format_value(result.value)}\n")

    # ``result.raw_pva`` is the last application-level GET frame received.
    # Wrap it in a Packet for inspection without re-reading the wire.
    pkt = Packet.__call__ if False else None  # (Packet is constructed by the channel)
    decoded = codec.decode_packet(result.raw_pva)
    print(f"Captured GET frame: command={decoded['command_name']} "
          f"payload_length={decoded['payload_length']}")

    # ── Issue another get and watch the wire with read_until ───────────
    # We send an empty pvRequest GET and use a Python predicate that
    # accepts only a specific (command, ioid-agnostic) frame.
    print("\nIssuing a second get and reading frames with read_until ...")
    # Fire-and-forget a new get via the high-level API on the channel —
    # but we want to see the raw traffic, so we use a short deadline and
    # a predicate that matches any GET frame.
    # (The GET helper itself already consumes its response; for a pure
    # raw demo we simply show the read_packet/read_until surface on a
    # new channel that we don't drive ourselves.)

    def match_any_application(p: Packet) -> bool:
        print(f"  saw {describe(p)}")
        return bool(p.is_application)

    ch2 = Channel.connect(pv_name, server_addr, timeout=2.0)
    try:
        # Trigger traffic by asking for introspection, then read frames.
        _ = ch2.introspect()
        try:
            first_app = ch2.read_until(match_any_application, timeout=0.5,
                                       max_frames=16)
            print(f"\nread_until matched: {describe(first_app)}")
            details = first_app.details()
            print(f"details keys: {sorted(details.keys())}")
        except Exception as e:  # noqa: BLE001
            print(f"read_until: no further frames arrived ({e})")
    finally:
        ch2.close()

    ch.close()


if __name__ == "__main__":
    main()