aiden/h8-536-decoder
1
0
Fork 0
Code Activity

RX-tx understanding

Browse Source
This commit is contained in:
Aiden
2026-05-26 10:48:39 +10:00
parent d1d924c408
commit 421c9f4567
13 changed files with 3968 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
tests/test_emulator_state_search.py Normal file
View File

@@ -0,0 +1,120 @@
import io
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from h8536.emulator.state_search import (
SearchCase,
SearchResult,
StatePatch,
build_cases,
classify_display,
main,
parse_address,
parse_matrix_patch,
parse_single_patch,
target_matches,
)
class EmulatorStateSearchTest(unittest.TestCase):
def test_parse_patch_accepts_hex_address_and_value(self):
patch = parse_single_patch("F730=0x41", size=1)
self.assertEqual(patch, StatePatch(1, 0xF730, 0x41, "user"))
self.assertEqual(patch.label(), "byte:H'F730=0x41")
def test_parse_matrix_patch_expands_values(self):
patches = parse_matrix_patch("E000=0x4080,0x8080", size=2)
self.assertEqual(patches, [
StatePatch(2, 0xE000, 0x4080, "user"),
StatePatch(2, 0xE000, 0x8080, "user"),
])
def test_parse_address_accepts_h_quote(self):
self.assertEqual(parse_address("H'F970"), 0xF970)
def test_connect_queue_preset_builds_small_rom_driven_matrix(self):
parser = __import__("h8536.emulator.state_search", fromlist=["build_arg_parser"]).build_arg_parser()
args = parser.parse_args(["--preset", "connect-queue"])
cases = build_cases(args)
self.assertEqual(len(cases), 25)
first = cases[0]
self.assertEqual(first.pc, 0x2806)
self.assertIn(StatePatch(2, 0xF970, 0x0000, "preset"), first.patches)
self.assertIn(StatePatch(2, 0xE000, 0x0000, "preset"), first.patches)
def test_custom_matrix_combines_fixed_and_matrix_patches(self):
parser = __import__("h8536.emulator.state_search", fromlist=["build_arg_parser"]).build_arg_parser()
args = parser.parse_args([
"--preset",
"custom",
"--pc",
"0x2CB9",
"--byte",
"F730=0",
"--matrix-word",
"E000=0x4080,0x8080",
])
cases = build_cases(args)
self.assertEqual(len(cases), 2)
self.assertEqual(cases[0], SearchCase((StatePatch(1, 0xF730, 0, "user"), StatePatch(2, 0xE000, 0x4080, "user")), 0x2CB9))
def test_classify_display(self):
self.assertEqual(classify_display(" CONNECT: OK | "), "ok")
self.assertEqual(classify_display(" CONNECT:DXC-637 | "), "dxc")
self.assertEqual(classify_display(" CONNECT:NOT ACT | "), "not-act")
def test_target_matching(self):
self.assertTrue(target_matches("ok", "ok"))
self.assertTrue(target_matches("dxc", "any-connect"))
self.assertTrue(target_matches("ok", "changed"))
self.assertFalse(target_matches("not-act", "changed"))
def test_cli_dry_run_lists_cases(self):
stdout = io.StringIO()
rc = main(["--dry-run", "--preset", "connect-branch", "--limit", "2"], stdout=stdout)
self.assertEqual(rc, 0)
output = stdout.getvalue()
self.assertIn("preset=connect-branch cases=2", output)
self.assertIn("case[0] pc=H'2CB9", output)
def test_cli_json_output_uses_results_from_run_search(self):
fake_result = SearchResult(
case_index=0,
patches=(StatePatch(2, 0xE000, 0x8080, "preset"),),
pc=0x2CB9,
steps=10,
stopped_reason="stop_pc",
final_pc=0xFFFF,
display=" CONNECT: OK | ",
line0=" CONNECT: OK ",
outcome="ok",
f730=0x81,
e000=0x8080,
f9b4=0,
f9b9=0,
)
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "out.json"
stdout = io.StringIO()
with patch("h8536.emulator.state_search.run_search", return_value=[fake_result]):
rc = main(["--preset", "custom", "--word", "E000=0x8080", "--json-out", str(path)], stdout=stdout)
payload = json.loads(path.read_text(encoding="utf-8"))
self.assertEqual(rc, 0)
self.assertIn("hits=1", stdout.getvalue())
self.assertEqual(payload["hits"][0]["outcome"], "ok")
if __name__ == "__main__":
unittest.main()

93
tests/test_rx_branch_trace.py Normal file
View File

@@ -0,0 +1,93 @@
import io
import json
import tempfile
import unittest
from pathlib import Path
from h8536.rx_branch_trace import analyze_rx_branch_trace, format_text_report, main
def ins(address: int, text: str | None = None) -> dict[str, object]:
return {
"address": address,
"text": text or f"INS_{address:04X}",
"mnemonic": (text or "NOP").split()[0],
"operands": "",
"kind": "normal",
"targets": [],
"references": [],
}
def fixture_payload() -> dict[str, object]:
addresses = {
0x3FEF, 0x3FF3, 0x3FF5, 0x3FF9, 0x3FFD, 0x4001, 0x4003, 0x4007,
0xBB57, 0xBB5B, 0xBB5F, 0xBB63, 0xBB67, 0xBB6D, 0xBB71, 0xBB75,
0xBB77, 0xBB7D, 0xBB82, 0xBB84, 0xBB88, 0xBB8A, 0xBB90, 0xBB96,
0xBB9A, 0xBB9C, 0xBB9E, 0xBBA3, 0xBBAB, 0xBBB0, 0xBBB3, 0xBBCB,
0xBBCF, 0xBBD3, 0xBBD6, 0xBBD8, 0xBBDC, 0xBBE0, 0xBBE4, 0xBBE8,
0xBBEC, 0xBBF0, 0xBBF3, 0xBBF7, 0xBBFD, 0xBC01, 0xBC08, 0xBC0C,
0xBC0F, 0xBC13, 0xBC15, 0xBC19, 0xBC1D, 0xBC20, 0xBC24, 0xBC29,
0xBC2E, 0xBC33, 0xBC37, 0xBC3A, 0xBC3C, 0xBC3E, 0xBC42, 0xBC45,
0xBC4A, 0xBC4F, 0xBC54, 0xBC5C, 0xBC60, 0xBC63, 0xBC67, 0xBC69,
0xBC75, 0xBC79, 0xBC82, 0xBC86, 0xBCB0, 0xBCCD, 0xBCD0, 0xBCD7,
0xBCE0, 0xBCE8, 0xBCEC, 0xBCF0, 0xBCF6, 0xBCFA, 0xBCFD, 0xBD04,
0xBD08, 0xBD0B, 0xBD0E, 0xBD1A, 0xBD1E, 0xBD22, 0xBD26, 0xBD35,
0xBD64, 0xBD67, 0xBD6D, 0xBD75, 0xBD79, 0xBD80, 0xBD85, 0xBD94,
0xBD9A, 0xBDB5, 0xBDBF, 0xBDC2, 0xBDC8, 0xBDD0, 0xBDD4, 0xBDDB,
0xBDE5, 0xBDE9, 0xBDED, 0xBDF3, 0xBDFB, 0xBDFF, 0xBE05, 0xBE09,
0xBE0D, 0xBE11, 0xBE15, 0xBE19, 0xBE1D, 0xBE22, 0xBE27, 0xBE29,
0xBE2D, 0xBE31, 0xBE33, 0xBE37, 0xBE3C, 0xBE3E, 0xBE43, 0xBE47,
0xBE4D, 0xBE52, 0xBE5A, 0xBE62, 0xBE6A, 0xBE70, 0xBE78, 0xBE80,
0xBE82, 0xBE84, 0xBE88, 0xBE91, 0xBE95, 0xBE99, 0xBE9D, 0xBE9E,
0xBEA5, 0xBEA9, 0xBEAD, 0xBEAF, 0xBEB5, 0xBEBB, 0xBEBF, 0xBEC1,
0xBEC5, 0xBECB, 0xBED1, 0xBED5, 0xBEE4,
}
return {"instructions": [ins(address) for address in sorted(addresses)]}
class RxBranchTraceTest(unittest.TestCase):
def test_analyzes_dispatch_split_and_commands(self):
analysis = analyze_rx_branch_trace(fixture_payload())
self.assertEqual(analysis["summary"]["confidence"], "high")
self.assertEqual(analysis["frame_model"]["checksum_seed"], 0x5A)
self.assertTrue(analysis["stages"][2]["present"])
self.assertIn("FAA2 != 0", analysis["stages"][2]["summary"])
commands = {command["command"]: command for command in analysis["commands"]}
self.assertIn("continuation path only", commands[0x04]["availability"])
self.assertIn("selector zero is special", "\n".join(commands[0x04]["side_effects"]))
self.assertIn("selectors 0x006C", "\n".join(commands[0x05]["side_effects"]))
self.assertIn("previous finalized TX frame", commands[0x07]["summary"])
def test_text_report_mentions_bench_implications(self):
text = format_text_report(analyze_rx_branch_trace(fixture_payload()))
self.assertIn("H8/536 SCI1 RX Branch Trace", text)
self.assertIn("cmd 0x04 continuation_set_value_candidate", text)
self.assertIn("standalone command 4 frame from idle should not hit BD0E", text)
self.assertIn("Command 5 is not a generic always-live ACK", text)
self.assertIn("Selector Decode", text)
self.assertIn("TXI/RXI race and continuation collapse", text)
self.assertIn("RX-to-TX Feedback Loops", text)
def test_cli_writes_json_output(self):
with tempfile.TemporaryDirectory() as tmp:
input_path = Path(tmp) / "rom.json"
output_path = Path(tmp) / "rx.json"
input_path.write_text(json.dumps(fixture_payload()), encoding="utf-8")
stdout = io.StringIO()
rc = main(["--json", "--out", str(output_path), str(input_path)], stdout=stdout)
self.assertEqual(rc, 0)
self.assertIn("wrote", stdout.getvalue())
payload = json.loads(output_path.read_text(encoding="utf-8"))
self.assertEqual(payload["kind"], "rx_branch_trace")
self.assertIn("downstream_traces", payload)
self.assertIn("feedback_loops", payload)
if __name__ == "__main__":
unittest.main()

114
tests/test_state_map_runner.py Normal file
View File

@@ -0,0 +1,114 @@
import io
import json
import tempfile
import unittest
from pathlib import Path
from h8536.state_map_runner import (
CONNECT_FORCE_PRESETS,
StateMapEvent,
analyze_events,
format_analysis_report,
main,
parse_bench_log,
)
class StateMapRunnerTest(unittest.TestCase):
def test_force_presets_match_expected_selector_zero_words(self):
self.assertEqual(CONNECT_FORCE_PRESETS["dxc"][0], bytes.fromhex("04000040001E"))
self.assertEqual(CONNECT_FORCE_PRESETS["dxc"][1], 0x4080)
self.assertEqual(CONNECT_FORCE_PRESETS["ok"][0], bytes.fromhex("0400008000DE"))
self.assertEqual(CONNECT_FORCE_PRESETS["ok"][1], 0x8080)
self.assertEqual(CONNECT_FORCE_PRESETS["both"][0], bytes.fromhex("040000C0009E"))
self.assertEqual(CONNECT_FORCE_PRESETS["both"][1], 0xC080)
def test_analyzes_successful_selector_zero_readback(self):
events = [
StateMapEvent("rx", bytes.fromhex("0780C060205D"), timestamp_ms=1000),
StateMapEvent("tx", bytes.fromhex("0400008000DE"), timestamp_ms=1006),
StateMapEvent("tx", bytes.fromhex("01000000005B"), timestamp_ms=1060),
StateMapEvent("rx", bytes.fromhex("04001280804C"), timestamp_ms=1070),
]
analysis = analyze_events(events, expected_word=0x8080)
self.assertEqual(analysis["outcome"]["name"], "selector_zero_retained")
self.assertEqual(analysis["direct_readbacks"][0]["value"], 0x8080)
self.assertTrue(analysis["direct_readbacks"][0]["matches_expected"])
def test_warns_when_command1_readback_is_between_trigger_and_force(self):
events = [
StateMapEvent("rx", bytes.fromhex("07804040A07D"), timestamp_ms=1000),
StateMapEvent("tx", bytes.fromhex("01000000005B"), timestamp_ms=1005),
StateMapEvent("tx", bytes.fromhex("04000040001E"), timestamp_ms=1010),
]
analysis = analyze_events(events, expected_word=0x4080)
self.assertIn("command1_readback_between_trigger_and_force_can_spend_token", analysis["warnings"])
self.assertEqual(analysis["outcome"]["name"], "force_not_proven")
def test_parse_bench_log_uses_detect_lines_for_rx_and_tx_chunks_for_host_frames(self):
log = (
"00:00:01.000 RX 006 bytes 07 80 C0 60 20 5D\n"
"00:00:01.000 DETECT visible_C0_6020_family_candidate 07 80 C0 60 20 5D\n"
"00:00:01.006 TX 006 bytes 04 00 00 80 00 DE\n"
)
events = parse_bench_log(log)
self.assertEqual([(event.direction, event.frame.hex()) for event in events], [
("rx", "0780c060205d"),
("tx", "0400008000de"),
])
def test_format_report_mentions_outcome_and_readback_value(self):
analysis = analyze_events(
[
StateMapEvent("rx", bytes.fromhex("0780C060205D"), timestamp_ms=1000),
StateMapEvent("tx", bytes.fromhex("0400008000DE"), timestamp_ms=1006),
StateMapEvent("rx", bytes.fromhex("04001280804C"), timestamp_ms=1070),
],
expected_word=0x8080,
)
report = format_analysis_report(analysis)
self.assertIn("outcome=selector_zero_retained", report)
self.assertIn("value=0x8080 expected", report)
def test_cli_dry_run_prints_state_map_sequence(self):
stdout = io.StringIO()
rc = main(["--dry-run", "--preset", "dxc", "--prime-frame", "01 80 40 40 30 EB"], stdout=stdout)
output = stdout.getvalue()
self.assertEqual(rc, 0)
self.assertIn("PT2 state-map proof runner", output)
self.assertIn("force=04 00 00 40 00 1E", output)
self.assertIn("expected_e0000=0x4080", output)
self.assertIn("prime=01 80 40 40 30 EB", output)
def test_cli_analyze_log_writes_json(self):
log = (
"00:00:01.000 DETECT visible_C0_6020_family_candidate 07 80 C0 60 20 5D\n"
"00:00:01.006 TX 006 bytes 04 00 00 80 00 DE\n"
"00:00:01.070 DETECT table_readback_candidate 04 00 12 80 80 4C\n"
)
with tempfile.TemporaryDirectory() as tmpdir:
log_path = Path(tmpdir) / "capture.txt"
json_path = Path(tmpdir) / "analysis.json"
log_path.write_text(log, encoding="utf-8")
stdout = io.StringIO()
rc = main(["--analyze-log", str(log_path), "--preset", "ok", "--json-out", str(json_path)], stdout=stdout)
payload = json.loads(json_path.read_text(encoding="utf-8"))
self.assertEqual(rc, 0)
self.assertIn("outcome=selector_zero_retained", stdout.getvalue())
self.assertEqual(payload["outcome"]["name"], "selector_zero_retained")
if __name__ == "__main__":
unittest.main()