Python¶

# © 2026 Unit Circle Inc.
# SPDX-License-Identifier: Apache-2.0

"""A pytest fixture for hardware tests: `dev`, one connection to the device for the whole run.

    pytest examples/pytest --uclog-device serial:/dev/ttyACM0 --uclog-logdata build/app.elf
    pytest examples/pytest --uclog-device loopback:rate=20 \\
        --uclog-logdata ../spec/vectors/sample.logdata       # the simulated device

Without --uclog-device it uses $UCLOG_DEVICE, else the one local device (as `uclog.connect()`).
A hub is started for the run and stopped at the end.
"""

import os

import pytest

import uclog


def pytest_addoption(parser):
  g = parser.getgroup('uclog')
  g.addoption('--uclog-device', help='device URI or serial number (default: $UCLOG_DEVICE, else the one local device)')
  g.addoption('--uclog-host', help='a hub to use, instead of finding or starting one')
  g.addoption('--uclog-logdata', action='append', help='.logdata or ELF file (repeatable)')


@pytest.fixture(scope='session')
def dev(request):
  o = request.config.option
  device = o.uclog_device or os.environ.get('UCLOG_DEVICE')
  with uclog.connect(o.uclog_host, device, logdata=o.uclog_logdata, skip_backlog=True) as d:
    yield d


@pytest.fixture
def no_errors(dev):
  """Fails the test if the device logs an ERROR (or worse) while it runs, or a log cannot be
  decoded."""
  for _ in dev.logs(timeout=0):  # what came before the test is not its business
    pass
  yield
  bad = [r for r in dev.logs(duration=0.5) if r.error or r.level in ('ERROR', 'FATAL', 'PANIC')]
  assert not bad, '\n'.join(map(str, bad))

# © 2026 Unit Circle Inc.
# SPDX-License-Identifier: Apache-2.0

"""Hardware tests with the `dev` fixture (conftest.py). These hold for any ucLog device; add your
own: commands and their replies, logs a command should produce, limits on values in logs."""

import re

import pytest


def test_device_info(dev):
  info = dev.device_info
  assert info['board']
  assert len(info['app_hash']) == 64


def test_logs_decode(dev, no_errors):
  recs = list(dev.logs(duration=3.0))
  assert recs, 'no logs in 3 s'
  assert all(r.error is None for r in recs), [str(r) for r in recs if r.error]


def test_terminal_answers(dev, no_errors):
  dev.send(61, 'help\r')
  reply = dev.recv(61, timeout=2.0)
  assert reply, 'no reply to help on the terminal'


def test_values_in_logs(dev):
  # Checking what a device reports: every `temp=` value above a limit. Skipped when the device
  # logs no temperatures.
  temps = [int(m[1]) for r in dev.logs(duration=3.0) if (m := re.search(r'temp=(-?\d+)', r.text))]
  if not temps:
    pytest.skip('no temp= logs from this device')
  assert min(temps) > -40, temps

#! /usr/bin/env python3
# © 2026 Unit Circle Inc.
# SPDX-License-Identifier: Apache-2.0

"""Decode a raw capture of what a device sent (no device or hub needed), with its logdata.

    python decode_capture.py -l build/app.elf capture.bin
    cat /dev/ttyACM0 > capture.bin       # a capture: the bytes as they came off the link
    python decode_capture.py -l build/app.elf - < capture.bin

Logs are printed as decoded; terminal text (port 61) is printed as it is; other ports are summed
up at the end. `uclog decode` does the same from the command line.
"""

import argparse
import collections
import sys

import uclog
from uclog import proto

ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument('capture', help="the captured bytes, or - for stdin")
ap.add_argument('-l', '--logdata', action='append', required=True, help='.logdata or ELF file (repeatable)')
a = ap.parse_args()

# Logdata by target digit: a bootloader and an app can log on the same link.
decoders = {ld.target: ld for ld in map(uclog.LogData.load, a.logdata)}
data = sys.stdin.buffer.read() if a.capture == '-' else open(a.capture, 'rb').read()

ports = collections.Counter()
for event in proto.FrameScanner().feed(data):
  if event[0] != 'frame':
    print(f'<frame of {event[1]} bytes dropped: too long>')
    continue
  try:
    payload = proto.cobs_decode(event[1])
  except proto.CobsError:
    print(f'<invalid frame: {event[1][:16].hex(" ")}…>')
    continue
  m = proto.mux_decode(payload)
  if m[0] == 'log':
    _, _, target, addr, args = m
    ld = decoders.get(target)
    if ld is None:
      print(f'<no logdata for target {target}>')
      continue
    try:
      level, file, line, text = ld.decode(addr, args)
      print(f'{level}:{file}:{line}:{text}')
    except uclog.DecodeError as e:
      print(f'<{e.kind}: {e}>')
  elif m[0] == 'port' and m[1] == proto.PORT_TERMINAL:
    sys.stdout.write(m[2].decode('utf-8', 'replace'))
  elif m[0] == 'port':
    ports[m[1]] += 1

for port, n in sorted(ports.items()):
  print(f'port {port}: {n} frames')