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

more serial improvements

Browse Source
This commit is contained in:
Aiden
2026-05-25 16:32:58 +10:00
parent 6ceed81765
commit 56829b6e0b
10 changed files with 16843 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

294
h8536/serial_pseudocode.py
View File

@@ -40,6 +40,7 @@ def generate_serial_pseudocode(
lines.extend(_board_comment_lines(payload))
if opts.include_manual:
lines.extend(_manual_reference_lines(payload))
lines.extend(_sci_link_lines(payload))
lines.extend(_declarations(tx_candidate, rx_candidate))
if opts.include_semantics:
lines.extend(_semantics_lines(serial_semantics, opts))
@@ -196,6 +197,72 @@ def _manual_reference_lines(payload: JsonObject) -> list[str]:
return lines
def _sci_link_lines(payload: JsonObject) -> list[str]:
config = _first_sci1_configuration(payload)
dtc_note = _dtc_sci_note(payload)
if not config and not dtc_note:
return []
lines = ["/* SCI1 link layer:"]
if config:
mode_summary = str(config.get("mode_summary") or config.get("mode") or "SCI mode")
char_format = "8E1" if "8-bit even parity 1 stop" in mode_summary else mode_summary
smr = config.get("smr_hex") or _optional_hex(config.get("smr"), width=2)
brr = config.get("brr_hex") or _optional_hex(config.get("brr"), width=2)
scr = config.get("scr_hex") or _optional_hex(config.get("scr"), width=2)
clock_source = config.get("clock_source") or "clock source unknown"
lines.append(
" * - "
+ _comment_text(
f"{char_format} SCI characters carry the 6 protocol bytes; "
f"{mode_summary}, clock {clock_source}, SMR={smr}, BRR={brr}, SCR={scr}",
),
)
baud = config.get("baud_bps")
if isinstance(baud, (int, float)):
lines.append(f" * - candidate baud: {_comment_text(str(baud))} bps")
else:
lines.append(" * - baud is clock-dependent; pass --clock-hz on the decompiler for bps.")
if dtc_note:
lines.append(f" * - {_comment_text(dtc_note)}")
lines.append(" */")
lines.append("")
return lines
def _first_sci1_configuration(payload: JsonObject) -> JsonObject | None:
sci = payload.get("sci")
if not isinstance(sci, dict):
return None
channels = sci.get("channels")
if not isinstance(channels, dict):
return None
sci1 = channels.get("SCI1")
if not isinstance(sci1, dict):
return None
configurations = sci1.get("configurations")
if not isinstance(configurations, list):
return None
for item in configurations:
if isinstance(item, dict):
return item
return None
def _dtc_sci_note(payload: JsonObject) -> str:
vectors = payload.get("dtc_vectors")
if not isinstance(vectors, list):
return ""
sources = {
str(item.get("source", "")).lower()
for item in vectors
if isinstance(item, dict)
}
if "sci1_rxi" in sources or "sci1_txi" in sources:
return "SCI1 DTC vector entries are present; review DTC metadata before treating RX/TX as CPU-only."
return "No SCI1 RXI/TXI DTC vector entries are present in JSON; RX/TX are modeled as CPU ISR paths."
def _declarations(tx_candidate: JsonObject | None, rx_candidate: JsonObject | None) -> list[str]:
candidate = tx_candidate or rx_candidate or {}
channel = str(candidate.get("channel") or "SCI1")
@@ -250,6 +317,10 @@ def _declarations(tx_candidate: JsonObject | None, rx_candidate: JsonObject | No
timeout = _int_field(rx_candidate, "interbyte_timeout_address", 0xF9C1)
complete = _int_field(rx_candidate, "complete_timer_address", 0xF9C5)
length = _int_field(rx_candidate, "frame_length", 6)
error_handling = rx_candidate.get("rx_error_handling")
error_latch = 0xFAA4
if isinstance(error_handling, dict):
error_latch = _int_field(error_handling, "error_latch_address", 0xFAA4)
lines.extend(
[
f"#define RX_FRAME_LENGTH {length}u",
@@ -258,6 +329,8 @@ def _declarations(tx_candidate: JsonObject | None, rx_candidate: JsonObject | No
f"#define RX_INDEX MEM8[{_c_hex(rx_index)}]",
f"#define RX_INTERBYTE_TIMEOUT MEM8[{_c_hex(timeout)}]",
f"#define RX_COMPLETE_TIMER MEM8[{_c_hex(complete)}]",
f"#define RX_ERROR_LATCH MEM8[{_c_hex(error_latch)}]",
"#define RX_ERROR_LATCH_PHYSICAL_ERROR 0x80u",
"",
],
)
@@ -326,6 +399,11 @@ def _semantics_lines(
handler = command.get("handler_start_hex") or "multiple"
responses = ", ".join(str(item) for item in command.get("response_candidates", [])) or "none"
lines.append(f" * - {value} {name}: {summary}; handler {handler}; responses {responses}")
lines.extend(_command_effect_comment_lines(protocol.get("command_effects"), opts, prefix=" * "))
lines.extend(_response_schema_comment_lines(_schema_list(protocol), opts, prefix=" * "))
lines.extend(_table_map_comment_lines(_table_map_list(protocol), opts, prefix=" * "))
lines.extend(_state_variable_comment_lines(protocol.get("state_variable_candidates"), opts, prefix=" * "))
lines.extend(_retry_error_comment_lines(protocol.get("retry_error_model"), opts, prefix=" * "))
lines.append(" */")
lines.append("")
@@ -376,6 +454,8 @@ def _semantics_lines(
evidence = _hex_join(command.get("evidence_addresses_hex"))
lines.append(f" case 0x{value:02X}u:")
lines.append(f" /* {name}: {summary}")
for effect_line in _command_effect_switch_lines(command):
lines.append(f" * {effect_line}")
if opts.include_evidence and evidence:
lines.append(f" * evidence: {evidence}")
lines.append(" */")
@@ -394,6 +474,203 @@ def _semantics_lines(
return lines
def _command_effect_comment_lines(
value: object,
opts: SerialPseudocodeOptions,
*,
prefix: str,
) -> list[str]:
effects = [item for item in _object_list(value) if item.get("effects") or item.get("summary")]
if not effects:
return []
lines = [f"{prefix}command effects:"]
for item in effects:
command = item.get("command_value_hex") or _command_hex(item.get("command_value"))
name = item.get("name_candidate") or "unknown_command"
summary = _comment_text(str(item.get("summary") or "candidate effects"))
lines.append(f"{prefix}- {command} {name}: {summary}")
for effect in _object_list(item.get("effects"))[:3]:
effect_text = _effect_summary(effect)
if effect_text:
lines.append(f"{prefix} effect: {effect_text}")
evidence = _hex_join(item.get("evidence_addresses_hex"))
if opts.include_evidence and evidence:
lines.append(f"{prefix} evidence: {evidence}")
return lines
def _response_schema_comment_lines(
schemas: list[JsonObject],
opts: SerialPseudocodeOptions,
*,
prefix: str,
) -> list[str]:
if not schemas:
return []
lines = [f"{prefix}response schemas:"]
for schema in schemas[:4]:
response_id = schema.get("response_id") or schema.get("call_address_hex") or "candidate_response"
byte_text = _response_schema_summary(schema)
lines.append(f"{prefix}- {response_id}: {byte_text}")
evidence = _hex_join(schema.get("evidence_addresses_hex"))
if opts.include_evidence and evidence:
lines.append(f"{prefix} evidence: {evidence}")
if len(schemas) > 4:
lines.append(f"{prefix}- ... {len(schemas) - 4} more candidate response schemas")
return lines
def _table_map_comment_lines(
tables: list[JsonObject],
opts: SerialPseudocodeOptions,
*,
prefix: str,
) -> list[str]:
if not tables:
return []
lines = [f"{prefix}table map candidates:"]
for table in tables[:6]:
name = table.get("name_candidate") or "unnamed_table_candidate"
address = table.get("logical_base_address_hex") or table.get("address_hex") or "?"
element = table.get("element_candidate")
accesses = ", ".join(str(item) for item in table.get("observed_accesses", []) if item) or "access?"
detail = f"{name} at {address}"
if element:
detail += f" ({element})"
lines.append(f"{prefix}- {_comment_text(detail)}; observed {accesses}")
evidence = _hex_join(table.get("evidence_addresses_hex"))
if opts.include_evidence and evidence:
lines.append(f"{prefix} evidence: {evidence}")
if len(tables) > 6:
lines.append(f"{prefix}- ... {len(tables) - 6} more table candidates")
return lines
def _state_variable_comment_lines(
value: object,
opts: SerialPseudocodeOptions,
*,
prefix: str,
) -> list[str]:
states = sorted(
_object_list(value),
key=lambda item: item.get("address") if isinstance(item.get("address"), int) else 0x10000,
)
if not states:
return []
lines = [f"{prefix}state variable candidates:"]
for state in states[:6]:
name = state.get("name_candidate") or "unnamed_state_candidate"
address = state.get("address_hex") or _command_hex(state.get("address"))
reads = state.get("read_count", "?")
writes = state.get("write_count", "?")
bits = ", ".join(str(item) for item in state.get("bit_candidates", []))
suffix = f"; bits {bits}" if bits else ""
lines.append(f"{prefix}- {_comment_text(str(name))} {address}: reads {reads}, writes {writes}{suffix}")
evidence = _hex_join(state.get("evidence_addresses_hex"))
if opts.include_evidence and evidence:
lines.append(f"{prefix} evidence: {evidence}")
if len(states) > 6:
lines.append(f"{prefix}- ... {len(states) - 6} more state-variable candidates")
return lines
def _retry_error_comment_lines(
value: object,
opts: SerialPseudocodeOptions,
*,
prefix: str,
) -> list[str]:
if not isinstance(value, dict):
return []
checksum = value.get("checksum_failure_path")
retry = value.get("retry_path")
command_07 = value.get("command_0x07_path")
lines = [f"{prefix}retry/error model candidate:"]
if isinstance(checksum, dict):
condition = _comment_text(str(checksum.get("condition_candidate") or "checksum failure"))
target = checksum.get("error_target") or checksum.get("error_target_address_hex") or "error target"
lines.append(f"{prefix}- checksum path: {condition} -> {target}")
if isinstance(retry, dict):
counter = retry.get("counter_address_hex") or "counter?"
threshold = retry.get("threshold_candidate", "?")
summary = _comment_text(str(retry.get("summary") or "candidate retry path"))
lines.append(f"{prefix}- retry path: counter {counter}, threshold {threshold}; {summary}")
if isinstance(command_07, dict):
summary = _comment_text(str(command_07.get("summary") or "candidate command 0x07 path"))
lines.append(f"{prefix}- command 0x07 path: {summary}")
evidence = _hex_join(value.get("evidence_addresses_hex"))
if opts.include_evidence and evidence:
lines.append(f"{prefix}- evidence: {evidence}")
return lines
def _command_effect_switch_lines(command: JsonObject) -> list[str]:
effects = _object_list(command.get("effects"))[:3]
lines = []
for effect in effects:
text = _effect_summary(effect)
if text:
lines.append(f"candidate effect: {text}")
return lines
def _effect_summary(effect: JsonObject) -> str:
kind = str(effect.get("kind") or "effect_candidate")
parts = [kind]
target = effect.get("target_candidate") or effect.get("destination_candidate")
source = effect.get("source_candidate")
operation = effect.get("operation_candidate")
table = effect.get("table_base_hex")
state = effect.get("state_address_hex")
if target:
parts.append(f"target {target}")
if source:
parts.append(f"source {source}")
if operation:
parts.append(str(operation))
if table:
parts.append(f"table {table}")
if state:
parts.append(f"state {state}")
return _comment_text("; ".join(parts))
def _schema_list(protocol: JsonObject) -> list[JsonObject]:
schemas = _object_list(protocol.get("response_schemas"))
if schemas:
return schemas
return _object_list(protocol.get("response_schema"))
def _table_map_list(protocol: JsonObject) -> list[JsonObject]:
tables = _object_list(protocol.get("logical_table_map_candidates"))
if tables:
return tables
return _object_list(protocol.get("table_map_candidates"))
def _response_schema_summary(schema: JsonObject) -> str:
parts = []
for item in _object_list(schema.get("bytes")):
label = item.get("byte") or f"byte{item.get('offset', '?')}"
source = item.get("source_expression") or item.get("source_kind") or "unknown"
parts.append(f"{label}={source}")
return _comment_text("; ".join(parts) if parts else "candidate byte sources unknown")
def _object_list(value: object) -> list[JsonObject]:
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, dict)]
def _command_hex(value: object) -> str:
if isinstance(value, int):
return f"0x{value:02X}"
return "?"
def _tx_functions(candidate: JsonObject, opts: SerialPseudocodeOptions) -> list[str]:
length = _int_field(candidate, "frame_length", 6)
seed = _int_field(candidate, "checksum_seed", 0x5A)
@@ -511,8 +788,9 @@ def _rx_functions(candidate: JsonObject, opts: SerialPseudocodeOptions) -> list[
"",
"void sci1_rx_error_candidate_isr(void)",
"{",
" RX_ERROR_LATCH |= RX_ERROR_LATCH_PHYSICAL_ERROR;",
" SCI1_SSR &= (u8)~(SCI_SSR_ORER | SCI_SSR_FER | SCI_SSR_PER);",
" RX_INDEX = 0u;",
" sci1_rx_byte_received_candidate_isr();",
"}",
"",
],
@@ -532,6 +810,14 @@ def _candidate_comment_block(
formula = str(candidate.get("checksum_formula") or "").strip()
if formula:
lines.append(f" * checksum formula: {_comment_text(formula)}")
error_handling = candidate.get("rx_error_handling")
if isinstance(error_handling, dict):
summary = str(error_handling.get("summary") or "").strip()
if summary:
lines.append(f" * RX error handling: {_comment_text(summary)}")
caveat = str(error_handling.get("manual_caveat") or "").strip()
if caveat:
lines.append(f" * RX error caveat: {_comment_text(caveat)}")
if opts.include_evidence:
evidence = candidate.get("evidence_addresses_hex")
if isinstance(evidence, dict):
@@ -596,6 +882,12 @@ def _c_hex(value: int, *, width: int = 4) -> str:
return f"0x{value & 0xFFFF:0{width}X}u"
def _optional_hex(value: object, *, width: int = 4) -> str:
if isinstance(value, int):
return f"H'{value & 0xFFFF:0{width}X}"
return "unknown"
def _h(value: int) -> str:
return f"H'{value & 0xFFFF:04X}"