Merge pull request #1 from floesche/optimize

Add examples

Author: peterpolidoro

.gitattributes

@@ -0,0 +1,2 @@
+# SCM syntax highlighting & preventing 3-way merges
+pixi.lock merge=binary linguist-language=YAML linguist-generated=true -diff

examples/all_on.py

@@ -0,0 +1,16 @@
+import time
+
+from arena_interface import ArenaInterface
+
+ai = ArenaInterface(debug=True)
+ai.set_ethernet_mode(ip_address="10.103.40.45")
+
+start_time = time.time()
+ai.all_on()
+end_time = time.time()
+duration = end_time - start_time
+
+time.sleep(5)
+ai.all_off()
+
+print(f"Duration of all_on() call: {duration:.6f} seconds")

examples/mode2.py

@@ -0,0 +1,27 @@
+"""Play 5 patterns for 5 seconds each using play_pattern.
+
+Pattern 1 runs at 300 fps; the rest at random rates between 10 and 300 fps.
+The 4 additional patterns are chosen randomly from IDs 2–700.
+"""
+
+import os
+import random
+
+from arena_interface import ArenaInterface
+
+PATTERN_IDS = [1] + random.sample(range(2, 10), 4)
+RUNTIME_DURATION = 50  # 50 × 100 ms = 5 s
+
+ip = os.environ.get("ARENA_ETH_IP", "10.103.40.45")
+ai = ArenaInterface(debug=True)
+ai.set_ethernet_mode(ip_address=ip)
+
+for pat_id in PATTERN_IDS:
+    fps = 300 if pat_id == PATTERN_IDS[0] else random.randint(10, 300)
+    print(f"\n--- Playing pattern {pat_id} at {fps} fps for 5 s ---")
+    ai.play_pattern(
+        pattern_id=pat_id,
+        frame_rate=fps,
+        runtime_duration=RUNTIME_DURATION,
+    )
+    print(f"  Pattern {pat_id} done.")

examples/mode3.py

@@ -0,0 +1,44 @@
+"""Show 5 patterns for 5 seconds each using show_pattern_frame.
+
+For each pattern, send as many show_pattern_frame commands as possible
+at a target rate of 300 Hz with random frame indices between 1 and 15.
+"""
+
+import os
+import random
+import time
+
+from arena_interface import ArenaInterface
+
+PATTERN_IDS = [1] + random.sample(range(2, 10), 4)
+DURATION_S = 5.0
+TARGET_RATE_HZ = 300
+FRAME_INDEX_MIN = 1
+FRAME_INDEX_MAX = 15
+
+ip = os.environ.get("ARENA_ETH_IP", "10.103.40.45")
+ai = ArenaInterface(debug=True)
+ai.set_ethernet_mode(ip_address=ip)
+
+for pat_id in PATTERN_IDS:
+    print(f"\n--- Showing pattern {pat_id} for {DURATION_S} s at {TARGET_RATE_HZ} Hz ---")
+    interval = 1.0 / TARGET_RATE_HZ
+    count = 0
+    t_start = time.perf_counter()
+    deadline = t_start + DURATION_S
+
+    while True:
+        t_now = time.perf_counter()
+        if t_now >= deadline:
+            break
+        frame_idx = random.randint(FRAME_INDEX_MIN, FRAME_INDEX_MAX)
+        ai.show_pattern_frame(pattern_id=pat_id, frame_index=frame_idx)
+        count += 1
+        # spin-wait until next slot
+        next_time = t_start + count * interval
+        while time.perf_counter() < next_time:
+            pass
+
+    elapsed = time.perf_counter() - t_start
+    actual_hz = count / elapsed if elapsed > 0 else 0
+    print(f"  Pattern {pat_id}: {count} frames in {elapsed:.2f} s ({actual_hz:.1f} Hz)")

examples/play_pat.py

@@ -0,0 +1,14 @@
+import time
+
+from arena_interface import ArenaInterface
+
+ai = ArenaInterface(debug=True)
+ai.set_ethernet_mode(ip_address="10.103.40.45")
+
+# Measure time for play_pattern call
+start_time = time.time()
+ai.play_pattern(pattern_id=532, frame_rate=20, runtime_duration=10)
+end_time = time.time()
+
+duration = end_time - start_time
+print(f"play_pattern() duration: {duration:.6f} seconds")

examples/stream_all_patterns.py

@@ -0,0 +1,29 @@
+"""Stream each pattern file in patterns/ for 5 seconds."""
+
+import os
+from pathlib import Path
+
+from arena_interface import ArenaInterface
+
+PATTERNS_DIR = Path(__file__).resolve().parent.parent / "patterns"
+FRAME_RATE = 0
+RUNTIME_DURATION = 50  # 50 × 100 ms = 5 s
+
+ip = os.environ.get("ARENA_ETH_IP", "10.103.40.45")
+ai = ArenaInterface(debug=True)
+ai.set_ethernet_mode(ip_address=ip)
+
+for pat_file in sorted(PATTERNS_DIR.glob("*.pat")):
+    print(f"\n--- Streaming {pat_file.name} for 5 s at {FRAME_RATE} Hz ---")
+    result = ai.stream_frames(
+        pattern_path=str(pat_file),
+        frame_rate=FRAME_RATE,
+        runtime_duration=RUNTIME_DURATION,
+        analog_out_waveform="constant",
+        analog_update_rate=1.0,
+        analog_frequency=0.0,
+    )
+    print(f"  frames:  {result['frames']}")
+    print(f"  elapsed: {result['elapsed_s']:.2f} s")
+    print(f"  rate:    {result['rate_hz']:.1f} Hz")
+    print(f"  tx:      {result['tx_mbps']:.2f} Mb/s")

pixi.lock

@@ -84,6 +84,9 @@ platforms = ["linux-64", "osx-64", "osx-arm64", "win-64"]
 
 [tool.pixi.dependencies]
 git = "*"
+pyserial = ">=3.5,<4"
+click = ">=8.3.1,<9"
+schedule = ">=1.2.2,<2"
 
 [tool.pixi.pypi-dependencies]
 "arena-interface" = { path = ".", editable = true }
@@ -110,6 +113,8 @@ qtools-install = "python tools/quantum_leaps_tools.py qtools-install"
 qspy = "python tools/quantum_leaps_tools.py qspy"
 bench = "arena-interface bench"
 bench-full = "arena-interface bench --stream-path patterns/pat0004.pat"
+bench-max-rate = "arena-interface bench --stream-path patterns/pat0004.pat --stream-max-rate"
+bench-max-rate-smoke = "arena-interface bench --cmd-iters 250 --spf-seconds 2 --stream-path patterns/pat0004.pat --stream-seconds 2 --stream-max-rate --stream-max-rate-seconds 2"
 bench-smoke = "arena-interface bench --cmd-iters 250 --spf-seconds 2 --stream-path patterns/pat0004.pat --stream-seconds 2"
 bench-persistent = "arena-interface bench --cmd-connect-mode persistent"
 bench-new-connection = "arena-interface bench --cmd-connect-mode new_connection"

pyproject.toml

@@ -9,10 +9,11 @@
     __url__,
     __version__,
 )
-from .arena_interface import ArenaInterface
+from .arena_interface import ArenaInterface, CommandTimeouts
 
 __all__ = [
     "ArenaInterface",
+    "CommandTimeouts",
     "__author__",
     "__copyright__",
     "__description__",