Skip to content

API Reference

Auto-generated from docstrings via mkdocstrings. All public names are re-exported at the package top level via the lazy-attr table in chumicro_repl/__init__.py; the per-module sections below mirror the internal layout for readers who want to navigate by source file.

Programmatic raw-REPL session

chumicro_repl.session

Programmatic REPL session — raw REPL over pyserial.

:class:ReplSession is a context manager that opens a serial connection to a CircuitPython or MicroPython board, drives the raw REPL (Ctrl-A entry, Ctrl-D exec, OK<stdout>\x04<stderr>\x04> framing), and exposes three primitives:

  • :meth:~ReplSession.exec — run a block of code, return stdout.
  • :meth:~ReplSession.call — call a named function with literal arguments, return the parsed repr of the result.
  • :meth:~ReplSession.read_until — read raw bytes until a pattern matches, for callers that bypass raw REPL entirely (tailing the friendly REPL, for instance).

The raw-REPL framing is identical between CircuitPython and MicroPython — neither runtime adds a runtime-specific header to either the prompt or the response — so a single code path handles both. The only divergence is what a soft-reboot (Ctrl-D in the friendly REPL) prints: MicroPython emits MPY: soft reboot, CircuitPython is silent. :meth:ReplSession.exec never exits raw REPL, so that divergence never surfaces here; it is a concern for the :func:~chumicro_repl.tail path and the interactive TUI.

ReplSessionError

Bases: Exception

Raised when a raw-REPL operation fails.

Covers four broad classes:

  • Timeout — the board did not respond within the per-op timeout.
  • Protocol error — the board emitted bytes that don't match the raw-REPL framing (typically because it was not in raw REPL; e.g. bootloader mode, a firmware-less chip, wrong baudrate).
  • Execution error — the board executed the code but raised an exception. The exception's stderr is attached as :attr:stderr.
  • Disconnect — the device dropped mid-call. Surfaced as the :class:ReplSessionDisconnected subclass so callers can catch "the cable came out" without conflating it with the other three classes.

ReplSessionDisconnected

Bases: ReplSessionError

Raised when the underlying serial port drops mid-call.

Subclass of :class:ReplSessionError so callers that catch the base class still get the failure, while callers that want to differentiate "device unplugged" from "device returned a traceback" can except ReplSessionDisconnected directly.

The original :class:OSError (typically a serial.SerialException) is attached as :attr:cause for callers that want the underlying errno or message.

ReplSession

Raw-REPL session over pyserial.

Use as a context manager::

with ReplSession(device) as session:
    output = session.exec("print(1 + 2)")
    value = session.call("os.uname")

The constructor opens the serial port, interrupts whatever was running, and enters raw REPL. The __exit__ path sends Ctrl-B (exit raw REPL) and closes the port — the board is left in the friendly REPL, ready for an interactive user or another session.

Parameters:

Name Type Description Default
device DeviceLike | str

Either an object exposing .address (e.g. a chumicro_deploy.Device) or a bare serial-port path string. A bare string is treated as the address; baudrate / time / port-factory come from the other keyword arguments.

required
baudrate int

Only consulted when device is a string. Ignored when device exposes its own baudrate attribute — the device's value wins.

115200
time TimeSource | None

Injectable time source (for tests). Defaults to the stdlib time module.

None
port_factory PortFactory | None

Injectable port factory (for tests). Defaults to :func:chumicro_repl._serial.default_port_factory, which constructs a real pyserial Serial.

None
connect_timeout float

Upper bound in seconds for the initial raw-REPL handshake. Covers the Ctrl-C + Ctrl-A + prompt-read sequence, not the per-exec timeout.

DEFAULT_TIMEOUT
exec(code, *, timeout=DEFAULT_TIMEOUT)

Execute code on the board and return stdout.

Sends code terminated by Ctrl-D, then reads the response until the second \x04 marker. The response format is::

OK<stdout>\x04<stderr>\x04>

Returns stdout as a UTF-8 string. If stderr is non-empty the call raises :class:ReplSessionError with stderr attached — raw REPL never raises an exception object on the host side, only on the board, so surfacing it as a stderr blob is the closest host-side analog.

Parameters:

Name Type Description Default
code str

Python source to execute. May contain newlines.

required
timeout float

Per-exec deadline in seconds.

DEFAULT_TIMEOUT
call(function_name, *args, timeout=DEFAULT_TIMEOUT, **kwargs)

Call function_name on the board with literal arguments.

Builds a one-line print(repr(<function_name>(*args, **kwargs))) and execs it, then parses the stdout via :func:ast.literal_eval. Round-trips anything :func:literal_eval handles — numbers, strings, bytes, tuples, lists, dicts, sets, booleans, None. Anything else raises :class:ReplSessionError because the repr is not a literal.

Parameters:

Name Type Description Default
function_name str

Fully-qualified dotted name. The board must already have the name bound — import whatever the caller needs in a prior :meth:exec.

required
*args object

Positional arguments. Rendered via :func:repr.

()
timeout float

Per-call deadline in seconds.

DEFAULT_TIMEOUT
**kwargs object

Keyword arguments. Rendered via :func:repr.

{}

Returns:

Type Description
object

The parsed return value. None round-trips as

object

None.

read_until(pattern, *, timeout)

Read bytes until pattern matches the accumulated string.

Unlike :meth:exec + :meth:call, this primitive operates on raw bytes from the port — it does not send anything, and it accepts output produced by either the friendly or the raw REPL. Callers that want to stream output from a board that is not under raw-REPL control (for instance, tailing a deploy) use this.

Parameters:

Name Type Description Default
pattern str | Pattern[str]

Regex (compiled or string). The returned text includes the match.

required
timeout float

Deadline in seconds.

required

Returns:

Type Description
str

The accumulated text up to and including the match.

Raises:

Type Description
ReplSessionError

Timeout elapsed before the pattern matched.

tail() and ExitCode

The tail() function and its ExitCode enum are exposed at the package top level (from chumicro_repl import tail, ExitCode). Implementation lives in chumicro_repl._follow — the underscore prefix is just to keep Python's submodule-import machinery from shadowing the top-level tail attribute.

chumicro_repl._follow

Stream REPL output for a bounded window.

:func:tail opens a serial port, reads for seconds of wall-clock time, decodes bytes UTF-8 safely, writes them to output with ANSI highlighting, and returns an :class:ExitCode describing why the window ended. When the device drops mid-stream, tail can hold the window open and reopen the port on replug.

Tail does not touch the REPL mode (no Ctrl-A / Ctrl-B sent). The caller decides whether the board is in raw or friendly REPL before tail runs, and what to do with the returned :class:ExitCode.

ExitCode

Bases: int, Enum

Outcome of a :func:tail invocation.

Int-valued so the CLI can return them directly to the shell. 0 is success; every other value is a distinct failure mode so the caller can differentiate "tail timed out" from "tail saw a traceback" from "the board got unplugged" in scripts.

tail(device, seconds, *, fail_on_traceback=True, output=None, theme=None, baudrate=115200, time=None, port_factory=None, reconnect_seconds=DEFAULT_TAIL_RECONNECT_SECONDS, reconnect_interval=DEFAULT_TAIL_RECONNECT_INTERVAL)

Stream seconds of serial output from device to output.

Runs until the window elapses, the caller interrupts, the device drops past the reconnect budget, or (when fail_on_traceback is True) a noteworthy pattern is detected. Highlighting is inline, so the caller sees tracebacks in red as bytes stream by.

Parameters:

Name Type Description Default
device DeviceLike | str

:class:chumicro_deploy.Device or a serial port path string. Only the address + baudrate are consulted; the deploy-mode fields are ignored.

required
seconds float

Length of the tail window. A :class:float is accepted so sub-second tails are possible in tests.

required
fail_on_traceback bool

When True (default), a matched traceback / safe-mode / hard-fault pattern ends the tail early and the function returns :attr:ExitCode.TRACEBACK_DETECTED. Set False to run the full window regardless.

True
output TextIO | None

Destination for the streamed text. Defaults to sys.stdout. ANSI escapes are emitted whether or not the stream is a TTY — pipe into :func:chumicro_repl.highlight.strip_ansi_sequences downstream for plain text.

None
theme Theme | None

Color theme. Defaults to :data:chumicro_repl.highlight.DEFAULT_THEME.

None
baudrate int

Only consulted when device is a string.

115200
time TimeSource | None

Injectable time source (tests). Defaults to the stdlib time module.

None
port_factory PortFactory | None

Injectable port factory (tests). Defaults to a real pyserial Serial.

None
reconnect_seconds float

When the device drops mid-tail, retry opening the port through port_factory for up to this many seconds before giving up and returning :attr:ExitCode.DISCONNECTED. Default :data:DEFAULT_TAIL_RECONNECT_SECONDS (30 s); set to 0.0 to disable retries (CI-friendly fail-fast). The window is additional to seconds — time spent reconnecting does not count against the tail budget, since the user's intent is "watch for seconds of output".

DEFAULT_TAIL_RECONNECT_SECONDS
reconnect_interval float

Sleep between reconnect attempts. Default :data:DEFAULT_TAIL_RECONNECT_INTERVAL (0.5 s).

DEFAULT_TAIL_RECONNECT_INTERVAL

Returns:

Type Description
ExitCode

class:ExitCode describing how the tail ended.

Interactive TUI

chumicro_repl.tui

Interactive REPL TUI — forwards stdin ↔ serial with highlighting.

The TUI is deliberately thin: stdin bytes are passed through to the board unchanged, serial bytes are UTF-8 decoded and ANSI-highlighted before being written to stdout. The board does its own line editing — arrow keys, backspace, paste-mode (Ctrl-E) all reach the device verbatim.

Keybindings mirror mpremote repl:

  • Ctrl-C — forwarded to the board. Raises :class:KeyboardInterrupt on-device, which cancels whatever is running and returns to the friendly REPL.
  • Ctrl-D — forwarded to the board. Soft-reboots the runtime (MicroPython prints MPY: soft reboot, CircuitPython is silent but rewinds).
  • Ctrl-E — forwarded to the board. Enters MicroPython paste mode; CircuitPython ignores it.
  • Ctrl-X — intercepted locally. Exits the TUI without rebooting or interrupting the board.

The loop is structured so tests can drive it without real terminal machinery: :func:run_loop takes injectable input / output / port / time dependencies; :func:interactive is the thin wrapper that opens a real pyserial port and puts stdin into raw mode.

KeyInputReader

Bases: Protocol

Structural interface for a non-blocking keyboard source.

The POSIX adapter reads from a termios raw-mode fd; the test fake replays scripted byte sequences. Both must expose a non-blocking read_available so the interactive loop never stalls waiting for input.

read_available()

Return every byte currently buffered, or b"" if none.

run_loop(port, keyboard, output, *, time=None, theme=None, exit_key=CTRL_X, welcome_banner='', initial_send=b'', reopen=None, reconnect_seconds=0.0, reconnect_interval=0.5)

Run the interactive I/O loop until exit_key is pressed.

Returns 0 on a normal Ctrl-X exit, or :attr:ExitCode.DISCONNECTED (3) when the device drops and no reconnect callback is supplied (or the reconnect budget is exhausted). Callers driving the loop from tests can inject custom keyboard / output / time to validate the forward-and-highlight behavior without a real serial port.

Parameters:

Name Type Description Default
port SerialPort

Open :class:SerialPort — the TUI writes keystrokes and reads device output through this port. Closing the port remains the caller's responsibility.

required
keyboard KeyInputReader

Non-blocking keystroke source.

required
output TextIO

Destination for serial output. ANSI escapes are written whether or not the stream is a TTY — the caller owns the decision to strip them downstream.

required
time TimeSource | None

Injectable time source.

None
theme Theme | None

Color theme for pattern highlighting.

None
exit_key bytes

Byte sequence that ends the loop without being forwarded to the device. Defaults to Ctrl-X.

CTRL_X
welcome_banner str

Local text written to output before the loop starts. Used by :func:interactive to introduce the connection (transport, address, keybinding hints); tests pass an empty string to keep output deterministic.

''
initial_send bytes

Bytes written to the port before the loop starts. Defaults to empty; :func:interactive sends a single carriage return so the friendly REPL reprints its >>> prompt instead of staring at a blank line.

b''
reopen Callable[[], SerialPort] | None

Callable returning a fresh :class:SerialPort, invoked when the current port drops and reconnect_seconds allows. Defaults to None (no auto-reconnect — the loop returns :attr:ExitCode.DISCONNECTED on the first disconnect). :func:interactive wires this to a closure over the original port factory so a replug picks up automatically.

None
reconnect_seconds float

Reconnect budget per disconnect, in seconds. 0.0 (default) means no retry; positive values loop reopen until success or budget exhausted. Ctrl-X during reconnect aborts immediately.

0.0
reconnect_interval float

Sleep between reconnect attempts.

0.5

interactive(device, *, baudrate=115200, input_stream=None, output=None, theme=None, time=None, port_factory=None, raw_mode_context=None, reconnect_seconds=DEFAULT_TUI_RECONNECT_SECONDS, reconnect_interval=0.5)

Open device and run the interactive TUI until Ctrl-X.

Thin wrapper over :func:run_loop — pulls in the POSIX terminal raw-mode setup, the real pyserial port factory, and the default stdin/stdout streams. Tests that want to drive the loop deterministically should call :func:run_loop directly with a fake :class:KeyInputReader and :class:SerialPort.

Parameters:

Name Type Description Default
device DeviceLike | str

:class:chumicro_deploy.Device or a serial path.

required
baudrate int

Consulted when device is a string.

115200
input_stream BinaryIO | None

Binary stdin. Defaults to sys.stdin.buffer.

None
output TextIO | None

Text output. Defaults to sys.stdout.

None
theme Theme | None

Color theme.

None
time TimeSource | None

Injectable time source.

None
port_factory PortFactory | None

Injectable port factory.

None
raw_mode_context Callable[[int], AbstractContextManager[None]] | None

Callable yielding a terminal raw-mode context manager for a given fd. Defaults to :func:_posix_raw_mode; tests pass a no-op.

None
reconnect_seconds float

How long to keep retrying the port factory after the device drops mid-session. Default :data:DEFAULT_TUI_RECONNECT_SECONDS (60 s); pass 0.0 to disable reconnect (the loop returns :attr:ExitCode.DISCONNECTED on the first OSError). Press Ctrl-X during the retry window to abort immediately.

DEFAULT_TUI_RECONNECT_SECONDS
reconnect_interval float

Sleep between reconnect attempts.

0.5

Returns:

Type Description
int

0 on normal Ctrl-X exit (including Ctrl-X during a

int

reconnect window); :attr:ExitCode.DISCONNECTED (3)

int

when the device drops and the reconnect budget runs out.

interactive_line(device, *, baudrate=115200, output=None, theme=None, time=None, port_factory=None, history_root=None)

Open device and run the line-mode REPL.

Sibling of :func:interactive — instead of forwarding keystrokes byte-by-byte to the device, runs a host-side prompt_toolkit line editor with persistent per-device history. Each completed line ships to the device; serial output streams back through the same pattern-detector + traceback highlighter the passthrough TUI uses.

Falls through to the same port factory / device-resolution shape as :func:interactive; the only delta is the input loop.

Parameters:

Name Type Description Default
device DeviceLike | str

:class:chumicro_deploy.Device or a serial path.

required
baudrate int

Consulted when device is a string.

115200
output TextIO | None

Text output. Defaults to sys.stdout.

None
theme Theme | None

Color theme.

None
time TimeSource | None

Injectable time source (for the per-line drain).

None
port_factory PortFactory | None

Injectable port factory.

None
history_root object | None

Override the persistent-history root directory. None (default) uses :data:chumicro_repl.line_mode.DEFAULT_HISTORY_ROOT.

None

Returns:

Type Description
int

0 on a clean exit (:quit / Ctrl-D / Ctrl-C at the

int

empty prompt).

Recovery layer

InteractiveReplSession is the wrapper around ReplSession that classifies session-start failures (port not found, port busy, permission denied, raw-REPL unresponsive) and walks the user through a recovery plan. Mid-session disconnects are not routed through this module — those use the auto-reconnect loop in tail() and run_loop().

chumicro_repl.recovery

Classifies chumicro-repl session-start failures and coaches the user through a fix.

A bad cable, a held port, a missing dialout membership, or a board stuck in unresponsive user code all surface as OSError or :class:ReplSessionError when a caller opens a :class:ReplSession. This module maps those exceptions to a small enum of recognized failure kinds, pairs each kind with a :class:RecoveryPlan of ordered physical steps the user can take, and offers a coaching loop that prints the plan and retries the session start.

Public surface:

  • :class:ReplFailureKind — enum of session-start failure modes.
  • :class:RecoveryPlan — headline + ordered fix-steps for one kind.
  • :func:classify_session_failure — map a session-start exception to a :class:ReplFailureKind.
  • :func:recovery_plan_for — look up the plan registered for a kind.
  • :func:coached_session_start — reusable classify-render-retry loop around any zero-arg session-opening callable.
  • :class:InteractiveReplSession — context manager that applies the coaching loop around a :class:ReplSession.

Mid-session disconnects do not route through this module. The auto-reconnect loop in :func:chumicro_repl.tail and :func:chumicro_repl.tui.run_loop handles the "device was open and then dropped" case; this module only addresses the "it never opened in the first place" failures.

ReplFailureKind

Bases: Enum

Classification of session-start failures.

Each kind has a :class:RecoveryPlan keyed in :data:_PLANS; :func:recovery_plan_for is the typed accessor.

RecoveryPlan dataclass

User-facing guidance for a single :class:ReplFailureKind.

Attributes:

Name Type Description
headline str

One-line summary the user reads first.

fix_steps tuple[str, ...]

Ordered physical actions the user can take. Rendered as a bulleted list by :class:InteractiveReplSession.

InteractiveReplSession

:class:ReplSession wrapper that coaches the user through start failures.

On a session-start failure, classifies the exception, prints the matching :class:RecoveryPlan to output, prompts the user to fix the condition + press Enter to retry, and tries again — up to max_attempts times.

Use as a context manager that yields a live :class:ReplSession::

from chumicro_repl import InteractiveReplSession

with InteractiveReplSession(device) as session:
    session.exec("print('hello')")

Mid-session disconnects (after the handshake completed) do NOT route through this wrapper — those are handled by the auto-reconnect loop in :func:tail and :func:run_loop.

Parameters:

Name Type Description Default
device DeviceLike | str

An object exposing .address (e.g. a chumicro_deploy.Device) or a bare serial-port path string. Forwarded to :class:ReplSession on each attempt.

required
max_attempts int

Ceiling on retry attempts. Defaults to 3. Must be >= 1.

_DEFAULT_MAX_ATTEMPTS
prompt Callable[[str], str]

Injectable prompt callable. Defaults to :func:input. An empty / whitespace-only response continues with a retry; "q" / "quit" / "abort" / "exit" re-raises the last error without retrying.

input
output Callable[[str], None]

Injectable output sink. Defaults to :func:print. Tests inject a list-append to make assertions.

print
baudrate int

Forwarded to :class:ReplSession when device is a string.

115200
time TimeSource | None

Forwarded to :class:ReplSession.

None
port_factory PortFactory | None

Forwarded to :class:ReplSession.

None
connect_timeout float

Forwarded to :class:ReplSession.

DEFAULT_TIMEOUT

Raises:

Type Description
ValueError

max_attempts < 1.

classify_session_failure(error)

Classify a session-start exception into a :class:ReplFailureKind.

Recognized inputs:

  • :class:OSError (including :class:serial.SerialException, which subclasses :class:OSError in modern pyserial). The classifier inspects errno first, then falls back to substring matching on the message — pyserial sometimes wraps the OS error and re-formats the message, so neither alone is reliable.
  • :class:ReplSessionDisconnected — the device dropped during the handshake write. Treated as :attr:PORT_NOT_FOUND because the user-facing fix is the same ("plug the device in and retry").
  • :class:ReplSessionError — the handshake completed without a raw-REPL prompt response. Mapped to :attr:RAW_REPL_UNRESPONSIVE.
  • Anything else — :attr:UNKNOWN.

Parameters:

Name Type Description Default
error Exception

The exception caught from :meth:ReplSession.__enter__ (or its inner port_factory call).

required

Returns:

Type Description
ReplFailureKind

The matching :class:ReplFailureKind.

recovery_plan_for(kind)

Return the :class:RecoveryPlan registered for kind.

Every member of :class:ReplFailureKind has an entry; the function is total.

coached_session_start(factory, *, output=print, prompt=input, max_attempts=_DEFAULT_MAX_ATTEMPTS)

Run factory with classify-coach-retry on session-start failures.

Wraps any zero-arg callable that may raise :class:OSError, :class:ReplSessionDisconnected, or :class:ReplSessionError on its connect step. On a recognized failure:

  1. :func:classify_session_failure maps the exception to a :class:ReplFailureKind.
  2. The matching :class:RecoveryPlan (headline + bulleted fix-steps) renders to output.
  3. prompt asks whether to retry; an abort response (q / quit / abort / exit) re-raises the last error. Empty / whitespace-only retries.
  4. After max_attempts attempts, the last error re-raises.

Used by:

  • :class:InteractiveReplSession — which wraps a :class:ReplSession context manager around the coaching loop so programmatic callers get the same retry behavior.
  • The workspace repl CLI — which wraps the :func:~chumicro_repl.tui.interactive_line / :func:~chumicro_repl.tui.interactive calls so port-busy / port-not-found / permission-denied errors get the user-friendly coaching prompt instead of a bare traceback.

Parameters:

Name Type Description Default
factory Callable[[], _T]

Zero-arg callable performing the session start. Returns any value on success (caller's choice — could be a :class:ReplSession, an int exit code, None, etc.); raises on connect failure.

required
output Callable[[str], None]

Sink for plan rendering. Defaults to :func:print.

print
prompt Callable[[str], str]

Asks the user to retry vs abort. Defaults to :func:input.

input
max_attempts int

Ceiling on retry attempts. Defaults to 3. Must be >= 1.

_DEFAULT_MAX_ATTEMPTS

Returns:

Type Description
_T

Whatever factory returns on a successful attempt.

Raises:

Type Description
ValueError

max_attempts < 1.

Exception

The last classified error after retries are exhausted or the user aborts.

Pattern detection

chumicro_repl.patterns

Pattern detectors for CircuitPython and MicroPython REPL output.

Detects the banners and blocks that mean "something noteworthy happened on the board" so the TUI can colorize them and so tail() can fail a deploy when a traceback arrives.

Recognized kinds:

  • :attr:PatternKind.TRACEBACK — the Traceback (most recent call last): header that both runtimes emit at the start of an uncaught exception block. Spans from the header line through the blank line that follows the exception summary.
  • :attr:PatternKind.SAFE_MODE — CircuitPython's safe mode block, printed when the supervisor drops into safe mode after repeated crashes or after supervisor.reload() into a broken state.
  • :attr:PatternKind.HARD_FAULT — the CircuitPython Hard fault message and the register dump that follows it.
  • :attr:PatternKind.SOFT_REBOOT — the MicroPython MPY: soft reboot banner that follows a machine.soft_reset() or a Ctrl-D in the interactive REPL.

The detector is line-oriented — pass a decoded string of lines (or a single complete line) to :func:detect_patterns and get back a list of :class:PatternMatch records, each carrying a span into the input string plus the detected kind.

PatternKind

Bases: StrEnum

The kinds of noteworthy output the detector recognizes.

Inherits from :class:enum.StrEnum so kind.value is interchangeable with str(kind), which keeps JSON / dict serialization trivial.

PatternMatch dataclass

One pattern detected in the streamed output.

Attributes:

Name Type Description
kind PatternKind

The :class:PatternKind of this match.

start int

Start index into the input string (inclusive).

end int

End index into the input string (exclusive).

text str

The matched substring.

StreamingPatternDetector

Scan an unbounded byte stream for noteworthy patterns.

Feed decoded text into :meth:feed as it arrives. The detector buffers a trailing window of buffer_limit characters so a pattern that spans a chunk boundary still matches, yields every match contained entirely within that window, and drops older content as new text arrives.

Parameters:

Name Type Description Default
buffer_limit int

Soft cap on buffered characters. Patterns longer than this limit are not detected; the default (8 KiB) comfortably covers any realistic traceback.

8 * 1024
total_fed property

Total characters ever passed through :meth:feed.

Lets a streaming consumer translate an absolute match offset back into the most-recently-fed chunk: chunk_start = total_fed - len(chunk).

feed(text)

Append text and return every newly-detected match.

Matches are emitted once — subsequent :meth:feed calls will not re-emit a pattern that was already returned. The returned :attr:PatternMatch.start / :attr:~PatternMatch.end indices are positions in the logical stream (total bytes fed), so a caller tracking the stream can index back into its own log.

detect_patterns(text)

Return every pattern match in text, sorted by start index.

Overlapping matches are possible in principle (a traceback inside a safe-mode block, say) but the three non-traceback patterns all terminate cleanly, so in practice matches are disjoint.

Parameters:

Name Type Description Default
text str

Decoded REPL output to scan.

required

Returns:

Type Description
list[PatternMatch]

A list of :class:PatternMatch records in order of first

list[PatternMatch]

byte. Empty list when no pattern matches.

Highlighting

colorize() and Theme are exposed at the top level (from chumicro_repl import colorize, Theme). strip_ansi_sequences lives at the submodule path (from chumicro_repl.highlight import strip_ansi_sequences) since it's mostly used by tests + log-capture plumbing.

chumicro_repl.highlight

ANSI colorization for detected REPL patterns.

Renders :class:~chumicro_repl.patterns.PatternMatch spans into terminal escape sequences. A small :class:Theme dataclass maps each :class:~chumicro_repl.patterns.PatternKind to an ANSI SGR string; the default theme uses the same color choices as mpremote + rich.traceback: red/bold for tracebacks and hard faults, yellow for safe mode, dim/cyan for soft-reboot banners.

Emits raw ANSI escape sequences only, with no rich dependency. Output destined for a non-TTY stream can be passed through :func:strip_ansi_sequences downstream to recover plain text.

Theme dataclass

ANSI styles for each recognized pattern kind.

Each field is the SGR parameter string (the digits between \x1b[ and m). Default values chosen to read cleanly on both light and dark terminal backgrounds:

  • traceback: bold red (1;31).
  • safe_mode: bold yellow (1;33), distinct from errors but still attention-grabbing.
  • hard_fault: bold red background (1;41), maximum attention since hard faults mean the chip is unhappy at a level below Python.
  • soft_reboot: dim cyan (2;36), informational only. The user asked for it.
style_for(kind)

Return the ANSI SGR parameter string for kind.

colorize(text, *, theme=None, matches=None)

Return text with ANSI escapes around each detected pattern.

When matches is None the detector runs over text directly; pre-compute matches and pass them explicitly for streaming callers that already know where the patterns start.

Overlapping matches are handled by walking in start-index order and skipping any match whose start falls inside a previously-emitted span. This keeps the escape sequences balanced (one reset per style).

Parameters:

Name Type Description Default
text str

Decoded REPL output.

required
theme Theme | None

Color theme. Defaults to :data:DEFAULT_THEME.

None
matches list[PatternMatch] | None

Pre-computed matches. Defaults to running the detector over text.

None

Returns:

Type Description
str

text with ANSI SGR sequences wrapping each pattern. When

str

no patterns match the original string is returned unchanged.

strip_ansi_sequences(text)

Return text with every ANSI escape sequence removed.

Intended for tests that assert on highlighted output — the ANSI bytes are noise in an assert "Traceback" in output context. Also useful when piping colorized output into a log file without losing human-readable content.

colorize_stream_chunk(text, matches, detector, *, theme=None)

Render text with ANSI highlighting using stream-coordinate matches.

A :class:~chumicro_repl.patterns.StreamingPatternDetector fed chunk by chunk returns matches whose start / end are absolute offsets from the start of the logical stream. This helper translates those offsets back into indices within the most-recently-fed chunk and wraps each in-range span via :func:colorize. Matches whose translated coordinates fall entirely outside text are skipped.

UTF-8 streaming decoder

chumicro_repl.framing

UTF-8 safe streaming decoder.

Serial reads arrive in arbitrary-sized chunks. A single UTF-8 code-point can span up to four bytes, and a read boundary can fall in the middle of a multi-byte sequence — so naive bytes.decode("utf-8") on each chunk raises :class:UnicodeDecodeError on anything above ASCII. :class:Utf8StreamDecoder buffers the tail bytes of a chunk that look like the start of a not-yet-complete sequence and prepends them to the next chunk.

Also forgives one class of garbage: if the stream emits invalid bytes (e.g. an interrupted power glitch), the decoder replaces them with U+FFFD so the user sees a replacement character instead of a crash.

Utf8StreamDecoder

Incremental UTF-8 decoder with chunk-boundary buffering.

One instance per live stream. :meth:decode returns every fully-decodable code point in the chunk so far, holding any trailing incomplete sequence for the next call. :meth:flush releases the remaining buffered bytes at end-of-stream. Both methods decode with errors="replace", so a caller never sees :class:UnicodeDecodeError.

Example::

decoder = Utf8StreamDecoder()
first_half = bytes([0xF0, 0x9F])  # start of "🙂" (4 bytes)
second_half = bytes([0x99, 0x82])
assert decoder.decode(first_half) == ""
assert decoder.decode(second_half) == "🙂"
decode(chunk)

Decode chunk, returning every fully-decodable code point.

Bytes from the start of a not-yet-complete sequence are held over for the next call. When chunk is empty the return value is an empty string.

flush()

Return any buffered bytes, decoded with error replacement.

Call at end-of-stream so a final incomplete sequence produces a :data:~str with the replacement character rather than silently dropping bytes.

pending()

Return the lead bytes held back awaiting their continuation.

These are the start of an incomplete trailing code point the decoder has not emitted yet. Returns a copy without consuming them, so a caller draining decoded text around a boundary can carry the partial sequence forward instead of dropping it; the buffer keeps holding the same bytes for the next :meth:decode.

Test fakes

chumicro_repl.testing

Test fakes for chumicro-repl.

Provides three fakes for host-side tests:

  • :class:FakeSerialPort — drop-in replacement for serial.Serial in :class:chumicro_repl.session.ReplSession, :func:chumicro_repl.tail, and :func:chumicro_repl.tui.run_loop. Records writes, replays scripted reads.
  • :class:FakeKeyboard — replays scripted keystrokes for :func:chumicro_repl.tui.run_loop tests.
  • :class:FakeTime — deterministic seconds-domain time source that satisfies the TimeSource protocol so tests never sleep.

FakeTime

Deterministic seconds-domain time source for host-side tests.

monotonic() is stable — repeated calls return the same value until the clock is explicitly advanced. sleep() advances the clock by duration without a real wait.

monotonic()

Return the current fake time in seconds.

sleep(duration)

Advance the clock by duration seconds (no wall-clock wait).

advance(seconds)

Move the clock forward by seconds without auto-sleep.

FakeSerialPort

In-memory replacement for :class:serial.Serial.

Records every write into :attr:writes and replays scripted reads from :attr:read_chunks. Each successful read(n) consumes one chunk regardless of n — tests that need to model fine-grained byte-level delivery should script per-byte chunks.

Instances are callable and return themselves, so a FakeSerialPort instance can be passed directly as a serial_port_factory kwarg — no wrapper closure needed.

Parameters:

Name Type Description Default
read_chunks Iterable[bytes | BaseException] | None

Items the next read(n) call returns, in order. Each item is either bytes (returned verbatim) or an instance of :class:BaseException (raised on the call — used to script OSError / SerialException disconnects mid-stream). When the script is exhausted, subsequent reads return b"" and in_waiting is 0.

None
raise_on_write BaseException | None

When set, the next write(...) call raises this exception instead of recording the data. Used to script disconnects that fail the host's first attempt to send a keystroke.

None
open_error Exception | None

When set, calling the instance (used as a factory) raises this exception. Lets tests script "the factory failed to open a port" without a closure.

None
in_waiting property

Length of the next scripted chunk, or 0 when exhausted.

Scripted BaseException entries are reported as having in_waiting == 1 so polling loops actually call :meth:read (which then raises) instead of looping past the scripted disconnect.

__call__(*_args, **_kwargs)

Return this port, or raise the scripted open_error.

Accepts both positional and keyword args so the same fake plugs into factories that use either calling convention.

read(size=1)

Return the next scripted chunk verbatim, or raise.

size is honored insofar as the chunk is no larger than what was scripted; tests that depend on partial chunking should script the chunks at the granularity they care about. Scripted exceptions are raised instead of returned.

write(data)

Record a write, or raise the configured exception.

close()

Mark the port as closed.

reset_input_buffer()

No-op — required by the :class:SerialPort protocol.

feed(*chunks)

Append more scripted chunks (or exceptions) after construction.

Lets a test set up a session, exec one script, and then prepare the next round of bytes (or a scripted disconnect) without rebuilding the port.

FakeKeyboard

Scripted keyboard input for :func:chumicro_repl.tui.run_loop.

Each entry in :attr:scripted_input is delivered as a single read_available() return value. When the script is exhausted, subsequent calls return b"" — the loop will eventually stall unless the test arranges another exit (typically by sending Ctrl-X as the last scripted entry).

read_available()

Return the next scripted byte block or b"" at EOF.

queue(*chunks)

Append more scripted chunks after construction.