feat(rawcli): flexible entry + parser (Phase 2)
- parser.py: classify a line into control / function-call / raw payload; parse loose hex/binary with intermixed 0x/0b tokens (whitespace- insensitive, defaulting to the session entry mode) - entry.py: byte_space() byte-group formatting + key bindings — Ctrl-/ toggles hex/binary entry mode (breadcrumb updates), digits auto-space into byte groups as you type - app.py: dispatch() drives the loop via parse_line (testable without a live prompt); identify/call are stubbed for Phases 3/4 14 new tests (parser tokens/intermix/errors, byte_space, dispatch). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,8 @@ from __future__ import annotations
|
||||
import sys
|
||||
|
||||
from .session import RawSession
|
||||
from .parser import parse_line
|
||||
from .entry import install_key_bindings
|
||||
from .trace_view import render_exchange
|
||||
|
||||
_HELP = """\
|
||||
@@ -45,12 +47,48 @@ def _raw_exchange(device, payload: bytes) -> bytes | None:
|
||||
return None
|
||||
|
||||
|
||||
def dispatch(state: RawSession, line: str) -> bool:
|
||||
"""Handle one input line. Returns False to exit the REPL, True to keep going. Testable
|
||||
without a live prompt."""
|
||||
try:
|
||||
cmd = parse_line(line, state.entry_mode)
|
||||
except ValueError as exc:
|
||||
print(f"rawcli: {exc}")
|
||||
return True
|
||||
|
||||
if cmd.kind == "control":
|
||||
if cmd.name in ("quit", "exit", "q"):
|
||||
return False
|
||||
if cmd.name == "help":
|
||||
print(_HELP)
|
||||
else:
|
||||
# identify / transponder / close land in Phase 3
|
||||
print(f"rawcli: '{cmd.name}' is not wired yet (coming in a later phase)")
|
||||
return True
|
||||
|
||||
if cmd.kind == "call":
|
||||
# function-style transponder commands land in Phase 4 (the command catalog)
|
||||
print(f"rawcli: {cmd.name}(...) needs an identified transponder + catalog (Phase 4)")
|
||||
return True
|
||||
|
||||
# raw payload
|
||||
if not cmd.payload:
|
||||
return True
|
||||
response = _raw_exchange(state.device, cmd.payload)
|
||||
print(render_exchange(cmd.payload, response, protocol="hf14a",
|
||||
is_tty=sys.stdout.isatty()))
|
||||
return True
|
||||
|
||||
|
||||
def run(port=None) -> int:
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.patch_stdout import patch_stdout
|
||||
|
||||
state = RawSession(device=_connect(port))
|
||||
session = PromptSession()
|
||||
kb = KeyBindings()
|
||||
install_key_bindings(kb, state)
|
||||
session = PromptSession(key_bindings=kb)
|
||||
|
||||
def message():
|
||||
return state.breadcrumb() + " "
|
||||
@@ -58,23 +96,12 @@ def run(port=None) -> int:
|
||||
with patch_stdout():
|
||||
while True:
|
||||
try:
|
||||
line = session.prompt(message).strip()
|
||||
line = session.prompt(message)
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
break
|
||||
if not line:
|
||||
if not line.strip():
|
||||
continue
|
||||
if line in ("quit", "exit", "q"):
|
||||
if not dispatch(state, line):
|
||||
break
|
||||
if line == "help":
|
||||
print(_HELP)
|
||||
continue
|
||||
try:
|
||||
payload = bytes.fromhex(line.lower().replace("0x", "").replace(" ", ""))
|
||||
except ValueError:
|
||||
print(f"rawcli: could not parse {line!r} as hex")
|
||||
continue
|
||||
response = _raw_exchange(state.device, payload)
|
||||
print(render_exchange(payload, response, protocol="hf14a",
|
||||
is_tty=sys.stdout.isatty()))
|
||||
print("bye")
|
||||
return 0
|
||||
|
||||
52
pm3py/cli/rawcli/entry.py
Normal file
52
pm3py/cli/rawcli/entry.py
Normal file
@@ -0,0 +1,52 @@
|
||||
"""Numeric entry helpers: byte-group spacing for hex/binary input and the prompt_toolkit key
|
||||
bindings that toggle entry mode (Ctrl-/) and auto-space digits as you type.
|
||||
|
||||
The pure ``byte_space`` / ``group_size`` functions carry the logic and are unit-tested; the key
|
||||
bindings are thin wiring exercised in the live REPL.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
HEX = "hex"
|
||||
BIN = "bin"
|
||||
_DIGITS = {HEX: set("0123456789abcdefABCDEF"), BIN: set("01")}
|
||||
|
||||
|
||||
def group_size(mode: str) -> int:
|
||||
"""Characters per byte group: 2 for hex, 8 for binary."""
|
||||
return 2 if mode == HEX else 8
|
||||
|
||||
|
||||
def byte_space(text: str, mode: str = HEX) -> str:
|
||||
"""Re-group a run of hex/binary digits into space-separated byte groups.
|
||||
|
||||
Leaves a token that carries an explicit ``0x``/``0b`` prefix untouched (so intermixed input
|
||||
isn't mangled); only a bare digit run is regrouped."""
|
||||
if "x" in text.lower() or "b" in text.lower():
|
||||
return text
|
||||
n = group_size(mode)
|
||||
compact = text.replace(" ", "")
|
||||
return " ".join(compact[i:i + n] for i in range(0, len(compact), n))
|
||||
|
||||
|
||||
def install_key_bindings(kb, session) -> None:
|
||||
"""Wire Ctrl-/ (toggle entry mode) and live auto-byte-spacing of digit input onto ``kb``.
|
||||
|
||||
``session`` is the :class:`~pm3py.cli.rawcli.session.RawSession` whose ``entry_mode`` is
|
||||
toggled and read."""
|
||||
from prompt_toolkit.document import Document
|
||||
|
||||
@kb.add("c-/")
|
||||
def _toggle(event):
|
||||
session.toggle_entry_mode()
|
||||
event.app.invalidate() # redraw the breadcrumb
|
||||
|
||||
def _regroup_after(event):
|
||||
buf = event.current_buffer
|
||||
buf.insert_text(event.data)
|
||||
# only auto-space a bare digit run (no prefix), matching byte_space's guard
|
||||
grouped = byte_space(buf.text, session.entry_mode)
|
||||
if grouped != buf.text:
|
||||
buf.document = Document(grouped, len(grouped))
|
||||
|
||||
for ch in sorted(_DIGITS[HEX] | _DIGITS[BIN]):
|
||||
kb.add(ch)(_regroup_after)
|
||||
83
pm3py/cli/rawcli/parser.py
Normal file
83
pm3py/cli/rawcli/parser.py
Normal file
@@ -0,0 +1,83 @@
|
||||
"""rawcli line parsing.
|
||||
|
||||
Classifies an input line into a **control** command (help/identify/…), a **function-style call**
|
||||
(``READ(4)``), or a **raw** byte payload, and parses loose hex/binary — intermixed ``0x``/``0b``
|
||||
tokens, whitespace-insensitive, defaulting to the session's current entry mode when a token has no
|
||||
prefix.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
HEX = "hex"
|
||||
BIN = "bin"
|
||||
|
||||
#: single-word control verbs the REPL handles directly
|
||||
CONTROL_VERBS = {"help", "identify", "transponder", "close", "quit", "exit", "q"}
|
||||
|
||||
_CALL_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*\((.*)\)\s*$", re.DOTALL)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Command:
|
||||
"""A parsed rawcli line. ``kind`` is ``"control"``, ``"call"`` or ``"raw"``."""
|
||||
kind: str
|
||||
name: str = ""
|
||||
args: list[str] = field(default_factory=list)
|
||||
payload: bytes = b""
|
||||
|
||||
|
||||
def parse_token(token: str, default_mode: str = HEX) -> bytes:
|
||||
"""Parse one hex/binary token to bytes. A ``0x``/``0b`` prefix picks the base; otherwise
|
||||
``default_mode`` is used."""
|
||||
text = token
|
||||
mode = default_mode
|
||||
low = text.lower()
|
||||
if low.startswith("0x"):
|
||||
mode, text = HEX, text[2:]
|
||||
elif low.startswith("0b"):
|
||||
mode, text = BIN, text[2:]
|
||||
if text == "":
|
||||
return b""
|
||||
if mode == HEX:
|
||||
if len(text) % 2:
|
||||
raise ValueError(f"hex token {token!r} has an odd number of digits")
|
||||
try:
|
||||
return bytes.fromhex(text)
|
||||
except ValueError:
|
||||
raise ValueError(f"invalid hex token {token!r}") from None
|
||||
# binary
|
||||
if len(text) % 8:
|
||||
raise ValueError(f"binary token {token!r} is not a whole number of bytes "
|
||||
f"({len(text)} bits)")
|
||||
try:
|
||||
return bytes(int(text[i:i + 8], 2) for i in range(0, len(text), 8))
|
||||
except ValueError:
|
||||
raise ValueError(f"invalid binary token {token!r}") from None
|
||||
|
||||
|
||||
def parse_bytes(text: str, default_mode: str = HEX) -> bytes:
|
||||
"""Parse a whitespace-separated raw payload (intermixed 0x/0b tokens allowed)."""
|
||||
out = bytearray()
|
||||
for tok in text.split():
|
||||
out += parse_token(tok, default_mode)
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def _split_args(inside: str) -> list[str]:
|
||||
return [a.strip() for a in inside.split(",")] if inside.strip() else []
|
||||
|
||||
|
||||
def parse_line(line: str, default_mode: str = HEX) -> Command:
|
||||
"""Classify and parse a full input line."""
|
||||
stripped = line.strip()
|
||||
m = _CALL_RE.match(stripped)
|
||||
if m:
|
||||
return Command("call", name=m.group(1), args=_split_args(m.group(2)))
|
||||
first = stripped.split(None, 1)
|
||||
verb = first[0].lower() if first else ""
|
||||
if verb in CONTROL_VERBS:
|
||||
rest = first[1].split() if len(first) > 1 else []
|
||||
return Command("control", name=verb, args=rest)
|
||||
return Command("raw", payload=parse_bytes(stripped, default_mode))
|
||||
@@ -1,9 +1,14 @@
|
||||
"""rawcli Phase 1 — session state, breadcrumb, trace rendering, CLI dispatch. Hardware-free."""
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
from pm3py.cli.main import build_parser, main
|
||||
from pm3py.cli.rawcli.session import RawSession
|
||||
from pm3py.cli.rawcli.trace_view import render_exchange
|
||||
from pm3py.cli.rawcli.parser import parse_token, parse_bytes, parse_line
|
||||
from pm3py.cli.rawcli.entry import byte_space
|
||||
from pm3py.cli.rawcli.app import dispatch
|
||||
|
||||
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
||||
|
||||
@@ -65,3 +70,71 @@ class TestCliDispatch:
|
||||
rc = main([])
|
||||
assert rc == 1
|
||||
assert "rawcli" in capsys.readouterr().out
|
||||
|
||||
|
||||
class TestParser:
|
||||
def test_hex_tokens(self):
|
||||
assert parse_token("30") == b"\x30"
|
||||
assert parse_token("0x3004") == b"\x30\x04"
|
||||
assert parse_bytes("30 04") == b"\x30\x04"
|
||||
assert parse_bytes("3004") == b"\x30\x04" # default hex mode
|
||||
|
||||
def test_binary_tokens(self):
|
||||
assert parse_token("0b00110000") == b"\x30"
|
||||
assert parse_token("00110000", "bin") == b"\x30"
|
||||
|
||||
def test_intermixed(self):
|
||||
assert parse_bytes("0b00110000 0x04") == b"\x30\x04"
|
||||
assert parse_bytes("0x30 00000100", "bin") == b"\x30\x04" # bare token uses mode
|
||||
|
||||
def test_bad_tokens(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_token("303") # odd hex
|
||||
with pytest.raises(ValueError):
|
||||
parse_token("0b0011") # not a whole byte
|
||||
|
||||
def test_classify_control(self):
|
||||
c = parse_line("identify")
|
||||
assert c.kind == "control" and c.name == "identify"
|
||||
assert parse_line("help ntag215").kind == "control"
|
||||
|
||||
def test_classify_call(self):
|
||||
c = parse_line("READ(4)")
|
||||
assert c.kind == "call" and c.name == "READ" and c.args == ["4"]
|
||||
c = parse_line("WRITE(4, 0x00112233)")
|
||||
assert c.args == ["4", "0x00112233"]
|
||||
assert parse_line("GET_DATA()").kind == "call"
|
||||
|
||||
def test_classify_raw(self):
|
||||
c = parse_line("30 04")
|
||||
assert c.kind == "raw" and c.payload == b"\x30\x04"
|
||||
|
||||
|
||||
class TestByteSpace:
|
||||
def test_hex_grouping(self):
|
||||
assert byte_space("3004", "hex") == "30 04"
|
||||
assert byte_space("300", "hex") == "30 0"
|
||||
assert byte_space("30 04", "hex") == "30 04" # idempotent
|
||||
|
||||
def test_binary_grouping(self):
|
||||
assert byte_space("0011000000000100", "bin") == "00110000 00000100"
|
||||
|
||||
def test_prefixed_left_alone(self):
|
||||
assert byte_space("0x3004", "hex") == "0x3004" # explicit prefix not mangled
|
||||
|
||||
|
||||
class TestDispatch:
|
||||
def test_quit_returns_false(self):
|
||||
assert dispatch(RawSession(), "quit") is False
|
||||
|
||||
def test_help_prints(self, capsys):
|
||||
assert dispatch(RawSession(), "help") is True
|
||||
assert "rawcli" in capsys.readouterr().out
|
||||
|
||||
def test_raw_renders_without_device(self, capsys):
|
||||
assert dispatch(RawSession(), "30 04") is True
|
||||
assert "30 04" in _plain(capsys.readouterr().out)
|
||||
|
||||
def test_bad_hex_reports(self, capsys):
|
||||
assert dispatch(RawSession(), "0b0011") is True
|
||||
assert "whole number of bytes" in capsys.readouterr().out
|
||||
|
||||
Reference in New Issue
Block a user