13b95345c7
Spec with pseudocode and test vectors, APK disassembly, analysis tools, shelved reference Python and the macOS BLE probe. Findings are not yet validated against a real tag. The vendor XAPK and nested lib clone are excluded. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EinWd3UT2Xw7UakFtTp6hn
72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""Writes BM200 readings to stdout as JSON lines; pipe that into your logging system.
|
|
|
|
Direct BLE: bm200_logger.py --bm200-addr B0:B1:xx:xx:xx:xx --state-file bm200.state
|
|
From relay: relay-feed | bm200_logger.py --stdin
|
|
(one raw FFF4 notification per line: "<hex>" or "<epoch> <hex>")
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import dataclasses
|
|
import json
|
|
import logging
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from bm2.bm200 import BM200Client, BM200Decoder, HistoryBlock, LiveReading
|
|
|
|
logging.basicConfig(level=logging.INFO, stream=sys.stderr, format="[%(asctime)s] %(message)s")
|
|
|
|
|
|
def emit(reading: Any) -> None:
|
|
row = dataclasses.asdict(reading)
|
|
row["type"] = "live" if isinstance(reading, LiveReading) else "history"
|
|
row["time"] = reading.time.isoformat()
|
|
print(json.dumps(row), flush=True)
|
|
|
|
|
|
def from_stdin() -> None:
|
|
decoder = BM200Decoder()
|
|
for line in sys.stdin:
|
|
parts = line.split()
|
|
if not parts:
|
|
continue
|
|
try:
|
|
events = decoder.feed(bytes.fromhex(parts[-1]), float(parts[0]) if len(parts) > 1 else None)
|
|
except ValueError as e: # relay input: skip it, keep logging
|
|
logging.warning("bad line %r: %s", line, e)
|
|
continue
|
|
for event in events:
|
|
if isinstance(event, LiveReading):
|
|
emit(event)
|
|
elif isinstance(event, HistoryBlock):
|
|
for record in event.records:
|
|
emit(record)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--bm200-addr", metavar="ADDR")
|
|
parser.add_argument("--state-file", type=Path, help="remembers the newest history record between runs")
|
|
parser.add_argument("--stdin", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
if args.stdin:
|
|
return from_stdin()
|
|
if not args.bm200_addr:
|
|
parser.error("--bm200-addr or --stdin required")
|
|
|
|
since = None
|
|
if args.state_file and args.state_file.exists():
|
|
since = datetime.fromisoformat(args.state_file.read_text().strip())
|
|
|
|
def save(newest: datetime) -> None:
|
|
if args.state_file:
|
|
args.state_file.write_text(newest.isoformat())
|
|
|
|
asyncio.run(BM200Client(args.bm200_addr, since).run(emit, save))
|
|
|
|
|
|
main()
|