Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 62 additions & 2 deletions omotion/MotionConfig.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import struct
import json
import math
from dataclasses import dataclass
from typing import Optional, Any, Dict
import logging
Expand All @@ -11,6 +12,53 @@
MOTION_MAGIC = 0x4D4F5449 # 'MOTI'
MOTION_VER = 0x00010000 # v1.0.0

# Longest config JSON the console stores, in bytes: one 2036-byte flash page
# minus the 16-byte header, minus the terminating NUL (console-fw
# motion_config.h MOTION_CFG_JSON_MAX - 1). The console refuses a longer one.
MOTION_CFG_JSON_MAX_LEN = 2036 - 16 - 1


def _reject_constant(name: str):
raise ValueError(f"non-JSON constant {name}")


def _finite_float(text: str) -> float:
v = float(text)
if not math.isfinite(v):
raise ValueError(f"number {text} is out of range")
return v


def strict_json_problem(text: str) -> Optional[str]:
"""Why ``text`` is not a config the console firmware accepts, or None.

The console (console-fw fix/jsmn-strict) stores a config only when it is
a strict JSON object: no NaN / Infinity / -Infinity, no out-of-range
numbers, and at most MOTION_CFG_JSON_MAX_LEN bytes. Python's json module
accepts NaN and Infinity by default, so this check is stricter than
``json.loads``.
"""
if len(text.encode("utf-8")) > MOTION_CFG_JSON_MAX_LEN:
return f"{len(text.encode('utf-8'))} bytes, more than {MOTION_CFG_JSON_MAX_LEN}"
try:
data = json.loads(text, parse_constant=_reject_constant, parse_float=_finite_float)
except ValueError as e:
return str(e)
if not isinstance(data, dict):
return "root is not a JSON object"
return None


def _non_finite_paths(value: Any, path: str = "") -> list:
"""Key paths of float values that are NaN or infinite."""
if isinstance(value, float):
return [] if math.isfinite(value) else [path or "<root>"]
if isinstance(value, dict):
return [p for k, v in value.items() for p in _non_finite_paths(v, f"{path}.{k}" if path else str(k))]
if isinstance(value, (list, tuple)):
return [p for i, v in enumerate(value) for p in _non_finite_paths(v, f"{path}[{i}]")]
return []


@dataclass
class MotionConfigHeader:
Expand Down Expand Up @@ -138,10 +186,22 @@ def to_wire_bytes(self) -> bytes:

Returns:
bytes: Wire format [header][json_bytes]

Raises:
ValueError: if a value is NaN or infinite (not JSON; the console
refuses it), or the JSON is longer than MOTION_CFG_JSON_MAX_LEN.
"""
# Convert JSON to bytes
json_str = json.dumps(self.json_data, separators=(",", ":"))
bad = _non_finite_paths(self.json_data)
if bad:
raise ValueError(f"config values are NaN or infinite: {', '.join(bad)}")
# allow_nan=False: json.dumps would otherwise write NaN / Infinity.
json_str = json.dumps(self.json_data, separators=(",", ":"), allow_nan=False)
json_bytes = json_str.encode("utf-8")
if len(json_bytes) > MOTION_CFG_JSON_MAX_LEN:
raise ValueError(
f"config JSON is {len(json_bytes)} bytes; the console stores at most "
f"{MOTION_CFG_JSON_MAX_LEN}"
)

# Update header with JSON length
self.header.json_len = len(json_bytes)
Expand Down
8 changes: 7 additions & 1 deletion omotion/MotionConsole.py
Original file line number Diff line number Diff line change
Expand Up @@ -2282,7 +2282,13 @@ def write_config(self, config: MotionConfig) -> Optional[MotionConfig]:
self.uart.clear_buffer()

if r.packetType == OW_ERROR:
logger.error("Error writing config to device")
# Console firmware with the strict-config check refuses a
# config that is not a strict JSON object or does not fit the
# page, and keeps the stored one.
logger.error(
"Console refused the config write (not strict JSON, or too "
"long); the stored config is unchanged"
)
return None

# The firmware write ACK returns only the updated 16-byte header
Expand Down
96 changes: 96 additions & 0 deletions scripts/check_console_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
#!/usr/bin/env python3
"""
check_console_config.py

Check the user config stored on a console before updating it to firmware
with the strict-config check (console-fw fix/jsmn-strict, CFW-12/CFW-15).

That firmware parses the stored config in jsmn strict mode at boot. A
stored config that is not strict JSON (NaN, Infinity, -Infinity, a bare
word, an out-of-range number) is no longer applied, which leaves the TEC
thermal trip and the safety limits unset. Run this on each console while it
still runs the old firmware; any console that fails needs its config fixed
and rewritten first.

The raw stored JSON text is checked, not Python's view of it: json.loads
accepts NaN and Infinity, the console does not.

Reads only; writes nothing.

Usage
-----
python scripts/check_console_config.py

Exit code 0 when the stored config passes, 1 when it does not, 2 when the
console cannot be read.
"""

import json
import math
import struct
import sys

from omotion import MotionInterface
from omotion.MotionConfig import MOTION_CFG_JSON_MAX_LEN, strict_json_problem
from omotion.config import OW_CMD, OW_CMD_USR_CFG, OW_ERROR


def read_raw_config(console) -> str:
"""The stored config JSON exactly as the console holds it."""
r = console.uart.send_packet(
id=None, packetType=OW_CMD, command=OW_CMD_USR_CFG, reserved=0
)
console.uart.clear_buffer()
if r is None or r.packetType == OW_ERROR or r.data_len < 16:
raise RuntimeError("console did not return its config")
data = bytes(r.data)
_magic, _version, _seq, _crc, json_len = struct.unpack("<IIIHH", data[:16])
return data[16:16 + json_len].decode("utf-8", errors="replace").rstrip("\x00")


def main() -> int:
iface = MotionInterface()
iface.start(wait=False)
try:
if not iface.wait_for_ready(console=True, sensors=0, timeout=15):
print("console not connected")
return 2
console = iface.console
try:
console.telemetry.stop()
except Exception:
pass
print(f"console firmware {console.get_version()}")
try:
text = read_raw_config(console)
except Exception as e:
print(f"cannot read the stored config: {e}")
return 2

print(f"stored config: {len(text.encode('utf-8'))} bytes (limit {MOTION_CFG_JSON_MAX_LEN})")
problems = []
why = strict_json_problem(text)
if why:
problems.append(f"not strict JSON: {why}")
else:
data = json.loads(text)
tec = data.get("TEC_TRIP")
if tec is None:
problems.append("TEC_TRIP missing: the TEC thermal trip would not be armed")
elif isinstance(tec, bool) or not isinstance(tec, (int, float)) or not math.isfinite(tec) or tec <= 0:
problems.append(f"TEC_TRIP {tec!r} is not a positive number")
if problems:
print("FAIL")
for p in problems:
print(f" - {p}")
print("Fix the config and rewrite it (write_config / write_config_json) before "
"updating this console's firmware.")
return 1
print("OK: strict JSON object with a TEC_TRIP threshold")
return 0
finally:
iface.stop()


if __name__ == "__main__":
sys.exit(main())
78 changes: 78 additions & 0 deletions tests/test_motion_config_strict.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
"""MotionConfig writes only what the console stores: a strict JSON object.

Console firmware with the strict-config check (console-fw fix/jsmn-strict)
refuses a config that is not strict JSON or does not fit its page. The SDK
raises before sending instead (no NaN / Infinity from json.dumps), and
strict_json_problem() explains a stored config that would not pass. No
hardware needed.
"""

import json

import pytest

from omotion.MotionConfig import MOTION_CFG_JSON_MAX_LEN, MotionConfig, strict_json_problem

pytestmark = pytest.mark.unit


def _cfg(data):
m = MotionConfig()
m.json_data = data
return m


def test_valid_config_round_trips_compact():
data = {"TEC_TRIP": 40, "EE_GAIN": 1.86, "cal": {"c_min": [0.1, 0.2]}, "flag": True, "x": None}
wire = _cfg(data).to_wire_bytes()
text = wire[16:].decode()
assert json.loads(text) == data
assert strict_json_problem(text) is None
assert MotionConfig.from_wire_bytes(wire).json_data == data


@pytest.mark.parametrize(
("data", "path"),
[
({"TEC_TRIP": float("nan")}, "TEC_TRIP"),
({"a": float("inf")}, "a"),
({"a": {"b": [1.0, float("-inf")]}}, "a.b[1]"),
],
)
def test_non_finite_values_raise_naming_the_key(data, path):
with pytest.raises(ValueError, match=path.replace("[", r"\[").replace("]", r"\]")):
_cfg(data).to_wire_bytes()


def test_oversized_config_raises():
with pytest.raises(ValueError, match=str(MOTION_CFG_JSON_MAX_LEN)):
_cfg({"pad": "x" * MOTION_CFG_JSON_MAX_LEN}).to_wire_bytes()


def test_largest_config_that_fits():
filler = MOTION_CFG_JSON_MAX_LEN - len('{"pad":""}')
wire = _cfg({"pad": "x" * filler}).to_wire_bytes()
assert len(wire) - 16 == MOTION_CFG_JSON_MAX_LEN


@pytest.mark.parametrize(
("text", "fragment"),
[
('{"a":NaN}', "NaN"),
('{"a":Infinity}', "Infinity"),
('{"a":-Infinity}', "-Infinity"),
('{"a":1e999}', "out of range"),
('{"a":bare}', "Expecting value"),
('{"a":1', "Expecting"),
("[1,2]", "not a JSON object"),
("{" + '"p":"' + "x" * MOTION_CFG_JSON_MAX_LEN + '"}', "more than"),
],
)
def test_strict_json_problem_explains_rejections(text, fragment):
why = strict_json_problem(text)
assert why is not None and fragment in why


def test_strict_json_problem_accepts_strict_objects():
assert strict_json_problem('{"TEC_TRIP":40,"OPT_THRESH":7143,"EE_GAIN":1.86}') is None
assert strict_json_problem("{}") is None
Loading