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 parsedreprof 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:
ReplSessionDisconnectedsubclass 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 |
required |
baudrate
|
int
|
Only consulted when device is a string. Ignored
when device exposes its own |
115200
|
time
|
TimeSource | None
|
Injectable time source (for tests). Defaults to the
stdlib |
None
|
port_factory
|
PortFactory | None
|
Injectable port factory (for tests). Defaults
to :func: |
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: |
required |
*args
|
object
|
Positional arguments. Rendered via :func: |
()
|
timeout
|
float
|
Per-call deadline in seconds. |
DEFAULT_TIMEOUT
|
**kwargs
|
object
|
Keyword arguments. Rendered via :func: |
{}
|
Returns:
| Type | Description |
|---|---|
object
|
The parsed return value. |
object
|
|
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: |
required |
seconds
|
float
|
Length of the tail window. A :class: |
required |
fail_on_traceback
|
bool
|
When |
True
|
output
|
TextIO | None
|
Destination for the streamed text. Defaults to
|
None
|
theme
|
Theme | None
|
Color theme. Defaults to
:data: |
None
|
baudrate
|
int
|
Only consulted when device is a string. |
115200
|
time
|
TimeSource | None
|
Injectable time source (tests). Defaults to the stdlib
|
None
|
port_factory
|
PortFactory | None
|
Injectable port factory (tests). Defaults to
a real pyserial |
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: |
DEFAULT_TAIL_RECONNECT_SECONDS
|
reconnect_interval
|
float
|
Sleep between reconnect attempts.
Default :data: |
DEFAULT_TAIL_RECONNECT_INTERVAL
|
Returns:
| Type | Description |
|---|---|
ExitCode
|
class: |
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:
KeyboardInterrupton-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: |
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: |
''
|
initial_send
|
bytes
|
Bytes written to the port before the loop
starts. Defaults to empty; :func: |
b''
|
reopen
|
Callable[[], SerialPort] | None
|
Callable returning a fresh :class: |
None
|
reconnect_seconds
|
float
|
Reconnect budget per disconnect, in
seconds. |
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: |
required |
baudrate
|
int
|
Consulted when device is a string. |
115200
|
input_stream
|
BinaryIO | None
|
Binary stdin. Defaults to |
None
|
output
|
TextIO | None
|
Text output. Defaults to |
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: |
None
|
reconnect_seconds
|
float
|
How long to keep retrying the port
factory after the device drops mid-session. Default
:data: |
DEFAULT_TUI_RECONNECT_SECONDS
|
reconnect_interval
|
float
|
Sleep between reconnect attempts. |
0.5
|
Returns:
| Type | Description |
|---|---|
int
|
|
int
|
reconnect window); :attr: |
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: |
required |
baudrate
|
int
|
Consulted when device is a string. |
115200
|
output
|
TextIO | None
|
Text output. Defaults to |
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
|
Returns:
| Type | Description |
|---|---|
int
|
|
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
¶
: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 |
required |
max_attempts
|
int
|
Ceiling on retry attempts. Defaults to 3.
Must be |
_DEFAULT_MAX_ATTEMPTS
|
prompt
|
Callable[[str], str]
|
Injectable prompt callable. Defaults to
:func: |
input
|
output
|
Callable[[str], None]
|
Injectable output sink. Defaults to :func: |
print
|
baudrate
|
int
|
Forwarded to :class: |
115200
|
time
|
TimeSource | None
|
Forwarded to :class: |
None
|
port_factory
|
PortFactory | None
|
Forwarded to :class: |
None
|
connect_timeout
|
float
|
Forwarded to :class: |
DEFAULT_TIMEOUT
|
Raises:
| Type | Description |
|---|---|
ValueError
|
|
classify_session_failure(error)
¶
Classify a session-start exception into a :class:ReplFailureKind.
Recognized inputs:
- :class:
OSError(including :class:serial.SerialException, which subclasses :class:OSErrorin modern pyserial). The classifier inspectserrnofirst, 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_FOUNDbecause 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: |
required |
Returns:
| Type | Description |
|---|---|
ReplFailureKind
|
The matching :class: |
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:
- :func:
classify_session_failuremaps the exception to a :class:ReplFailureKind. - The matching :class:
RecoveryPlan(headline + bulleted fix-steps) renders to output. - prompt asks whether to retry; an abort response (
q/quit/abort/exit) re-raises the last error. Empty / whitespace-only retries. - After max_attempts attempts, the last error re-raises.
Used by:
- :class:
InteractiveReplSession— which wraps a :class:ReplSessioncontext manager around the coaching loop so programmatic callers get the same retry behavior. - The workspace
replCLI — which wraps the :func:~chumicro_repl.tui.interactive_line/ :func:~chumicro_repl.tui.interactivecalls 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: |
required |
output
|
Callable[[str], None]
|
Sink for plan rendering. Defaults to :func: |
print
|
prompt
|
Callable[[str], str]
|
Asks the user to retry vs abort. Defaults to
:func: |
input
|
max_attempts
|
int
|
Ceiling on retry attempts. Defaults to 3.
Must be |
_DEFAULT_MAX_ATTEMPTS
|
Returns:
| Type | Description |
|---|---|
_T
|
Whatever factory returns on a successful attempt. |
Raises:
| Type | Description |
|---|---|
ValueError
|
|
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— theTraceback (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'ssafe modeblock, printed when the supervisor drops into safe mode after repeated crashes or aftersupervisor.reload()into a broken state. - :attr:
PatternKind.HARD_FAULT— the CircuitPythonHard faultmessage and the register dump that follows it. - :attr:
PatternKind.SOFT_REBOOT— the MicroPythonMPY: soft rebootbanner that follows amachine.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: |
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: |
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: |
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 forserial.Serialin :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_looptests. - :class:
FakeTime— deterministic seconds-domain time source that satisfies theTimeSourceprotocol 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.
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 |
None
|
raise_on_write
|
BaseException | None
|
When set, the next |
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).