From 3b2cf434e851bcdcafbd4131d1a48acfd42810ac Mon Sep 17 00:00:00 2001 From: Source robotics <30388414+PCrnjak@users.noreply.github.com> Date: Thu, 5 Nov 2026 22:31:46 +0100 Subject: [PATCH] Add latest-frame acquisition and tracking health --- .github/workflows/ci.yml | 5 +- README.md | 39 ++- docs/LATEST_FRAMES.md | 125 +++++++++ examples/latest_frames.py | 31 +++ pyproject.toml | 2 +- src/source_robotics_leap_motion/__init__.py | 6 +- src/source_robotics_leap_motion/_latest.py | 149 +++++++++++ src/source_robotics_leap_motion/cli.py | 106 +++++--- src/source_robotics_leap_motion/errors.py | 4 + src/source_robotics_leap_motion/health.py | 134 +++++++++ src/source_robotics_leap_motion/tracker.py | 89 +++++- tests/test_cli_runtime.py | 21 ++ tests/test_latest.py | 283 ++++++++++++++++++++ tests/test_managed_runtime.py | 6 +- tools/benchmark_latest.py | 95 +++++++ 15 files changed, 1042 insertions(+), 53 deletions(-) create mode 100644 docs/LATEST_FRAMES.md create mode 100644 examples/latest_frames.py create mode 100644 src/source_robotics_leap_motion/_latest.py create mode 100644 src/source_robotics_leap_motion/health.py create mode 100644 tests/test_latest.py create mode 100644 tools/benchmark_latest.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e97c271..f4ca6b0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,9 +24,10 @@ jobs: python-version: ${{ matrix.python }} - run: python -m pip install .[dev] - run: python -m pytest -q - - run: python -m ruff check src tests examples tools/build_runtime_release.py tools/configure_release.py tools/check_release.py tools/probe_private_runtime.py - - run: python -m ruff format --check src tests examples tools/build_runtime_release.py tools/configure_release.py tools/check_release.py tools/probe_private_runtime.py + - run: python -m ruff check src tests examples tools/build_runtime_release.py tools/configure_release.py tools/check_release.py tools/probe_private_runtime.py tools/benchmark_latest.py + - run: python -m ruff format --check src tests examples tools/build_runtime_release.py tools/configure_release.py tools/check_release.py tools/probe_private_runtime.py tools/benchmark_latest.py - run: source-leap --demo --frames 3 --json + - run: source-leap --demo --latest --health --frames 3 --json build: runs-on: ubuntu-latest steps: diff --git a/README.md b/README.md index b129a47..5c39b3d 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ -**Source Robotics Leap Motion - 0.1.0b1 beta** +**Source Robotics Leap Motion** A Python package for the **Original Leap Motion Controller**, with owned hand data, recording/replay, and configurable coordinate mapping for custom robot integrations. It runs on current Python without `LeapPython.pyd` or a Python 3.8 process. -The beta is published on [PyPI](https://pypi.org/project/source-robotics-leap-motion/). +Beta **0.1.0b1** is published on [PyPI](https://pypi.org/project/source-robotics-leap-motion/). Install it, then start tracking: ```powershell @@ -17,6 +17,10 @@ require Windows approval. See [release status and validation](docs/BETA_VALIDATI and [publication steps](docs/RELEASING.md). Linux/macOS currently use an installed vendor tracking service and remain experimental for hardware use. +This checkout contains **0.1.0b2 development**, adding latest-frame acquisition and +tracking health. These additions are available from source; they are not in the +published 0.1.0b1 package. See [the latest-frame guide](docs/LATEST_FRAMES.md). + **Try it in this project's virtual environment.** Python 3.15.2 and the package are already installed in `.venv`. No activation is necessary: @@ -139,6 +143,37 @@ finger/arm rotations. The legacy backend exposes finger joint positions when sup by protocol v6, derives palm orientation from normal/direction, and reports unavailable bone rotations and pinch distance as `None`. It never substitutes a made-up rotation. +**Latest frames and health (0.1.0b2 development).** A slow application can request +the newest unread frame instead of falling behind the camera stream: + +```python +from source_robotics_leap_motion import Tracker + +with Tracker(mode="latest", max_frame_age=0.1) as tracker: + frame = tracker.read(timeout=0.25) + print(frame.right_hand) + print(tracker.health.to_dict()) +``` + +One background thread owns the source and replaces a single pending frame. Skipped +frames are counted. Reads wait for a new frame and never repeat a consumed frame. +An unread frame older than `max_frame_age` is discarded; if no fresh frame arrives +before the read deadline, `StaleFrameError` (a `TrackingTimeoutError`) is raised. +The default remains sequential, which is appropriate when recording every frame. + +`tracker.health` reports connection state, recent received FPS, local frame age, +received/delivered/skipped/duplicate counts, timeouts, timestamp resets, and hand +count. No hands still means the camera is streaming. Age is measured **since Python +received the frame**, not end-to-end camera latency. It cannot detect every delay +inside a service or transport. See [semantics and validation](docs/LATEST_FRAMES.md). + +```powershell +.\.venv\Scripts\source-leap.exe --latest --health --duration 10 +.\.venv\Scripts\source-leap.exe --demo --latest --health --frames 60 --json +``` + +Health JSON goes to stderr; frame JSON on stdout and the recording schema are unchanged. + **Record and replay.** Existing recordings are protected by default: ```powershell diff --git a/docs/LATEST_FRAMES.md b/docs/LATEST_FRAMES.md new file mode 100644 index 0000000..3e3aa53 --- /dev/null +++ b/docs/LATEST_FRAMES.md @@ -0,0 +1,125 @@ +**Latest-frame acquisition and tracking health — 0.1.0b2 development** + +These features are available in the source checkout and the locally built b2 wheel. +PyPI currently provides 0.1.0b1. The runtime engine, models, driver, frame data model, +and recording format are unchanged; b2 reuses the pinned b1 runtime download. + +Use `Tracker(mode="latest", max_frame_age=0.1)` when application work runs slower +than tracking. Use the default `Tracker()` when each frame matters, for example +recording a complete session. `Tracker.demo(mode="latest")` provides paced synthetic +data without hardware. An unpaced demo is rejected in latest mode. + +```python +from source_robotics_leap_motion import ConnectionError, Tracker, TrackingTimeoutError + +with Tracker(mode="latest", max_frame_age=0.1) as tracker: + try: + frame = tracker.read(timeout=0.25) + print(frame.right_hand) + except TrackingTimeoutError: # Includes StaleFrameError. + print("No fresh tracking frame", tracker.health.to_dict()) + except ConnectionError: + print("Device or service disconnected", tracker.health.to_dict()) +``` + +The background reader opens, polls, and closes the source on one thread, including +native sources. It keeps one pending immutable frame and a bounded statistics window +of 120 arrival times. It replaces unread frames as newer frames arrive. Calling +`read()` consumes the slot; another read waits for the next distinct frame. It does +not return the same cached frame repeatedly. Multiple consumers share this one +slot; applications needing independent streams should open separate trackers. + +The age limit applies to local receipt time. Expired pending frames are discarded, +and `read()` waits up to its timeout for a fresh replacement. A deadline with no +frames ever received raises `TrackingTimeoutError`. A deadline after the last +distinct frame has become too old raises `StaleFrameError`, a subclass of that +exception. Exact consecutive duplicate frame identities do not refresh freshness. +Disconnections invalidate any pending frame and propagate the source error. A finite +source can deliver its last fresh pending frame before `StopIteration`. + +Context exit wakes blocked consumers and joins the worker. Normal reads poll with +a 50 ms bound. Custom `FrameSource` implementations must honor `read(timeout)` and +return `None` on a polling timeout. A source that ignores deadlines can prevent +shutdown: after five seconds, close raises `ConnectionError` and reopening is refused +while the old worker is alive. The package never closes a native handle concurrently +with its owning reader. `reconnect()` explicitly closes and opens a new session; +there is no automatic reconnect or robot motion restart. + +`tracker.health` is a thread-safe immutable `TrackingHealth` snapshot. It is also +available in sequential mode. Reading it never polls hardware. + +| Field | Meaning | +| --- | --- | +| `status` | `closed`, `starting`, `waiting`, `streaming`, `stale`, `disconnected`, `ended`, or `error` | +| `received_frames` | Valid frame packets received in this session, including exact consecutive duplicates | +| `delivered_frames` | Frames returned to application reads | +| `skipped_frames` | Frames deliberately omitted by latest mode: replacements, duplicates, expirations, or discarded pending data on failure/close | +| `duplicate_frames` | Consecutive packets with the same clock domain, device ID, frame ID, and timestamp | +| `read_timeouts` | Application reads that reached their deadline; internal polling timeouts are excluded | +| `timestamp_resets` | Regressing source timestamps within the same clock domain and device | +| `received_fps` | Distinct local arrivals per second over at most 120 frames; zero when not streaming | +| `last_frame_age_s` | Monotonic time since Python received the last distinct frame; `None` before the first | +| `last_frame_id`, `last_frame_timestamp_us`, `device_id`, `clock_domain` | Last distinct frame identity | +| `last_hand_count` | Hands in the last distinct frame; zero is valid tracking data | +| `last_error` | Latest session error text, retained after close | + +Counts reset on opening or reconnecting. They remain available after close. Closed +snapshots report zero current FPS. In latest mode, at a synchronized instant, +received frames equal delivered plus skipped plus the optional pending frame. +Skipped counts are **not** a measure of camera or network packet loss. Sequential +mode preserves delivery of duplicates and reports zero deliberately skipped frames. + +`streaming` becomes `stale` when the local age exceeds `max_frame_age`, in both modes. +Sequential mode reports this observation but preserves its existing read behavior; +it does not reject queued frames. A camera streaming empty-hand frames remains +`streaming`. `last_hand_count` is historical during a stale or disconnected period; +readers must examine status as well as hand count. + +Frame timestamps belong to the named service clock. They are never subtracted from +Python's monotonic clock to invent an absolute latency. Local receipt age cannot +prove when exposure happened or detect all upstream buffering. Latest mode drains +the connection continuously, but it is not a hard real-time or robot safety system. + +The CLI accepts `--latest`, `--max-frame-age SECONDS`, and `--health` for monitoring +and recording. Latest-mode recordings intentionally omit overwritten frames. Replay +remains sequential and also accepts `--health`. Health snapshots are JSON records +under `tracking_health` on stderr, roughly once per second while streaming and on +exit after tracking opens. Frame JSON on stdout and recording files remain unchanged. + +```powershell +.\.venv\Scripts\source-leap.exe --latest --health --duration 10 +.\.venv\Scripts\python.exe tools\benchmark_latest.py --duration 10 --consumer-rate 20 +``` + +**Live validation, 5 November 2026.** On this PC's Original Leap Motion Controller +and existing legacy service, two application consumers ran at 20 Hz for ten seconds. +Latest mode received 783 packets and delivered 200. At the final in-session snapshot, +582 were skipped and one was pending. No duplicates, read timeouts, timestamp resets, +or tracking errors were recorded. The recent arrival rate was about 80 FPS. + +The latest stream's receipt age at sampling was **6.40 ms median, 12.07 ms p95, +27.36 ms maximum**. The sequential reader accumulated a **7.39-second timestamp +gap** behind the latest reader by the last sample, comparing the same service clock. +Combined CPU time for both Python clients was about 0.78% of one core for this run. +No hands were visible, so this validates queue behavior and camera streaming, not +full-skeleton processing cost, improved accuracy, or absolute camera latency. +Raw results are kept locally in `.tools/latest-benchmark-b2.json`. + +A separate one-minute slow-consumer run received 4,749 frames (including startup +warmup), with no duplicates, read timeouts, clock resets, or tracking errors. Close +joined the worker and left received counts equal to delivered plus skipped. Python +allocations traced during the measured minute peaked below 9 KB; this excludes +pre-existing allocations and native/service memory. No hands were visible in this +run either. Results are in `.tools/latest-soak-b2.json`. + +The built b2 wheel passed **107 tests on each of CPython 3.10.11, 3.11.9, 3.12.10, +3.13.17, 3.14.10, and 3.15.2** in separate Windows venvs. Each installed wheel received +30 live sequential frames and 30 live latest-mode frames, closed its worker, passed +dependency checks, and round-tripped the previously recorded full-hand session. +Results are in `.tools/compatibility-results-0.1.0b2.json`. + +Automated tests exercise frame replacement, duplicate freshness, expiration, +connection failures, EOF, source thread ownership, startup cleanup, shutdown, +reconnection, invalid options, bounded health history, and CLI recording/JSON output. +The b1 hardware and platform limits in [beta validation](BETA_VALIDATION.md) still +apply. Linux/macOS controller tests and clean-Windows driver setup remain pending. diff --git a/examples/latest_frames.py b/examples/latest_frames.py new file mode 100644 index 0000000..b3df60e --- /dev/null +++ b/examples/latest_frames.py @@ -0,0 +1,31 @@ +"""Read newest hand snapshots in a deliberately slower application loop.""" + +import time + +from source_robotics_leap_motion import ConnectionError, Tracker, TrackingTimeoutError + + +def main() -> None: + # Use Tracker.demo(mode="latest") to try this without hardware. + with Tracker(mode="latest", max_frame_age=0.1) as tracker: + while True: + try: + frame = tracker.read(timeout=0.25) + except TrackingTimeoutError: # Includes StaleFrameError. + print("Waiting for fresh tracking:", tracker.health.to_dict()) + continue + except ConnectionError: + print("Tracking disconnected:", tracker.health.to_dict()) + break + if hand := frame.right_hand: + print(hand.palm.position) + else: + print("Camera streaming; no right hand visible") + time.sleep(0.05) # Other work: latest mode drains tracking independently. + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + pass diff --git a/pyproject.toml b/pyproject.toml index 88f484c..78d8793 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "source-robotics-leap-motion" -version = "0.1.0b1" +version = "0.1.0b2" description = "Modern Leap Motion tracking and configurable robot-arm teleoperation primitives" readme = "README.md" requires-python = ">=3.10" diff --git a/src/source_robotics_leap_motion/__init__.py b/src/source_robotics_leap_motion/__init__.py index 88d997d..3895799 100644 --- a/src/source_robotics_leap_motion/__init__.py +++ b/src/source_robotics_leap_motion/__init__.py @@ -5,14 +5,16 @@ LeapError, RecordingError, RuntimeNotFoundError, + StaleFrameError, TrackingTimeoutError, ) +from .health import TrackingHealth from .mapping import Mapping from .models import Bone, Finger, Frame, Hand, Palm, Quaternion, Vector3 from .recording import Recorder, read_recording from .tracker import Tracker -__version__ = "0.1.0b1" +__version__ = "0.1.0b2" __all__ = [ "Bone", @@ -27,7 +29,9 @@ "Recorder", "RecordingError", "RuntimeNotFoundError", + "StaleFrameError", "Tracker", + "TrackingHealth", "TrackingTimeoutError", "Vector3", "read_recording", diff --git a/src/source_robotics_leap_motion/_latest.py b/src/source_robotics_leap_motion/_latest.py new file mode 100644 index 0000000..fcd59c5 --- /dev/null +++ b/src/source_robotics_leap_motion/_latest.py @@ -0,0 +1,149 @@ +"""Single-slot acquisition. All source methods run on the same owning thread.""" + +from __future__ import annotations + +import threading +import time +from typing import TYPE_CHECKING + +from .errors import ConnectionError, StaleFrameError, TrackingTimeoutError +from .health import _HealthMonitor +from .models import Frame + +if TYPE_CHECKING: + from .tracker import FrameSource + + +class _LatestReader: + _poll_interval = 0.05 + _shutdown_timeout = 5.0 + + def __init__(self, source: FrameSource, health: _HealthMonitor, max_age: float) -> None: + self._source = source + self._health = health + self._max_age = max_age + self._condition = threading.Condition() + self._stop = threading.Event() + self._ready = threading.Event() + self._opened = False + self._ended = False + self._error: BaseException | None = None + self._close_error: BaseException | None = None + self._pending: tuple[Frame, float] | None = None + self._thread = threading.Thread(target=self._run, name="source-leap-reader", daemon=True) + + @property + def alive(self) -> bool: + return self._thread.is_alive() + + def start(self) -> None: + self._thread.start() + # Opening can include a first-run runtime download, so it has no artificial + # frame deadline. Short waits allow KeyboardInterrupt during startup. + while not self._ready.wait(self._poll_interval): + pass + if not self._opened and self._error is not None: + raise self._error.with_traceback(None) + + def _discard(self) -> None: + if self._pending is not None: + self._pending = None + self._health.skip() + + def _run(self) -> None: + try: + self._source.open() + self._opened = True + self._health.set_state("waiting") + self._ready.set() + while not self._stop.is_set(): + before = time.perf_counter() + frame = self._source.read(self._poll_interval) + received = time.perf_counter() + if frame is None: + # A custom source may return None immediately. Avoid a busy loop. + self._stop.wait(max(0.0, self._poll_interval - (received - before))) + continue + with self._condition: + if self._stop.is_set(): + break + if not self._health.receive(frame, received): + self._health.skip() + else: + self._discard() + self._pending = (frame, received) + self._condition.notify_all() + except StopIteration: + with self._condition: + self._ended = True + self._health.set_state("ended") + self._condition.notify_all() + except BaseException as error: + self._fail(error) + finally: + try: + self._source.close() + except BaseException as error: + self._close_error = error + self._fail(error) + self._ready.set() + with self._condition: + self._condition.notify_all() + + def _fail(self, error: BaseException) -> None: + with self._condition: + self._discard() + if self._error is None: + self._error = error + self._health.fail(error) + self._condition.notify_all() + + def read(self, timeout: float) -> Frame: + deadline = time.perf_counter() + timeout + with self._condition: + while True: + if self._stop.is_set(): + raise RuntimeError("Tracker is closed") + if self._error is not None: + raise self._error.with_traceback(None) + now = time.perf_counter() + if self._pending is not None: + frame, received = self._pending + if now - received <= self._max_age: + self._pending = None + self._health.deliver() + return frame + self._discard() + if self._ended: + raise StopIteration + remaining = deadline - now + if remaining <= 0: + self._health.timeout() + health = self._health.snapshot() + if health.last_frame_age_s is not None and ( + health.last_frame_age_s > self._max_age + ): + raise StaleFrameError( + f"No fresh tracking frame within {timeout:g}s; last distinct frame " + f"arrived {health.last_frame_age_s:.3f}s ago " + f"(limit {self._max_age:g}s)." + ) + raise TrackingTimeoutError(f"No new tracking frame within {timeout:g}s.") + self._condition.wait(remaining) + + def close(self) -> None: + self._stop.set() + with self._condition: + self._discard() + self._condition.notify_all() + if self._thread.ident is not None: + self._thread.join(self._shutdown_timeout) + if self.alive: + error = ConnectionError( + "Tracking source did not stop within the shutdown deadline. Custom sources " + "must honor read(timeout); reconnect is unavailable until it stops." + ) + self._health.fail(error) + raise error + if self._close_error is not None: + raise self._close_error.with_traceback(None) diff --git a/src/source_robotics_leap_motion/cli.py b/src/source_robotics_leap_motion/cli.py index 2774e9b..f6c0f76 100644 --- a/src/source_robotics_leap_motion/cli.py +++ b/src/source_robotics_leap_motion/cli.py @@ -40,6 +40,20 @@ def _tracking_options(parser: argparse.ArgumentParser) -> None: parser.add_argument("--frames", type=_positive_int, help="stop after this many frames") parser.add_argument("--duration", type=_positive_float, help="stop after this many seconds") parser.add_argument("--json", action="store_true", help="write every frame as JSON Lines") + parser.add_argument( + "--latest", action="store_true", help="keep only the newest unread frame; may skip frames" + ) + parser.add_argument( + "--max-frame-age", + type=_positive_float, + default=0.1, + help="latest-mode local receipt age limit in seconds (default: 0.1)", + ) + parser.add_argument( + "--health", + action="store_true", + help="write health JSON to stderr while streaming and on exit", + ) runtime = parser.add_mutually_exclusive_group() runtime.add_argument("--library", help="explicit LeapC library path for development") runtime.add_argument("--runtime-dir", help="use a prepared local legacy runtime if needed") @@ -90,6 +104,7 @@ def _parser() -> argparse.ArgumentParser: replay.add_argument("--frames", type=_positive_int) replay.add_argument("--duration", type=_positive_float) replay.add_argument("--json", action="store_true") + replay.add_argument("--health", action="store_true") replay.add_argument("--timeout", type=_positive_float, default=5) runtime = commands.add_parser("runtime", help="prepare or inspect a local runtime prototype") actions = runtime.add_subparsers(dest="runtime_action", required=True) @@ -247,6 +262,8 @@ def _runtime(args) -> int: def _run(args) -> int: + mode = "latest" if getattr(args, "latest", False) else "sequential" + max_age = getattr(args, "max_frame_age", 0.1) if args.command == "replay": tracker = Tracker( backend=ReplayBackend(args.path, realtime=args.realtime, speed=args.speed), @@ -255,13 +272,18 @@ def _run(args) -> int: elif args.demo: if args.library or args.runtime_dir: raise ValueError("--demo cannot be combined with a runtime or library selection") - tracker = Tracker.demo() + tracker = Tracker.demo(mode=mode, max_frame_age=max_age, timeout=args.timeout) else: tracker = Tracker( - library_path=args.library, runtime_dir=args.runtime_dir, timeout=args.timeout + library_path=args.library, + runtime_dir=args.runtime_dir, + timeout=args.timeout, + mode=mode, + max_frame_age=max_age, ) count = 0 last_print = -math.inf + last_health = time.perf_counter() started = time.perf_counter() recorder = Recorder(args.path, overwrite=args.overwrite) if args.command == "record" else None # Open tracking first so a missing runtime does not create an empty recording. @@ -276,45 +298,59 @@ def _run(args) -> int: else "live" ) print(f"Source: {source}. Positions in metres. Ctrl+C to stop.", file=sys.stderr) - while True: - elapsed = time.perf_counter() - started - if args.duration is not None and elapsed >= args.duration: - break - timeout = args.timeout - if args.duration is not None: - timeout = min(timeout, args.duration - elapsed) - try: - frame = tracker.read(timeout=timeout) - except StopIteration: - break - except TrackingTimeoutError: - if args.duration is not None and time.perf_counter() - started >= args.duration: + last_health = started + try: + while True: + elapsed = time.perf_counter() - started + if args.duration is not None and elapsed >= args.duration: + break + timeout = args.timeout + if args.duration is not None: + timeout = min(timeout, args.duration - elapsed) + try: + frame = tracker.read(timeout=timeout) + except StopIteration: break - raise - if recorder is not None: - recorder.write(frame) - count += 1 - now = time.perf_counter() - if args.json: - print(json.dumps(frame_to_dict(frame), allow_nan=False, separators=(",", ":"))) - elif now - last_print >= 0.1: - description = ( - " | ".join( - f"{hand.handedness} #{hand.id}: " - f"({hand.palm.position.x:+.3f}, {hand.palm.position.y:+.3f}, " - f"{hand.palm.position.z:+.3f}) grab={hand.grab_strength:.2f}" - for hand in frame.hands + except TrackingTimeoutError: + if args.duration is not None and time.perf_counter() - started >= args.duration: + break + raise + if recorder is not None: + recorder.write(frame) + count += 1 + now = time.perf_counter() + if args.json: + print(json.dumps(frame_to_dict(frame), allow_nan=False, separators=(",", ":"))) + elif now - last_print >= 0.1: + description = ( + " | ".join( + f"{hand.handedness} #{hand.id}: " + f"({hand.palm.position.x:+.3f}, {hand.palm.position.y:+.3f}, " + f"{hand.palm.position.z:+.3f}) grab={hand.grab_strength:.2f}" + for hand in frame.hands + ) + or "no hands" ) - or "no hands" - ) - print(f"Frame {frame.id}: {description}", flush=True) - last_print = now - if args.frames is not None and count >= args.frames: - break + print(f"Frame {frame.id}: {description}", flush=True) + last_print = now + if args.health and now - last_health >= 1: + _print_health(tracker) + last_health = now + if args.frames is not None and count >= args.frames: + break + finally: + if args.health: + _print_health(tracker) print(f"Processed {count} frame(s).", file=sys.stderr) return 0 +def _print_health(tracker: Tracker) -> None: + print( + json.dumps({"tracking_health": tracker.health.to_dict()}, allow_nan=False), file=sys.stderr + ) + + def main(argv: list[str] | None = None) -> int: arguments = list(sys.argv[1:] if argv is None else argv) commands = { diff --git a/src/source_robotics_leap_motion/errors.py b/src/source_robotics_leap_motion/errors.py index 5a373b1..dfbbbff 100644 --- a/src/source_robotics_leap_motion/errors.py +++ b/src/source_robotics_leap_motion/errors.py @@ -17,5 +17,9 @@ class TrackingTimeoutError(LeapError): """No tracking frame arrived within the requested deadline.""" +class StaleFrameError(TrackingTimeoutError): + """Latest mode has received frames, but no unread frame is fresh enough.""" + + class RecordingError(LeapError): """A recording is invalid, unsupported, or truncated.""" diff --git a/src/source_robotics_leap_motion/health.py b/src/source_robotics_leap_motion/health.py new file mode 100644 index 0000000..8052df8 --- /dev/null +++ b/src/source_robotics_leap_motion/health.py @@ -0,0 +1,134 @@ +"""Bounded, thread-safe tracking statistics; ages use the local monotonic clock.""" + +from __future__ import annotations + +import threading +import time +from collections import deque +from dataclasses import asdict, dataclass +from typing import Literal + +from .errors import ConnectionError +from .models import Frame + +TrackingMode = Literal["sequential", "latest"] +TrackingStatus = Literal[ + "closed", "starting", "waiting", "streaming", "stale", "disconnected", "ended", "error" +] + + +@dataclass(frozen=True, slots=True) +class TrackingHealth: + """An immutable snapshot of one tracking session. + + ``last_frame_age_s`` is time since Python received the last distinct frame, + not camera-to-application latency. ``received_fps`` measures distinct local + arrivals over at most 120 frames. No hands still counts as streaming. + ``skipped_frames`` counts frames deliberately omitted by latest mode, not + frames lost upstream. Counters restart on open/reconnect. + """ + + mode: TrackingMode + status: TrackingStatus + received_frames: int = 0 + delivered_frames: int = 0 + skipped_frames: int = 0 + duplicate_frames: int = 0 + read_timeouts: int = 0 + timestamp_resets: int = 0 + received_fps: float = 0.0 + last_frame_age_s: float | None = None + last_frame_id: int | None = None + last_frame_timestamp_us: int | None = None + device_id: int | None = None + clock_domain: str | None = None + last_hand_count: int | None = None + last_error: str | None = None + + def to_dict(self) -> dict: + return asdict(self) + + +class _HealthMonitor: + def __init__(self, mode: TrackingMode, max_age: float) -> None: + self._lock = threading.Lock() + self.mode = mode + self.max_age = max_age + self.state: TrackingStatus = "closed" + self.received = self.delivered = self.skipped = self.duplicates = 0 + self.timeouts = self.resets = 0 + self.last_error: str | None = None + self._key: tuple | None = None + self._last: float | None = None + self._arrivals: deque[float] = deque(maxlen=120) + self._hand_count: int | None = None + + def set_state(self, state: TrackingStatus) -> None: + with self._lock: + self.state = state + + def receive(self, frame: Frame, now: float) -> bool: + key = (frame.clock_domain, frame.device_id, frame.id, frame.timestamp_us) + with self._lock: + self.received += 1 + if key == self._key: + self.duplicates += 1 + return False + if self._key is not None and key[:2] == self._key[:2] and key[3] < self._key[3]: + self.resets += 1 + self._key = key + self._last = now + self._hand_count = len(frame.hands) + self._arrivals.append(now) + self.state = "streaming" + return True + + def deliver(self) -> None: + with self._lock: + self.delivered += 1 + + def skip(self) -> None: + with self._lock: + self.skipped += 1 + + def timeout(self) -> None: + with self._lock: + self.timeouts += 1 + + def fail(self, error: BaseException) -> None: + with self._lock: + self.last_error = f"{type(error).__name__}: {error}" + self.state = ( + "disconnected" if isinstance(error, (ConnectionError, OSError)) else "error" + ) + + def snapshot(self) -> TrackingHealth: + with self._lock: + now = time.perf_counter() + age = None if self._last is None else max(0.0, now - self._last) + status = self.state + if status == "streaming" and age is not None and age > self.max_age: + status = "stale" + fps = 0.0 + if status == "streaming" and len(self._arrivals) > 1: + span = self._arrivals[-1] - self._arrivals[0] + if span > 0: + fps = (len(self._arrivals) - 1) / span + return TrackingHealth( + mode=self.mode, + status=status, + received_frames=self.received, + delivered_frames=self.delivered, + skipped_frames=self.skipped, + duplicate_frames=self.duplicates, + read_timeouts=self.timeouts, + timestamp_resets=self.resets, + received_fps=fps, + last_frame_age_s=age, + last_frame_id=None if self._key is None else self._key[2], + last_frame_timestamp_us=None if self._key is None else self._key[3], + device_id=None if self._key is None else self._key[1], + clock_domain=None if self._key is None else self._key[0], + last_hand_count=self._hand_count, + last_error=self.last_error, + ) diff --git a/src/source_robotics_leap_motion/tracker.py b/src/source_robotics_leap_motion/tracker.py index b4b3cdd..76e0cdb 100644 --- a/src/source_robotics_leap_motion/tracker.py +++ b/src/source_robotics_leap_motion/tracker.py @@ -4,15 +4,20 @@ import math import os +import time from collections.abc import Iterator from typing import Protocol from ._compat import Self +from ._latest import _LatestReader from .errors import TrackingTimeoutError +from .health import TrackingHealth, TrackingMode, _HealthMonitor from .models import Frame class FrameSource(Protocol): + """A source must honor read deadlines. Latest mode owns it on one thread.""" + def open(self) -> None: ... def read(self, timeout: float) -> Frame | None: ... def close(self) -> None: ... @@ -32,26 +37,53 @@ def __init__( library_path: str | None = None, runtime_dir: str | None = None, timeout: float = 5.0, + mode: TrackingMode = "sequential", + max_frame_age: float = 0.1, ) -> None: if not math.isfinite(timeout) or timeout <= 0: raise ValueError("Timeout must be positive and finite") if sum(option is not None for option in (backend, library_path, runtime_dir)) > 1: raise ValueError("Choose a backend, a native library, or a local runtime directory") + if mode not in ("sequential", "latest"): + raise ValueError("Mode must be 'sequential' or 'latest'") + if not math.isfinite(max_frame_age) or max_frame_age <= 0: + raise ValueError("Maximum frame age must be positive and finite") self._backend = backend self._library_path = library_path self._runtime_dir = runtime_dir self.timeout = timeout + self._mode = mode + self._max_frame_age = max_frame_age + self._health = _HealthMonitor(mode, max_frame_age) + self._reader: _LatestReader | None = None self._opened = False @classmethod - def demo(cls, *, rate: float = 60.0, realtime: bool = True) -> Self: + def demo( + cls, + *, + rate: float = 60.0, + realtime: bool = True, + mode: TrackingMode = "sequential", + max_frame_age: float = 0.1, + timeout: float = 5.0, + ) -> Self: from .backends.demo import DemoBackend - return cls(backend=DemoBackend(rate, realtime=realtime)) + if mode == "latest" and not realtime: + raise ValueError("Latest demo mode requires realtime pacing") + return cls( + backend=DemoBackend(rate, realtime=realtime), + mode=mode, + max_frame_age=max_frame_age, + timeout=timeout, + ) def __enter__(self) -> Self: if self._opened: raise RuntimeError("Tracker is already open") + if self._reader is not None and self._reader.alive: + raise RuntimeError("Previous tracking reader is still stopping") if self._backend is None: if self._runtime_dir is not None: from .backends.shared import SharedRuntimeBackend @@ -67,14 +99,31 @@ def __enter__(self) -> Self: from .backends.auto import AutoBackend self._backend = AutoBackend(connect_timeout=self.timeout) - try: - self._backend.open() - except BaseException: - self._backend.close() - raise + self._health = _HealthMonitor(self._mode, self._max_frame_age) + self._health.set_state("starting") + if self._mode == "latest": + self._reader = _LatestReader(self._backend, self._health, self._max_frame_age) + try: + self._reader.start() + except BaseException: + self._reader.close() + raise + else: + try: + self._backend.open() + except BaseException as error: + self._health.fail(error) + self._backend.close() + raise + self._health.set_state("waiting") self._opened = True return self + @property + def health(self) -> TrackingHealth: + """Current session statistics. Reading health never accesses the device.""" + return self._health.snapshot() + @property def backend_name(self) -> str: return getattr(self._backend, "name", "custom") @@ -89,12 +138,27 @@ def read(self, timeout: float | None = None) -> Frame: deadline = self.timeout if timeout is None else timeout if not math.isfinite(deadline) or deadline <= 0: raise ValueError("Timeout must be positive and finite") - frame = self._backend.read(deadline) + reader = self._reader + if self._mode == "latest": + if reader is None: + raise RuntimeError("Tracker is closed") + return reader.read(deadline) + try: + frame = self._backend.read(deadline) + except StopIteration: + self._health.set_state("ended") + raise + except BaseException as error: + self._health.fail(error) + raise if frame is None: + self._health.timeout() raise TrackingTimeoutError( f"No tracking frames within {deadline:g}s. Check camera USB connection " "and tracking service. An empty-hand frame would still be returned." ) + self._health.receive(frame, time.perf_counter()) + self._health.deliver() return frame def __iter__(self) -> Iterator[Frame]: @@ -106,8 +170,15 @@ def __iter__(self) -> Iterator[Frame]: def close(self) -> None: try: - if self._backend is not None: + if self._reader is not None: + self._reader.close() + self._reader = None + elif self._opened and self._backend is not None: self._backend.close() + self._health.set_state("closed") + except BaseException as error: + self._health.fail(error) + raise finally: self._opened = False diff --git a/tests/test_cli_runtime.py b/tests/test_cli_runtime.py index 9d032df..19d3528 100644 --- a/tests/test_cli_runtime.py +++ b/tests/test_cli_runtime.py @@ -20,6 +20,27 @@ def test_short_duration_exits_normally(capsys): assert main(["--demo", "--duration", "0.02"]) == 0 +def test_latest_cli_keeps_health_separate_from_frame_json(capsys): + assert main(["--demo", "--latest", "--health", "--frames", "3", "--json"]) == 0 + captured = capsys.readouterr() + frames = [json.loads(line) for line in captured.out.splitlines()] + assert len(frames) == 3 + health = json.loads(captured.err.splitlines()[0])["tracking_health"] + assert health["mode"] == "latest" + assert health["delivered_frames"] == 3 + assert health["last_hand_count"] == 1 + + +def test_latest_record_and_replay_health(tmp_path, capsys): + path = tmp_path / "latest.jsonl" + assert main(["record", str(path), "--demo", "--latest", "--frames", "3"]) == 0 + capsys.readouterr() + assert main(["replay", str(path), "--json", "--health"]) == 0 + captured = capsys.readouterr() + assert len(captured.out.splitlines()) == 3 + assert json.loads(captured.err.splitlines()[0])["tracking_health"]["status"] == "ended" + + def test_cli_record_and_replay(tmp_path, capsys): path = tmp_path / "demo.jsonl" assert main(["record", str(path), "--demo", "--frames", "2"]) == 0 diff --git a/tests/test_latest.py b/tests/test_latest.py new file mode 100644 index 0000000..d5dce24 --- /dev/null +++ b/tests/test_latest.py @@ -0,0 +1,283 @@ +import queue +import threading +import time +from dataclasses import FrozenInstanceError + +import pytest + +from source_robotics_leap_motion import ( + ConnectionError, + Frame, + StaleFrameError, + Tracker, + TrackingTimeoutError, +) +from source_robotics_leap_motion._latest import _LatestReader +from source_robotics_leap_motion.health import _HealthMonitor + + +class Source: + def __init__(self): + self.queue = queue.Queue() + self.calls = [] + self.closed = threading.Event() + + def open(self): + self.calls.append(("open", threading.get_ident())) + self.closed.clear() + + def read(self, timeout): + self.calls.append(("read", threading.get_ident())) + try: + item = self.queue.get(timeout=timeout) + except queue.Empty: + return None + if isinstance(item, BaseException): + raise item + return item + + def close(self): + self.calls.append(("close", threading.get_ident())) + self.closed.set() + + +def wait_until(predicate, timeout=3): + deadline = time.perf_counter() + timeout + while not predicate(): + if time.perf_counter() >= deadline: + pytest.fail("Reader did not reach the expected state") + time.sleep(0.001) + + +def test_latest_has_one_slot_and_never_repeats_a_consumed_frame(): + source = Source() + with Tracker(backend=source, mode="latest", max_frame_age=5) as tracker: + for index in range(100): + source.queue.put(Frame(index, index)) + wait_until(lambda: tracker.health.received_frames == 100) + assert tracker.read().id == 99 + health = tracker.health + assert health.status == "streaming" + assert health.last_hand_count == 0 # No hands is not a connection fault. + assert (health.received_frames, health.delivered_frames, health.skipped_frames) == ( + 100, + 1, + 99, + ) + with pytest.raises(TrackingTimeoutError): + tracker.read(timeout=0.01) + assert tracker.health.read_timeouts == 1 + with pytest.raises(FrozenInstanceError): + health.status = "closed" + assert tracker.health.status == "closed" + assert source.closed.is_set() + owners = {owner for _name, owner in source.calls} + assert len(owners) == 1 and threading.get_ident() not in owners + assert [name for name, _owner in source.calls].count("close") == 1 + tracker.close() # Idempotent, without closing the source twice. + assert [name for name, _owner in source.calls].count("close") == 1 + + +def test_stale_pending_frame_is_discarded_and_duplicate_does_not_refresh_it(): + source = Source() + with Tracker(backend=source, mode="latest", max_frame_age=0.02) as tracker: + frame = Frame(1, 1) + source.queue.put(frame) + wait_until(lambda: tracker.health.received_frames == 1) + wait_until(lambda: tracker.health.last_frame_age_s > 0.04) + source.queue.put(frame) + wait_until(lambda: tracker.health.duplicate_frames == 1) + assert tracker.health.status == "stale" + with pytest.raises(StaleFrameError): + tracker.read(timeout=0.01) + assert tracker.health.skipped_frames == 2 + source.queue.put(Frame(2, 2)) + assert tracker.read().id == 2 + assert tracker.health.status == "streaming" + + +@pytest.mark.parametrize("error", [ConnectionError("USB lost"), ValueError("Invalid frame")]) +def test_source_failure_invalidates_pending_data_and_preserves_error_type(error): + source = Source() + with Tracker(backend=source, mode="latest", max_frame_age=5) as tracker: + source.queue.put(Frame(1, 1)) + source.queue.put(error) + wait_until(lambda: source.closed.is_set()) + for _ in range(3): + with pytest.raises(type(error), match=str(error)): + tracker.read() + health = tracker.health + assert health.delivered_frames == 0 and health.skipped_frames == 1 + assert health.status == ("disconnected" if isinstance(error, ConnectionError) else "error") + assert str(error) in health.last_error + + +def test_latest_eof_delivers_final_unread_frame_once(): + source = Source() + source.queue.put(Frame(1, 1)) + source.queue.put(Frame(2, 2)) + source.queue.put(StopIteration()) + with Tracker(backend=source, mode="latest", max_frame_age=5) as tracker: + wait_until(lambda: source.closed.is_set()) + assert tracker.read().id == 2 + with pytest.raises(StopIteration): + tracker.read() + assert tracker.health.status == "ended" + + +def test_latest_open_failure_closes_on_owner_thread_without_leaking_worker(): + class Failed(Source): + def open(self): + super().open() + raise OSError("Cannot open") + + source = Failed() + tracker = Tracker(backend=source, mode="latest") + with pytest.raises(OSError, match="Cannot open"): + with tracker: + pass + assert source.closed.is_set() + assert len({owner for _name, owner in source.calls}) == 1 + assert not tracker._reader.alive + tracker.close() + + +def test_close_wakes_waiting_consumer_and_reconnect_starts_a_new_session(): + source = Source() + tracker = Tracker(backend=source, mode="latest") + errors = queue.Queue() + waiting = threading.Event() + + def consume(): + waiting.set() + try: + tracker.read(timeout=30) + except BaseException as error: + errors.put(error) + + with tracker: + consumer = threading.Thread(target=consume) + consumer.start() + assert waiting.wait(3) + tracker.close() + consumer.join(3) + assert not consumer.is_alive() + assert isinstance(errors.get_nowait(), RuntimeError) + tracker.reconnect() + assert tracker.health.received_frames == 0 + source.queue.put(Frame(10, 20)) + assert tracker.read().id == 10 + assert not any(thread.name == "source-leap-reader" for thread in threading.enumerate()) + + +def test_misbehaving_source_shutdown_is_bounded_and_prevents_unsafe_reopen(monkeypatch): + release = threading.Event() + reading = threading.Event() + + class Blocking(Source): + def read(self, timeout): + reading.set() + release.wait(5) + return Frame(1, 1) + + monkeypatch.setattr(_LatestReader, "_shutdown_timeout", 0.02) + source = Blocking() + tracker = Tracker(backend=source, mode="latest") + tracker.__enter__() + assert reading.wait(3) + try: + with pytest.raises(ConnectionError, match="shutdown deadline"): + tracker.close() + assert not source.closed.is_set() # Never close a source concurrently with its read. + with pytest.raises(RuntimeError, match="still stopping"): + tracker.__enter__() + finally: + release.set() + assert source.closed.wait(3) + tracker.close() + assert tracker.health.received_frames == 0 + + +def test_read_during_final_close_never_polls_source_from_a_different_thread(monkeypatch): + source = Source() + tracker = Tracker(backend=source, mode="latest") + tracker.__enter__() + finishing = threading.Event() + finish = threading.Event() + original_set_state = tracker._health.set_state + + def set_state(state): + if state == "closed": + finishing.set() + assert finish.wait(3) + original_set_state(state) + + monkeypatch.setattr(tracker._health, "set_state", set_state) + closer = threading.Thread(target=tracker.close) + closer.start() + try: + assert finishing.wait(3) + with pytest.raises(RuntimeError, match="closed"): + tracker.read(timeout=0.01) + finally: + finish.set() + closer.join(3) + assert not closer.is_alive() + assert len({owner for _name, owner in source.calls}) == 1 + + +def test_health_uses_local_arrivals_reports_clock_reset_and_bounds_history(monkeypatch): + health = _HealthMonitor("latest", 0.1) + health.set_state("waiting") + assert health.snapshot().last_frame_age_s is None + for index in range(150): + health.receive(Frame(index, 1_000_000 + index), index * 0.01) + monkeypatch.setattr("source_robotics_leap_motion.health.time.perf_counter", lambda: 1.5) + snapshot = health.snapshot() + assert snapshot.received_fps == pytest.approx(100) + assert snapshot.last_frame_age_s == pytest.approx(0.01) + assert snapshot.last_hand_count == 0 + assert len(health._arrivals) == 120 + health.receive(Frame(151, 10), 1.5) + assert health.snapshot().timestamp_resets == 1 + monkeypatch.setattr("source_robotics_leap_motion.health.time.perf_counter", lambda: 2) + assert health.snapshot().status == "stale" + assert health.snapshot().received_fps == 0 + + +def test_sequential_health_preserves_every_frame_and_resets_on_reopen(): + source = Source() + tracker = Tracker(backend=source) + with tracker: + frame = Frame(1, 1) + for _ in range(2): + source.queue.put(frame) + assert tracker.read() is frame + assert tracker.health.received_frames == tracker.health.delivered_frames == 2 + assert tracker.health.duplicate_frames == 1 + assert tracker.health.skipped_frames == 0 + with tracker: + assert tracker.health.received_frames == 0 + with pytest.raises(TrackingTimeoutError): + tracker.read(timeout=0.001) + assert tracker.health.status == "waiting" + + +@pytest.mark.parametrize("mode", ["invalid", "", None]) +def test_rejects_unknown_modes(mode): + with pytest.raises(ValueError, match="Mode"): + Tracker(mode=mode) + + +@pytest.mark.parametrize("age", [0, -1, float("inf"), float("nan")]) +def test_rejects_invalid_age_limits(age): + with pytest.raises(ValueError, match="age"): + Tracker(max_frame_age=age) + + +def test_latest_demo_is_paced_and_runs_without_hardware(): + with pytest.raises(ValueError, match="pacing"): + Tracker.demo(mode="latest", realtime=False) + with Tracker.demo(mode="latest") as tracker: + first = tracker.read() + assert tracker.read().timestamp_us > first.timestamp_us diff --git a/tests/test_managed_runtime.py b/tests/test_managed_runtime.py index 2e5d422..4568fa9 100644 --- a/tests/test_managed_runtime.py +++ b/tests/test_managed_runtime.py @@ -6,7 +6,7 @@ import pytest -from source_robotics_leap_motion import Tracker +from source_robotics_leap_motion import Frame, Tracker from source_robotics_leap_motion.backends import _job, managed from source_robotics_leap_motion.cli import main from source_robotics_leap_motion.errors import ConnectionError, RuntimeNotFoundError @@ -113,7 +113,7 @@ def launcher(assets, monkeypatch): prepare_local_runtime(source, destination) monkeypatch.setattr(managed.sys, "platform", "win32") backend = Mock() - backend.read.return_value = object() + backend.read.return_value = Frame(1, 1) backend_factory = Mock(return_value=backend) monkeypatch.setattr(managed, "LegacyWebSocketBackend", backend_factory) busy = Mock(return_value=False) @@ -222,7 +222,7 @@ def test_child_exit_is_reported_instead_of_reading_another_server(launcher): def test_private_startup_requires_a_frame_and_preserves_it_for_the_caller(launcher): - first, second = object(), object() + first, second = Frame(1, 1), Frame(2, 2) launcher.backend.read.side_effect = [first, second] with Tracker(backend=launcher.runtime) as tracker: launcher.backend.read.assert_called_once() diff --git a/tools/benchmark_latest.py b/tools/benchmark_latest.py new file mode 100644 index 0000000..494ab28 --- /dev/null +++ b/tools/benchmark_latest.py @@ -0,0 +1,95 @@ +"""Compare slow consumers of the same service; never sends robot commands. + +Timestamp differences compare two clients of the same clock, not absolute camera +latency. Local frame age is measured immediately after each latest-mode read. +""" + +from __future__ import annotations + +import argparse +import json +import math +import statistics +import time +from pathlib import Path + +from source_robotics_leap_motion import Tracker, __version__ + + +def positive(value: str) -> float: + number = float(value) + if not math.isfinite(number) or number <= 0: + raise argparse.ArgumentTypeError("must be positive and finite") + return number + + +def distribution(values: list[float]) -> dict: + ordered = sorted(values) + return { + "median": statistics.median(ordered), + "p95": ordered[min(len(ordered) - 1, math.ceil(len(ordered) * 0.95) - 1)], + "max": ordered[-1], + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--duration", type=positive, default=10) + parser.add_argument("--consumer-rate", type=positive, default=20) + parser.add_argument("--demo", action="store_true") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + factory = Tracker.demo if args.demo else Tracker + gaps_ms: list[float] = [] + local_ages_ms: list[float] = [] + samples = 0 + hand_frames = 0 + with factory() as sequential, factory(mode="latest") as latest: + started = time.perf_counter() + cpu_started = time.process_time() + while time.perf_counter() - started < args.duration: + old = sequential.read() + new = latest.read() + health = latest.health + if not args.demo and (old.clock_domain, old.device_id) != ( + new.clock_domain, + new.device_id, + ): + raise RuntimeError("Clients are receiving different device clocks") + if not args.demo: + gaps_ms.append((new.timestamp_us - old.timestamp_us) / 1000) + local_ages_ms.append(health.last_frame_age_s * 1000) + hand_frames += bool(new.hands) + samples += 1 + target = started + samples / args.consumer_rate + time.sleep(max(0, min(target - time.perf_counter(), args.duration))) + elapsed = time.perf_counter() - started + cpu_seconds = time.process_time() - cpu_started + result = { + "package": __version__, + "source": "synthetic" if args.demo else "live", + "backend": latest.backend_name, + "elapsed_s": elapsed, + "consumer_fps": samples / elapsed, + "delivered_samples": samples, + "latest_hand_frames": hand_frames, + "latest_local_receipt_age_ms": distribution(local_ages_ms), + "latest_health": latest.health.to_dict(), + "combined_client_cpu_percent_of_one_core": cpu_seconds / elapsed * 100, + } + if gaps_ms: + result["latest_ahead_of_sequential_ms"] = { + **distribution(gaps_ms), + "first": gaps_ms[0], + "last": gaps_ms[-1], + } + result["reader_closed"] = latest.health.status == "closed" + text = json.dumps(result, indent=2, allow_nan=False) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(text + "\n", encoding="utf-8") + print(text) + + +if __name__ == "__main__": + main()