fix(rawcli): raw response crash + hex/binary opcode completion
- Crash on raw hex: reader raw() returns `data` as a hex STRING; the raw
path fed that to bytes() -> "string argument without an encoding".
_raw_exchange now returns the bytes (`raw` field, or converts `data`);
render_exchange accepts bytes or a hex string.
- Completion for numeric entry, transponder-dependent: typing hex ("60")
or binary ("01100000") now matches the identified tag's command opcodes
and surfaces the named command (GET_VERSION) with its help. Honours a
0x/0b prefix and the current entry mode.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -52,7 +52,13 @@ def _raw_exchange(device, payload: bytes, protocol: str = "hf14a") -> bytes | No
|
|||||||
resp = raw(payload) if raw else None
|
resp = raw(payload) if raw else None
|
||||||
else:
|
else:
|
||||||
return None # lf raw exchange lands in a later phase
|
return None # lf raw exchange lands in a later phase
|
||||||
return resp.get("data") if isinstance(resp, dict) else None
|
if not isinstance(resp, dict):
|
||||||
|
return None
|
||||||
|
data = resp.get("raw") # bytes; "data" is the hex string form
|
||||||
|
if data is None:
|
||||||
|
hexstr = resp.get("data")
|
||||||
|
data = bytes.fromhex(hexstr) if isinstance(hexstr, str) else hexstr
|
||||||
|
return data
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
print(f"rawcli: exchange failed ({type(exc).__name__}: {exc})", file=sys.stderr)
|
print(f"rawcli: exchange failed ({type(exc).__name__}: {exc})", file=sys.stderr)
|
||||||
return None
|
return None
|
||||||
|
|||||||
@@ -1,9 +1,14 @@
|
|||||||
"""ipython-style completion for rawcli.
|
"""ipython-style completion for rawcli.
|
||||||
|
|
||||||
Completes the leading command token — control verbs plus the identified transponder's catalog
|
Completes the leading token in two ways, both driven by the **identified transponder**:
|
||||||
commands — with each command's help as the completion **meta** (the tooltip). ``help <cmd>``
|
|
||||||
completes catalog command names too. Paired in the app with menu completion while typing and
|
- by name — control verbs and the transponder's catalog commands (help text as the tooltip);
|
||||||
history auto-suggestion for the ipython feel.
|
- by **opcode** — as you type hex or binary bytes, the transponder's command opcodes are matched
|
||||||
|
against the input (respecting a ``0x``/``0b`` prefix and the current entry mode), so ``60`` and
|
||||||
|
``01100000`` both surface ``GET_VERSION``.
|
||||||
|
|
||||||
|
``help <cmd>`` completes catalog command names too. Paired in the app with menu completion while
|
||||||
|
typing and history auto-suggestion for the ipython feel.
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
@@ -12,6 +17,18 @@ from prompt_toolkit.completion import Completer, Completion
|
|||||||
from .parser import CONTROL_VERBS
|
from .parser import CONTROL_VERBS
|
||||||
from .catalog import catalog_for
|
from .catalog import catalog_for
|
||||||
|
|
||||||
|
_HEXDIGITS = set("0123456789abcdefABCDEF")
|
||||||
|
_BINDIGITS = set("01")
|
||||||
|
|
||||||
|
|
||||||
|
def _opcode(tc) -> int | None:
|
||||||
|
"""The command's first byte (opcode), obtained by building it with placeholder args."""
|
||||||
|
try:
|
||||||
|
built = tc.build(*(["0"] * len(tc.params)))
|
||||||
|
return built[0] if built else None
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
class RawCompleter(Completer):
|
class RawCompleter(Completer):
|
||||||
def __init__(self, session):
|
def __init__(self, session):
|
||||||
@@ -26,22 +43,59 @@ class RawCompleter(Completer):
|
|||||||
|
|
||||||
completing_first = (" " not in stripped) or (len(tokens) == 1 and at_word)
|
completing_first = (" " not in stripped) or (len(tokens) == 1 and at_word)
|
||||||
if completing_first:
|
if completing_first:
|
||||||
yield from self._commands(word)
|
yield from self._first_token(word)
|
||||||
elif tokens and tokens[0].lower() == "help":
|
elif tokens and tokens[0].lower() == "help":
|
||||||
yield from self._catalog_names(word)
|
yield from self._catalog_names(word)
|
||||||
|
|
||||||
def _commands(self, word: str):
|
def _first_token(self, word: str):
|
||||||
low = word.lower()
|
low = word.lower()
|
||||||
for verb in sorted(CONTROL_VERBS):
|
for verb in sorted(CONTROL_VERBS):
|
||||||
if verb.startswith(low):
|
if verb.startswith(low):
|
||||||
yield Completion(verb, start_position=-len(word), display_meta="command")
|
yield Completion(verb, start_position=-len(word), display_meta="command")
|
||||||
|
|
||||||
catalog = catalog_for(self._session)
|
catalog = catalog_for(self._session)
|
||||||
if catalog is not None:
|
if catalog is None:
|
||||||
for name in catalog.names():
|
return
|
||||||
if name.lower().startswith(low):
|
|
||||||
tc = catalog.get(name)
|
# by command name
|
||||||
yield Completion(f"{name}(", start_position=-len(word),
|
for name in catalog.names():
|
||||||
display=tc.usage(), display_meta=tc.help)
|
if name.lower().startswith(low):
|
||||||
|
tc = catalog.get(name)
|
||||||
|
yield Completion(f"{name}(", start_position=-len(word),
|
||||||
|
display=tc.usage(), display_meta=tc.help)
|
||||||
|
|
||||||
|
# by opcode, from hex/binary input (transponder-dependent)
|
||||||
|
yield from self._opcode_matches(word, catalog)
|
||||||
|
|
||||||
|
def _numeric(self, word: str) -> tuple[str, str]:
|
||||||
|
"""(digits, base) for an entry token — honours a 0x/0b prefix, else the session mode."""
|
||||||
|
low = word.lower()
|
||||||
|
if low.startswith("0x"):
|
||||||
|
return low[2:], "hex"
|
||||||
|
if low.startswith("0b"):
|
||||||
|
return low[2:], "bin"
|
||||||
|
return low, getattr(self._session, "entry_mode", "hex")
|
||||||
|
|
||||||
|
def _opcode_matches(self, word: str, catalog):
|
||||||
|
digits, base = self._numeric(word)
|
||||||
|
if not digits:
|
||||||
|
return
|
||||||
|
if base == "hex":
|
||||||
|
valid, render = _HEXDIGITS, (lambda op: f"{op:02x}")
|
||||||
|
else:
|
||||||
|
valid, render = _BINDIGITS, (lambda op: f"{op:08b}")
|
||||||
|
if any(ch not in valid for ch in digits):
|
||||||
|
return
|
||||||
|
low = word.lower()
|
||||||
|
for name in catalog.names():
|
||||||
|
if name.lower().startswith(low):
|
||||||
|
continue # already offered by name
|
||||||
|
tc = catalog.get(name)
|
||||||
|
op = _opcode(tc)
|
||||||
|
if op is not None and render(op).startswith(digits):
|
||||||
|
yield Completion(f"{name}(", start_position=-len(word),
|
||||||
|
display=f"{name} (0x{op:02X})",
|
||||||
|
display_meta=f"opcode 0x{op:02X} — {tc.help}")
|
||||||
|
|
||||||
def _catalog_names(self, word: str):
|
def _catalog_names(self, word: str):
|
||||||
catalog = catalog_for(self._session)
|
catalog = catalog_for(self._session)
|
||||||
|
|||||||
@@ -12,13 +12,20 @@ _PROTOCOLS = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def render_exchange(command: bytes, response: bytes | None, *,
|
def _as_bytes(value) -> bytes:
|
||||||
protocol: str = "hf14a", is_tty: bool | None = None) -> str:
|
"""Accept bytes or a hex string (which is what the reader `raw()` returns as `data`)."""
|
||||||
|
if isinstance(value, str):
|
||||||
|
return bytes.fromhex(value.replace(" ", ""))
|
||||||
|
return bytes(value)
|
||||||
|
|
||||||
|
|
||||||
|
def render_exchange(command, response, *, protocol: str = "hf14a", is_tty: bool | None = None) -> str:
|
||||||
"""Return the annotated trace for one exchange: a ``Reader → Tag`` line for ``command`` and,
|
"""Return the annotated trace for one exchange: a ``Reader → Tag`` line for ``command`` and,
|
||||||
if present, a ``Tag → Reader`` line for ``response``. ``is_tty`` forces/suppresses ANSI."""
|
if present, a ``Tag → Reader`` line for ``response``. Both accept bytes or a hex string.
|
||||||
|
``is_tty`` forces/suppresses ANSI."""
|
||||||
decoder, crc_len = _PROTOCOLS.get(protocol, (None, 0))
|
decoder, crc_len = _PROTOCOLS.get(protocol, (None, 0))
|
||||||
fmt = TraceFormatter(mode="reader", decoder=decoder, crc_len=crc_len, is_tty=is_tty)
|
fmt = TraceFormatter(mode="reader", decoder=decoder, crc_len=crc_len, is_tty=is_tty)
|
||||||
lines = [fmt.format(0, bytes(command)).strip("\n")]
|
lines = [fmt.format(0, _as_bytes(command)).strip("\n")]
|
||||||
if response:
|
if response:
|
||||||
lines.append(fmt.format(1, bytes(response)).strip("\n"))
|
lines.append(fmt.format(1, _as_bytes(response)).strip("\n"))
|
||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
|
|||||||
@@ -66,6 +66,11 @@ class TestTraceView:
|
|||||||
out = _plain(render_exchange(bytes([0x30, 0x04]), None, protocol="hf14a", is_tty=False))
|
out = _plain(render_exchange(bytes([0x30, 0x04]), None, protocol="hf14a", is_tty=False))
|
||||||
assert "READ" in out.upper()
|
assert "READ" in out.upper()
|
||||||
|
|
||||||
|
def test_accepts_hex_string_response(self):
|
||||||
|
# reader raw() returns `data` as a hex STRING — must not crash on bytes()
|
||||||
|
out = _plain(render_exchange(b"\x60", "0004040201000f03", protocol="hf14a", is_tty=False))
|
||||||
|
assert "60" in out and "00 04 04" in out
|
||||||
|
|
||||||
|
|
||||||
class TestCliDispatch:
|
class TestCliDispatch:
|
||||||
def test_rawcli_subcommand_parsed(self):
|
def test_rawcli_subcommand_parsed(self):
|
||||||
@@ -380,6 +385,19 @@ class TestCompleter:
|
|||||||
cs = self._complete(s, "help GET")
|
cs = self._complete(s, "help GET")
|
||||||
assert any(c.text == "GET_VERSION" for c in cs)
|
assert any(c.text == "GET_VERSION" for c in cs)
|
||||||
|
|
||||||
|
def test_opcode_completion_hex(self):
|
||||||
|
s = RawSession()
|
||||||
|
s.protocol, s.transponder = "hf14a", "NTAG213"
|
||||||
|
cs = self._complete(s, "60") # GET_VERSION opcode
|
||||||
|
assert any(c.text == "GET_VERSION(" for c in cs)
|
||||||
|
|
||||||
|
def test_opcode_completion_binary(self):
|
||||||
|
s = RawSession()
|
||||||
|
s.protocol, s.transponder = "hf14a", "NTAG213"
|
||||||
|
s.entry_mode = "bin"
|
||||||
|
cs = self._complete(s, "01100000") # 0x60 in binary
|
||||||
|
assert any(c.text == "GET_VERSION(" for c in cs)
|
||||||
|
|
||||||
|
|
||||||
class TestKeyBindings:
|
class TestKeyBindings:
|
||||||
def test_install_does_not_raise(self):
|
def test_install_does_not_raise(self):
|
||||||
|
|||||||
Reference in New Issue
Block a user