#!

/usr/bin/env python3
"""
pmdi_reverse_engineering.py

Barrido sistemático y adaptativo para estudiar una interfaz serie

Dräger/Infinity/PMDI en banco de pruebas.

FASE 0
Mide actividad RX espontánea sin transmitir.

FASE 1
Recorre todos los comandos 0x00..0xFF con una estructura fija y el
checksum SUM8, que coincide con:
00 A5 02 00 77 1E
00 A5 02 00 50 F7

FASE 2
Toma los comandos mejor puntuados y ensaya varias estructuras y
familias de checksum.

OPCIONAL
--full-cartesian prueba todos los comandos × todas las estructuras ×
todos los checksums. Puede tardar bastante.

SALIDAS
raw_observations.csv
phase1_ranking.csv
phase2_summary.csv
response_clusters.txt
[Link]
run_config.json

Dependencia:
pip install pyserial

Uso recomendado:
python pmdi_reverse_engineering.py --port COM3 --baud 9600 --yes

Barrido completo:
python pmdi_reverse_engineering.py --port COM3 --baud 9600 \
--full-cartesian --repetitions 2 --yes

IMPORTANTE:
Utilizar solamente con el equipo fuera de servicio clínico, sin
paciente conectado y en un banco de pruebas autorizado.
"""

from __future__ import annotations

import argparse
import csv
import json
import math
import statistics
import sys
import time
from collections import Counter
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, Iterable, Optional

import serial

# ---------------------------------------------------------------------------
# Modelos de datos
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class Layout:
name: str
description: str
builder: Callable[[int], bytes]

@dataclass(frozen=True)
class ChecksumMethod:
name: str
description: str
calculator: Callable[[bytes], int]

@dataclass(frozen=True)
class Candidate:
phase: str
command: int
layout: str
checksum_methods: tuple[str, ...]
frame: bytes

@dataclass
class Observation:
phase: str
candidate_index: int
command: int
layout: str
checksum_methods: str
repetition: int
 tx_hex: str
tx_len: int
rx_hex: str
rx_len: int
first_byte_ms: Optional[float]
rx_window_ms: float
cts: str
dsr: str
ri: str
cd: str
classification: str
timestamp: str

# ---------------------------------------------------------------------------
# Utilidades
# ---------------------------------------------------------------------------

def hexstr(data: bytes) -> str:

return " ".join(f"{b:02X}" for b in data)

def parse_hex_bytes(text: str) -> bytes:

cleaned = [Link](",", " ").replace("0x", "").replace("0X", "")
return [Link](cleaned)

def safe_line_state(ser: [Link], attr: str) -> str:

try:
return str(bool(getattr(ser, attr)))
except (OSError, [Link]):
return "N/D"

def line_states(ser: [Link]) -> dict[str, str]:

return {
"cts": safe_line_state(ser, "cts"),
"dsr": safe_line_state(ser, "dsr"),
"ri": safe_line_state(ser, "ri"),
"cd": safe_line_state(ser, "cd"),
}

def classify_rx(tx: bytes, rx: bytes) -> str:

if not rx:
return "sin_respuesta"
if rx == tx:
return "eco_exacto"
if tx in rx:
return "contiene_tx"
 if rx in tx:
return "eco_parcial_posible"
if [Link](b"\x00\xA5"):
return "inicio_00_A5"
if [Link](b"\xA5"):
return "inicio_A5"
if [Link](b"\x00"):
return "inicio_00"
return "respuesta_otra"

def byte_edit_distance(a: bytes, b: bytes) -> int:

"""Distancia de Levenshtein sobre secuencias de bytes."""
if a == b:
return 0
if not a:
return len(b)
if not b:
return len(a)

previous = list(range(len(b) + 1))

for i, xa in enumerate(a, start=1):
current = [i]
for j, xb in enumerate(b, start=1):
insert_cost = current[j - 1] + 1
delete_cost = previous[j] + 1
replace_cost = previous[j - 1] + (xa != xb)
[Link](min(insert_cost, delete_cost, replace_cost))
previous = current
return previous[-1]

def normalized_distance(a: bytes, b: bytes) -> float:

denominator = max(len(a), len(b), 1)
return byte_edit_distance(a, b) / denominator

def mean_pairwise_similarity(items: list[bytes]) -> float:

nonempty = [x for x in items if x]
if len(nonempty) < 2:
return 1.0 if nonempty else 0.0

similarities: list[float] = []
for i in range(len(nonempty)):
for j in range(i + 1, len(nonempty)):
[Link](1.0 - normalized_distance(nonempty[i],
nonempty[j]))
return [Link](similarities) if similarities else 0.0
# ---------------------------------------------------------------------------
# Checksums
# ---------------------------------------------------------------------------

def sum8(data: bytes) -> int:

return sum(data) & 0xFF

def sum8_no_leading_zero(data: bytes) -> int:

if data and data[0] == 0x00:
data = data[1:]
return sum(data) & 0xFF

def sum8_no_a5(data: bytes) -> int:

removed = False
out = bytearray()
for value in data:
if value == 0xA5 and not removed:
removed = True
continue
[Link](value)
return sum(out) & 0xFF

def twos_complement_sum(data: bytes) -> int:

return (-sum(data)) & 0xFF

def ones_complement_sum(data: bytes) -> int:

return (~sum(data)) & 0xFF

def xor8(data: bytes) -> int:

value = 0
for byte in data:
value ^= byte
return value & 0xFF

def xor8_inverted(data: bytes) -> int:

return xor8(data) ^ 0xFF

def xor8_no_a5(data: bytes) -> int:

value = 0
removed = False
for byte in data:
if byte == 0xA5 and not removed:
removed = True
 continue
value ^= byte
return value & 0xFF

def crc8_msb(data: bytes, polynomial: int, init: int = 0, xorout: int = 0) -> int:
crc = init & 0xFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x80:
crc = ((crc << 1) ^ polynomial) & 0xFF
else:
crc = (crc << 1) & 0xFF
return crc ^ xorout

def crc8_lsb(data: bytes, polynomial: int, init: int = 0, xorout: int = 0) -> int:
crc = init & 0xFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x01:
crc = ((crc >> 1) ^ polynomial) & 0xFF
else:
crc = (crc >> 1) & 0xFF
return crc ^ xorout

CHECKSUMS: dict[str, ChecksumMethod] = {

"sum8": ChecksumMethod(
"sum8",
"Suma de todos los bytes modulo 256",
sum8,
),
"sum8_no_lead0": ChecksumMethod(
"sum8_no_lead0",
"SUM8 omitiendo un 00 inicial",
sum8_no_leading_zero,
),
"sum8_no_a5": ChecksumMethod(
"sum8_no_a5",
"SUM8 omitiendo la primera aparicion de A5",
sum8_no_a5,
),
"twos_sum": ChecksumMethod(
"twos_sum",
"Complemento a dos de la suma",
twos_complement_sum,
),
 "ones_sum": ChecksumMethod(
"ones_sum",
"Complemento a uno de la suma",
ones_complement_sum,
),
"xor8": ChecksumMethod(
"xor8",
"XOR de todos los bytes",
xor8,
),
"xor8_inv": ChecksumMethod(
"xor8_inv",
"XOR invertido",
xor8_inverted,
),
"xor8_no_a5": ChecksumMethod(
"xor8_no_a5",
"XOR omitiendo la primera aparicion de A5",
xor8_no_a5,
),
"crc8_atm": ChecksumMethod(
"crc8_atm",
"CRC-8/ATM, polinomio 0x07, init 0x00",
lambda data: crc8_msb(data, 0x07, 0x00, 0x00),
),
"crc8_sae_j1850": ChecksumMethod(
"crc8_sae_j1850",
"CRC-8/SAE-J1850, polinomio 0x1D, init/xorout 0xFF",
lambda data: crc8_msb(data, 0x1D, 0xFF, 0xFF),
),
"crc8_maxim": ChecksumMethod(
"crc8_maxim",
"CRC-8/MAXIM-DOW reflejado, polinomio reflejado 0x8C",
lambda data: crc8_lsb(data, 0x8C, 0x00, 0x00),
),
"crc8_darc": ChecksumMethod(
"crc8_darc",
"CRC-8/DARC reflejado, polinomio reflejado 0x9C",
lambda data: crc8_lsb(data, 0x9C, 0x00, 0x00),
),
}

# ---------------------------------------------------------------------------
# Estructuras de trama
# ---------------------------------------------------------------------------

LAYOUTS: dict[str, Layout] = {

"primary": Layout(
"primary",
 "00 A5 02 00 CMD CHK",
lambda cmd: bytes((0x00, 0xA5, 0x02, 0x00, cmd)),
),
"no00": Layout(
"no00",
"A5 02 00 CMD CHK",
lambda cmd: bytes((0xA5, 0x02, 0x00, cmd)),
),
"short": Layout(
"short",
"00 A5 01 CMD CHK",
lambda cmd: bytes((0x00, 0xA5, 0x01, cmd)),
),
"short_no00": Layout(
"short_no00",
"A5 01 CMD CHK",
lambda cmd: bytes((0xA5, 0x01, cmd)),
),
}

# ---------------------------------------------------------------------------
# Comunicación serie
# ---------------------------------------------------------------------------

def read_response(
ser: [Link],
total_window_s: float,
end_silence_s: float,
) -> tuple[bytes, Optional[float], float]:
start = time.perf_counter()
last_byte = start
first_byte_s: Optional[float] = None
received = bytearray()

while True:
now = time.perf_counter()
elapsed = now - start

if elapsed >= total_window_s:

break

available = ser.in_waiting
if available:
block = [Link](available)
if block:
if first_byte_s is None:
first_byte_s = time.perf_counter() - start
[Link](block)
last_byte = time.perf_counter()
 continue

if received and (now - last_byte) >= end_silence_s:

break

[Link](0.001)

return bytes(received), first_byte_s, time.perf_counter() - start

def listen_without_tx(
ser: [Link],
window_s: float,
end_silence_s: float,
) -> tuple[bytes, Optional[float], float]:
ser.reset_input_buffer()
[Link](0.02)
return read_response(ser, window_s, end_silence_s)

def transmit_candidate(
ser: [Link],
candidate: Candidate,
repetition: int,
candidate_index: int,
pre_tx_s: float,
rx_window_s: float,
end_silence_s: float,
) -> Observation:
ser.reset_input_buffer()
ser.reset_output_buffer()
[Link](pre_tx_s)

states = line_states(ser)
timestamp = [Link]().isoformat(timespec="milliseconds")

written = [Link]([Link])
[Link]()

if written != len([Link]):
raise [Link](
f"Se escribieron {written} de {len([Link])} bytes"
)

rx, first_byte_s, rx_elapsed_s = read_response(

ser,
total_window_s=rx_window_s,
end_silence_s=end_silence_s,
)
 return Observation(
phase=[Link],
candidate_index=candidate_index,
command=[Link],
layout=[Link],
checksum_methods="|".join(candidate.checksum_methods),
repetition=repetition,
tx_hex=hexstr([Link]),
tx_len=len([Link]),
rx_hex=hexstr(rx),
rx_len=len(rx),
first_byte_ms=None if first_byte_s is None else round(first_byte_s * 1000,
3),
rx_window_ms=round(rx_elapsed_s * 1000, 3),
cts=states["cts"],
dsr=states["dsr"],
ri=states["ri"],
cd=states["cd"],
classification=classify_rx([Link], rx),
timestamp=timestamp,
)

# ---------------------------------------------------------------------------
# Generación de candidatos
# ---------------------------------------------------------------------------

def make_candidates(
phase: str,
commands: Iterable[int],
layout_names: list[str],
checksum_names: list[str],
) -> list[Candidate]:
"""
Deduplica tramas idénticas producidas por diferentes checksums.
Conserva todos los nombres de métodos que generaron el mismo byte.
"""
candidates: list[Candidate] = []

for command in commands:

for layout_name in layout_names:
layout = LAYOUTS[layout_name]
body = [Link](command)

frames_to_methods: dict[bytes, list[str]] = {}

for checksum_name in checksum_names:
checksum = CHECKSUMS[checksum_name].calculator(body)
frame = body + bytes((checksum,))
frames_to_methods.setdefault(frame, []).append(checksum_name)
 for frame, methods in frames_to_methods.items():
[Link](
Candidate(
phase=phase,
command=command,
layout=layout_name,
checksum_methods=tuple(methods),
frame=frame,
)
)

return candidates

# ---------------------------------------------------------------------------
# Análisis
# ---------------------------------------------------------------------------

def observations_by_candidate(
observations: list[Observation],
) -> dict[int, list[Observation]]:
grouped: dict[int, list[Observation]] = {}
for obs in observations:
[Link](obs.candidate_index, []).append(obs)
return grouped

def decode_rx(obs: Observation) -> bytes:

return parse_hex_bytes(obs.rx_hex) if obs.rx_hex else b""

def summarize_candidate(
candidate: Candidate,
observations: list[Observation],
baseline_counter: Counter[bytes],
) -> dict[str, object]:
responses = [decode_rx(obs) for obs in observations]
nonempty = [rx for rx in responses if rx]
response_rate = len(nonempty) / max(len(responses), 1)

counter = Counter(nonempty)
modal_response, modal_count = (counter.most_common(1)[0] if counter else (b"",
0))
modal_ratio = modal_count / max(len(nonempty), 1)
fuzzy_consistency = mean_pairwise_similarity(nonempty)

latencies = [
obs.first_byte_ms
for obs in observations
if obs.first_byte_ms is not None
 ]
median_latency = [Link](latencies) if latencies else None
latency_stdev = [Link](latencies) if len(latencies) > 1 else 0.0

framing_bonus = 0.0
if modal_response.startswith(b"\x00\xA5"):
framing_bonus = 18.0
elif modal_response.startswith(b"\xA5"):
framing_bonus = 15.0
elif modal_response.startswith(b"\x00"):
framing_bonus = 4.0

baseline_penalty = 15.0 if modal_response and baseline_counter[modal_response]

else 0.0

length_bonus = min(len(modal_response), 8) * 1.5

latency_bonus = 0.0
if median_latency is not None:
if 1.0 <= median_latency <= 50.0:
latency_bonus += 8.0
if latency_stdev <= 2.0:
latency_bonus += 5.0

score = (
response_rate * 35.0
+ modal_ratio * 15.0
+ fuzzy_consistency * 20.0
+ framing_bonus
+ length_bonus
+ latency_bonus
- baseline_penalty
)

return {
"phase": [Link],
"command_dec": [Link],
"command_hex": f"0x{[Link]:02X}",
"layout": [Link],
"checksum_methods": "|".join(candidate.checksum_methods),
"tx_hex": hexstr([Link]),
"attempts": len(responses),
"responses": len(nonempty),
"response_rate": round(response_rate, 4),
"unique_responses": len(counter),
"modal_rx_hex": hexstr(modal_response),
"modal_count": modal_count,
"modal_ratio": round(modal_ratio, 4),
"fuzzy_consistency": round(fuzzy_consistency, 4),
"median_latency_ms": (
"" if median_latency is None else round(float(median_latency), 3)
 ),
"latency_stdev_ms": round(float(latency_stdev), 3),
"modal_rx_len": len(modal_response),
"seen_in_baseline": bool(modal_response and
baseline_counter[modal_response]),
"score": round(score, 3),
}

def greedy_clusters(
summaries: list[dict[str, object]],
threshold: float,
) -> list[dict[str, object]]:
"""
Agrupa respuestas modales por distancia de edición normalizada.
Solo considera respuestas no vacías.
"""
clusters: list[dict[str, object]] = []

ordered = sorted(
(row for row in summaries if row["modal_rx_hex"]),
key=lambda row: float(row["score"]),
reverse=True,
)

for row in ordered:

rx = parse_hex_bytes(str(row["modal_rx_hex"]))
assigned = False

for cluster in clusters:

prototype = cluster["prototype_bytes"]
if abs(len(rx) - len(prototype)) > 2:
continue
distance = normalized_distance(rx, prototype)
if distance <= threshold:
cluster["members"].append(row)
assigned = True
break

if not assigned:
[Link](
{
"prototype_bytes": rx,
"members": [row],
}
)

return clusters
# ---------------------------------------------------------------------------
# Escritura de resultados
# ---------------------------------------------------------------------------

RAW_FIELDS = [[Link] for field in Observation.__dataclass_fields__.values()]

def append_observation(writer: [Link], observation: Observation) -> None:

[Link](asdict(observation))

def write_summary_csv(path: Path, rows: list[dict[str, object]]) -> None:

if not rows:
path.write_text("", encoding="utf-8")
return

with [Link]("w", newline="", encoding="utf-8-sig") as file:

writer = [Link](file, fieldnames=list(rows[0].keys()))
[Link]()
[Link](rows)

def write_clusters(path: Path, clusters: list[dict[str, object]]) -> None:

with [Link]("w", encoding="utf-8") as file:
for index, cluster in enumerate(clusters, start=1):
prototype = cluster["prototype_bytes"]
members = cluster["members"]

[Link]("=" * 90 + "\n")
[Link](f"CLUSTER {index}\n")
[Link](f"Prototipo RX: {hexstr(prototype)}\n")
[Link](f"Cantidad: {len(members)}\n")
[Link]("=" * 90 + "\n")

for row in members[:100]:

[Link](
f"score={row['score']:>7} | "
f"cmd={row['command_hex']} | "
f"layout={row['layout']:<10} | "
f"checksum={row['checksum_methods']:<35} | "
f"TX={row['tx_hex']} | "
f"RX={row['modal_rx_hex']}\n"
)
[Link]("\n")

def write_report(
path: Path,
args: [Link],
baseline_samples: list[bytes],
 phase1_rows: list[dict[str, object]],
phase2_rows: list[dict[str, object]],
clusters: list[dict[str, object]],
) -> None:
baseline_nonempty = [sample for sample in baseline_samples if sample]

with [Link]("w", encoding="utf-8") as file:

[Link]("INFORME AUTOMATICO DE BARRIDO PMDI / INFINITY\n")
[Link]("=" * 90 + "\n\n")
[Link](f"Fecha: {[Link]().isoformat(timespec='seconds')}\n")
[Link](f"Puerto: {[Link]}\n")
[Link](
f"Serie: {[Link]} {[Link]}{[Link]}{[Link]}\n"
)
[Link](
f"Flujo: RTS/CTS={[Link]}, "
f"DSR/DTR={[Link]}, XON/XOFF={[Link]}\n"
)
[Link](f"Modo completo: {args.full_cartesian}\n\n")

[Link]("LINEA DE BASE SIN TX\n")

[Link]("-" * 90 + "\n")
[Link](f"Muestras: {len(baseline_samples)}\n")
[Link](f"Con bytes RX: {len(baseline_nonempty)}\n")
for rx, count in Counter(baseline_nonempty).most_common(20):
[Link](f"{count:>4}x {hexstr(rx)}\n")
[Link]("\n")

[Link]("MEJORES RESULTADOS DE FASE 1\n")

[Link]("-" * 90 + "\n")
for row in phase1_rows[:30]:
[Link](
f"score={row['score']:>7} | "
f"cmd={row['command_hex']} | "
f"TX={row['tx_hex']} | "
f"RX={row['modal_rx_hex']} | "
f"resp={row['response_rate']:.2f} | "
f"cons={row['fuzzy_consistency']:.2f} | "
f"lat={row['median_latency_ms']} ms\n"
)
[Link]("\n")

[Link]("MEJORES RESULTADOS DE FASE 2\n")

[Link]("-" * 90 + "\n")
for row in phase2_rows[:50]:
[Link](
f"score={row['score']:>7} | "
f"cmd={row['command_hex']} | "
f"layout={row['layout']:<10} | "
f"checksum={row['checksum_methods']:<30} | "
 f"TX={row['tx_hex']} | "
f"RX={row['modal_rx_hex']}\n"
)
[Link]("\n")

[Link]("CLUSTERS DE RESPUESTAS\n")
[Link]("-" * 90 + "\n")
for index, cluster in enumerate(clusters, start=1):
[Link](
f"Cluster {index:>3}: "
f"{len(cluster['members']):>4} miembros | "
f"prototipo={hexstr(cluster['prototype_bytes'])}\n"
)

# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def parity_arg(value: str) -> str:

mapping = {
"N": serial.PARITY_NONE,
"E": serial.PARITY_EVEN,
"O": serial.PARITY_ODD,
"M": serial.PARITY_MARK,
"S": serial.PARITY_SPACE,
}
try:
return mapping[[Link]()]
except KeyError as exc:
raise [Link]("Use N, E, O, M o S") from exc

def stopbits_arg(value: str) -> float:

mapping = {
"1": serial.STOPBITS_ONE,
"1.5": serial.STOPBITS_ONE_POINT_FIVE,
"2": serial.STOPBITS_TWO,
}
try:
return mapping[value]
except KeyError as exc:
raise [Link]("Use 1, 1.5 o 2") from exc

def csv_names(value: str, allowed: set[str], label: str) -> list[str]:

names = [[Link]() for item in [Link](",") if [Link]()]
invalid = [name for name in names if name not in allowed]
if invalid:
raise [Link](
 f"{label} desconocidos: {', '.join(invalid)}. "
f"Disponibles: {', '.join(sorted(allowed))}"
)
return names

def build_parser() -> [Link]:

parser = [Link](
description="Barrido sistemático/adaptativo PMDI por RS-232."
)

parser.add_argument("--port", default="COM3")
parser.add_argument("--baud", type=int, default=9600)
parser.add_argument("--bytesize", type=int, choices=(5, 6, 7, 8), default=8)
parser.add_argument("--parity", type=parity_arg, default=serial.PARITY_NONE)
parser.add_argument("--stopbits", type=stopbits_arg,
default=serial.STOPBITS_ONE)

parser.add_argument("--rtscts", action="store_true")
parser.add_argument("--dsrdtr", action="store_true")
parser.add_argument("--xonxoff", action="store_true")

parser.add_argument("--start-command", type=lambda x: int(x, 0), default=0x00)

parser.add_argument("--end-command", type=lambda x: int(x, 0), default=0xFF)

parser.add_argument(
"--phase1-repetitions",
type=int,
default=2,
help="Repeticiones por comando en fase 1.",
)
parser.add_argument(
"--repetitions",
type=int,
default=3,
help="Repeticiones por candidato en fase 2.",
)
parser.add_argument(
"--top-commands",
type=int,
default=24,
help="Comandos de fase 1 que avanzan a fase 2.",
)
parser.add_argument(
"--baseline-samples",
type=int,
default=20,
help="Escuchas sin TX al inicio.",
)
 parser.add_argument("--pre-tx", type=float, default=0.03)
parser.add_argument("--rx-window", type=float, default=0.18)
parser.add_argument("--end-silence", type=float, default=0.04)
parser.add_argument("--pause", type=float, default=0.08)

parser.add_argument(
"--phase2-layouts",
default="primary,no00,short,short_no00",
help=f"Estructuras separadas por coma: {','.join(LAYOUTS)}",
)
parser.add_argument(
"--phase2-checksums",
default=",".join(CHECKSUMS),
help=f"Checksums separados por coma: {','.join(CHECKSUMS)}",
)

parser.add_argument(
"--cluster-threshold",
type=float,
default=0.35,
help="Distancia normalizada máxima para agrupar respuestas.",
)
parser.add_argument(
"--full-cartesian",
action="store_true",
help="Prueba todos los comandos × layouts × checksums.",
)
parser.add_argument(
"--yes",
action="store_true",
help="Confirma automáticamente que el equipo está en banco.",
)
return parser

def validate_args(args: [Link]) -> None:

if not 0 <= args.start_command <= 255:
raise ValueError("--start-command debe estar entre 0 y 255")
if not 0 <= args.end_command <= 255:
raise ValueError("--end-command debe estar entre 0 y 255")
if args.start_command > args.end_command:
raise ValueError("--start-command no puede ser mayor que --end-command")
if args.phase1_repetitions < 1 or [Link] < 1:
raise ValueError("Las repeticiones deben ser mayores o iguales a 1")
if args.top_commands < 1:
raise ValueError("--top-commands debe ser al menos 1")
if args.rx_window <= 0 or args.end_silence <= 0:
raise ValueError("Las ventanas temporales deben ser positivas")
# ---------------------------------------------------------------------------
# Ejecución principal
# ---------------------------------------------------------------------------

def main() -> int:

parser = build_parser()
args = parser.parse_args()

try:
validate_args(args)
phase2_layout_names = csv_names(
args.phase2_layouts,
set(LAYOUTS),
"Layouts",
)
phase2_checksum_names = csv_names(
args.phase2_checksums,
set(CHECKSUMS),
"Checksums",
)
except (ValueError, [Link]) as exc:
[Link](str(exc))

if not [Link]:
print()
print("ADVERTENCIA DE BANCO")
print("-" * 72)
print("Este programa enviará múltiples comandos al puerto serie.")
print("Úselo solamente con el equipo fuera de servicio clínico,")
print("sin paciente conectado y con autorización técnica.")
answer = input('Escriba "BANCO" para continuar: ').strip()
if answer != "BANCO":
print("Cancelado.")
return 2

stamp = [Link]().strftime("%Y%m%d_%H%M%S")
output_dir = [Link]() / f"pmdi_reverse_{stamp}"
output_dir.mkdir(parents=True, exist_ok=True)

config = vars(args).copy()
config["parity"] = str([Link])
config["stopbits"] = str([Link])
config["phase2_layout_names"] = phase2_layout_names
config["phase2_checksum_names"] = phase2_checksum_names
(output_dir / "run_config.json").write_text(
[Link](config, indent=2, ensure_ascii=False),
encoding="utf-8",
)

print("=" * 90)
print("BARRIDO SISTEMÁTICO PMDI / INFINITY")
print("=" * 90)
print(f"Puerto: {[Link]}")
print(f"Serie: {[Link]} {[Link]}{[Link]}{[Link]}")
print(f"Salida: {output_dir.resolve()}")
print("Ctrl+C detiene el ensayo conservando lo registrado.")
print()

try:
ser = [Link](
port=[Link],
baudrate=[Link],
bytesize=[Link],
parity=[Link],
stopbits=[Link],
timeout=0,
write_timeout=1,
xonxoff=[Link],
rtscts=[Link],
dsrdtr=[Link],
)
except [Link] as exc:
print(f"ERROR al abrir {[Link]}: {exc}", file=[Link])
return 1

if not [Link]:
try:
[Link] = False
except (OSError, [Link]):
pass
if not [Link]:
try:
[Link] = False
except (OSError, [Link]):
pass

raw_path = output_dir / "raw_observations.csv"

all_observations: list[Observation] = []
baseline_samples: list[bytes] = []
phase1_rows: list[dict[str, object]] = []
phase2_rows: list[dict[str, object]] = []
clusters: list[dict[str, object]] = []

interrupted = False

try:
with raw_path.open("w", newline="", encoding="utf-8-sig") as raw_file:
raw_writer = [Link](raw_file, fieldnames=RAW_FIELDS)
raw_writer.writeheader()
 # ---------------------------------------------------------------
# Fase 0: línea de base
# ---------------------------------------------------------------
print(f"FASE 0: línea de base ({args.baseline_samples} muestras sin
TX)")
for index in range(1, args.baseline_samples + 1):
rx, first_s, elapsed_s = listen_without_tx(
ser,
window_s=args.rx_window,
end_silence_s=args.end_silence,
)
baseline_samples.append(rx)
print(
f" {index:02d}/{args.baseline_samples} | "
f"RX={hexstr(rx) if rx else '(vacío)'} | "
f"primer_byte="
f"{'N/D' if first_s is None else f'{first_s * 1000:.3f} ms'}"
)
[Link]([Link])

baseline_counter = Counter(rx for rx in baseline_samples if rx)

# ---------------------------------------------------------------
# Fase 1: todos los comandos, estructura fija, SUM8
# ---------------------------------------------------------------
commands = list(range(args.start_command, args.end_command + 1))
phase1_candidates = make_candidates(
phase="phase1",
commands=commands,
layout_names=["primary"],
checksum_names=["sum8"],
)

total_phase1 = len(phase1_candidates) * args.phase1_repetitions

print()
print(
"FASE 1: todos los comandos con "
"00 A5 02 00 CMD SUM8 "
f"({total_phase1} transmisiones)"
)

candidate_index = 0
phase1_observations: list[Observation] = []

for candidate_number, candidate in enumerate(phase1_candidates,

start=1):
candidate_index += 1
for repetition in range(1, args.phase1_repetitions + 1):
obs = transmit_candidate(
ser=ser,
 candidate=candidate,
repetition=repetition,
candidate_index=candidate_index,
pre_tx_s=args.pre_tx,
rx_window_s=args.rx_window,
end_silence_s=args.end_silence,
)
phase1_observations.append(obs)
all_observations.append(obs)
append_observation(raw_writer, obs)
raw_file.flush()

print(
f" cmd=0x{[Link]:02X} "
f"{repetition}/{args.phase1_repetitions} | "
f"TX={obs.tx_hex} | "
f"RX={obs.rx_hex or '(sin respuesta)'} | "
f"{obs.first_byte_ms if obs.first_byte_ms is not None else
'N/D'} ms"
)
[Link]([Link])

phase1_grouped = observations_by_candidate(phase1_observations)
phase1_rows = [
summarize_candidate(
candidate,
phase1_grouped.get(index, []),
baseline_counter,
)
for index, candidate in enumerate(phase1_candidates, start=1)
]
phase1_rows.sort(key=lambda row: float(row["score"]), reverse=True)
write_summary_csv(output_dir / "phase1_ranking.csv", phase1_rows)

# ---------------------------------------------------------------
# Fase 2: adaptativa o cartesiana completa
# ---------------------------------------------------------------
if args.full_cartesian:
selected_commands = commands
print()
print(
"FASE 2 COMPLETA: todos los comandos × layouts × checksums"
)
else:
selected_commands = []
seen: set[int] = set()
for row in phase1_rows:
command = int(row["command_dec"])
if command not in seen:
selected_commands.append(command)
 [Link](command)
if len(selected_commands) >= args.top_commands:
break

print()
print(
"FASE 2 ADAPTATIVA: comandos seleccionados: "
+ ", ".join(f"0x{cmd:02X}" for cmd in selected_commands)
)

phase2_candidates = make_candidates(
phase="phase2",
commands=selected_commands,
layout_names=phase2_layout_names,
checksum_names=phase2_checksum_names,
)

total_phase2 = len(phase2_candidates) * [Link]

estimated_seconds = total_phase2 * (
args.pre_tx + args.rx_window + [Link]
)

print(
f"Candidatos deduplicados: {len(phase2_candidates)} | "
f"Transmisiones: {total_phase2} | "
f"Estimación máxima: {estimated_seconds / 60:.1f} min"
)

phase2_observations: list[Observation] = []
phase2_candidate_start_index = candidate_index

for local_index, candidate in enumerate(phase2_candidates, start=1):

candidate_index += 1
for repetition in range(1, [Link] + 1):
obs = transmit_candidate(
ser=ser,
candidate=candidate,
repetition=repetition,
candidate_index=candidate_index,
pre_tx_s=args.pre_tx,
rx_window_s=args.rx_window,
end_silence_s=args.end_silence,
)
phase2_observations.append(obs)
all_observations.append(obs)
append_observation(raw_writer, obs)
raw_file.flush()

print(
f" {local_index:04d}/{len(phase2_candidates):04d} | "
 f"cmd=0x{[Link]:02X} | "
f"{[Link]:<10} | "
f"{'/'.join(candidate.checksum_methods):<28} | "
f"TX={obs.tx_hex} | "
f"RX={obs.rx_hex or '(sin respuesta)'}"
)
[Link]([Link])

phase2_grouped = observations_by_candidate(phase2_observations)
phase2_rows = []

for offset, candidate in enumerate(phase2_candidates, start=1):

global_index = phase2_candidate_start_index + offset
phase2_rows.append(
summarize_candidate(
candidate,
phase2_grouped.get(global_index, []),
baseline_counter,
)
)

phase2_rows.sort(key=lambda row: float(row["score"]), reverse=True)

write_summary_csv(output_dir / "phase2_summary.csv", phase2_rows)

clusters = greedy_clusters(
phase2_rows,
threshold=args.cluster_threshold,
)
write_clusters(output_dir / "response_clusters.txt", clusters)

except KeyboardInterrupt:
interrupted = True
print("\nEnsayo interrumpido. Se conservaron los datos ya escritos.")
except ([Link], OSError) as exc:
print(f"\nERROR de comunicación: {exc}", file=[Link])
interrupted = True
finally:
if ser.is_open:
[Link]()

# Genera reporte incluso si se interrumpió, cuando hay datos suficientes.

try:
write_report(
output_dir / "[Link]",
args=args,
baseline_samples=baseline_samples,
phase1_rows=phase1_rows,
phase2_rows=phase2_rows,
clusters=clusters,
)
 except Exception as exc:
print(f"No se pudo generar [Link]: {exc}", file=[Link])

print()
print("=" * 90)
print("FINALIZADO" if not interrupted else "FINALIZADO PARCIALMENTE")
print(f"Resultados: {output_dir.resolve()}")
print("Archivos principales:")
print(" raw_observations.csv")
print(" phase1_ranking.csv")
print(" phase2_summary.csv")
print(" response_clusters.txt")
print(" [Link]")
print("=" * 90)

return 0 if not interrupted else 1

if __name__ == "__main__":
raise SystemExit(main())