Advanced Command Queries

The query command allows you to send arbitrary OSECHI detector commands and receive responses as JSONL. This is the lower-level API for accessing all available device commands beyond the convenience wrappers.

Purpose of Command Queries

The query command provides:

• Direct device access: Send any OSECHI protocol command to the detector
• Structured responses: All responses are returned as JSONL for consistent processing
• Full protocol support: Access to all device commands including proprietary/advanced operations
• Programmatic integration: Python API for building custom workflows

Available Commands

All OSECHI detector commands can be queried:

• Info: GET_HELP, GET_STATUS, GET_VERSION, GET_UPTIME, GET_MAC_ADDRESS, GET_QUEUE_STATS, GET_RTC_TIME
• Configuration: SET_POLL_COUNT, SET_DEADTIME, SET_THRESHOLD, RESET
• Control: TEST_LED

Consult the device specification (GET_HELP) for the complete command list.

---

Basic Query via Python API

Using the Python API for direct command queries:

#!/usr/bin/env python3
"""
query_commands.py - Execute various OSECHI detector commands
"""

from pathlib import Path
from kazunoko import connect, Command, ResponseView

def main():
    """Query detector using various commands."""
    results_path = Path("queries.jsonl")

    with connect() as device:
        cmd = Command(device)

        # Get help (device command reference)
        help_resp = cmd.help()
        view = ResponseView(help_resp)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ HELP")

        # Get version
        version = cmd.version()
        view = ResponseView(version)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ VERSION")

        # Get status
        status = cmd.status()
        view = ResponseView(status)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ STATUS")

        # Get uptime
        uptime = cmd.uptime()
        view = ResponseView(uptime)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ UPTIME")

        # Get MAC address
        mac = cmd.mac_address()
        view = ResponseView(mac)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ MAC_ADDRESS")

        # Get queue statistics
        queue = cmd.queue_stats()
        view = ResponseView(queue)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ QUEUE_STATS")

        # Get RTC time
        rtc = cmd.rtc_time()
        view = ResponseView(rtc)
        with results_path.open("a") as f:
            f.write(view.flat.model_dump_json() + "\n")
        print("✓ RTC_TIME")

        print(f"\nAll responses saved to {results_path}")

if __name__ == "__main__":
    main()

---

Use Case: System Diagnostic Report

Create a comprehensive system diagnostic by querying multiple commands:

#!/usr/bin/env python3
"""
diagnostics.py - Generate system diagnostic report
"""

from pathlib import Path
import json
from kazunoko import connect, Command, ResponseView

def main():
    """Generate system diagnostic report."""
    report_path = Path("diagnostics.jsonl")
    commands = [
        ("version", lambda cmd: cmd.version()),
        ("uptime", lambda cmd: cmd.uptime()),
        ("status", lambda cmd: cmd.status()),
        ("mac_address", lambda cmd: cmd.mac_address()),
        ("queue_stats", lambda cmd: cmd.queue_stats()),
    ]

    with connect() as device:
        cmd = Command(device)

        with report_path.open("w") as f:
            for name, query_fn in commands:
                try:
                    resp = query_fn(cmd)
                    view = ResponseView(resp)
                    # Add command name to flat view for tracking
                    data = view.flat.model_dump()
                    data["_command"] = name
                    f.write(json.dumps(data) + "\n")
                    print(f"✓ {name}")
                except Exception as e:
                    print(f"✗ {name}: {e}")

        print(f"\nDiagnostic report saved to {report_path}")

if __name__ == "__main__":
    main()

Example output (diagnostics.jsonl):

{"type": "response", "status": "ok", "system_version": "1.11.4", "_command": "version"}
{"type": "response", "status": "ok", "system_uptime_ms": 1234567, "_command": "uptime"}
{"type": "response", "status": "ok", "system_version": "1.11.4", "system_adc_channel": 1, "detection_threshold1": 320, "_command": "status"}
{"type": "response", "status": "ok", "mac_address": "00:11:22:33:44:55", "_command": "mac_address"}
{"type": "response", "status": "ok", "queue_size": 42, "queue_max": 1024, "_command": "queue_stats"}

---

Storing Responses as JSONL

All command responses are naturally structured as JSONL, making them perfect for:

• Archival: Store complete query results for reproducibility
• Analysis: Process responses programmatically
• Comparison: Track changes over time by comparing timestamped results
• Integration: Feed JSONL into data pipelines

Recommendation: Always save query responses to JSONL files using ResponseView with flat perspective for consistent, queryable data storage.