#!
/usr/bin/env python3
r"""
Diagnóstico de capa física y UART para RS-232/Dräger.
Modos:
- disconnected: adaptador conectado al PC y desconectado del equipo.
- loopback: puente físico entre DB9 pines 2 y 3.
- device: cable conectado al equipo en banco.
Resultados por defecto:
C:\Users\[Link]\Desktop\scripts
Uso:
pip install pyserial
python diagnostico_capa_fisica_rs232.py --port COM5 --mode disconnected
python diagnostico_capa_fisica_rs232.py --port COM5 --mode loopback
python diagnostico_capa_fisica_rs232.py --port COM5 --mode device
"""
from __future__ import annotations
import argparse
import csv
import json
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 Optional
import serial
DEFAULT_OUTPUT_DIR = Path(r"C:\Users\[Link]\Desktop\scripts")
PATTERNS = [
("mixed_reference", [Link]("00 FF 55 AA 33 CC 0F F0")),
("alternating_55", [Link]("55 55 55 55 55 55 55 55")),
("alternating_AA", [Link]("AA AA AA AA AA AA AA AA")),
("all_00", [Link]("00 00 00 00 00 00 00 00")),
("all_FF", [Link]("FF FF FF FF FF FF FF FF")),
("walking_ones", [Link]("01 02 04 08 10 20 40 80")),
("walking_zeros", [Link]("FE FD FB F7 EF DF BF 7F")),
("ascending", [Link]("00 11 22 33 44 55 66 77 88 99 AA BB CC DD EE
FF")),
]
DEFAULT_CONFIGS = [
(4800, 8, "N", 1.0),
(9600, 8, "N", 1.0),
(19200, 8, "N", 1.0),
(38400, 8, "N", 1.0),
(9600, 8, "E", 1.0),
(19200, 8, "E", 1.0),
(9600, 8, "N", 2.0),
(19200, 8, "N", 2.0),
]
@dataclass
class Observation:
timestamp: str
mode: str
port: str
baud: int
bytesize: int
parity: str
stopbits: float
pattern_name: str
repetition: int
tx_hex: str
tx_len: int
rx_hex: str
rx_len: int
exact_echo: bool
tx_found_in_rx: bool
rx_found_in_tx: bool
first_byte_ms: Optional[float]
receive_window_ms: float
bit_similarity: float
best_shift: int
best_shift_similarity: float
inverted_similarity: float
classification: str
cts: str
dsr: str
ri: str
cd: str
def hexstr(data: bytes) -> str:
return " ".join(f"{b:02X}" for b in data)
def parity_value(letter: str) -> str:
return {
"N": serial.PARITY_NONE,
"E": serial.PARITY_EVEN,
"O": serial.PARITY_ODD,
"M": serial.PARITY_MARK,
"S": serial.PARITY_SPACE,
}[[Link]()]
def stopbits_value(value: float) -> float:
return {
1.0: serial.STOPBITS_ONE,
1.5: serial.STOPBITS_ONE_POINT_FIVE,
2.0: serial.STOPBITS_TWO,
}[float(value)]
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 bits_of(data: bytes) -> list[int]:
bits = []
for byte in data:
[Link]((byte >> bit) & 1 for bit in range(8))
return bits
def similarity_bits(a: bytes, b: bytes) -> float:
if not a and not b:
return 1.0
if not a or not b:
return 0.0
abits = bits_of(a)
bbits = bits_of(b)
n = min(len(abits), len(bbits))
equal = sum(1 for i in range(n) if abits[i] == bbits[i])
length_penalty = n / max(len(abits), len(bbits))
return (equal / n) * length_penalty if n else 0.0
def inverted(data: bytes) -> bytes:
return bytes(b ^ 0xFF for b in data)
def shifted_byte_similarity(tx: bytes, rx: bytes, max_shift: int = 3) -> tuple[int,
float]:
best_shift = 0
best_similarity = similarity_bits(tx, rx)
for shift in range(-max_shift, max_shift + 1):
if shift == 0:
a, b = tx, rx
elif shift > 0:
a, b = tx[shift:], rx
else:
a, b = tx, rx[-shift:]
score = similarity_bits(a, b)
if score > best_similarity:
best_shift = shift
best_similarity = score
return best_shift, best_similarity
def classify(
mode: str,
tx: bytes,
rx: bytes,
exact_echo: bool,
similarity: float,
shifted_similarity: float,
inverted_similarity: float,
) -> str:
if not rx:
if mode == "disconnected":
return "correcto_sin_rx"
if mode == "loopback":
return "fallo_loopback_sin_rx"
return "sin_respuesta"
if exact_echo:
if mode == "loopback":
return "loopback_correcto"
if mode == "disconnected":
return "eco_inesperado_adaptador"
return "eco_exacto"
if tx in rx:
return "rx_contiene_tx"
if rx in tx:
return "eco_parcial"
if inverted_similarity >= 0.90:
return "posible_inversion_logica"
if shifted_similarity >= 0.85 and shifted_similarity > similarity + 0.08:
return "posible_desplazamiento_bytes"
if similarity >= 0.80:
return "eco_deformado_fuerte"
if similarity >= 0.60:
return "eco_deformado_posible"
if mode == "disconnected":
return "rx_inesperado_sin_equipo"
if mode == "loopback":
return "loopback_corrupto"
return "respuesta_no_correlacionada"
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 = None
received = bytearray()
while True:
now = time.perf_counter()
if now - start >= total_window_s:
break
available = ser.in_waiting
if available:
block = [Link](available)
if block:
if first_byte is None:
first_byte = 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, time.perf_counter() - start
def open_serial(port: str, baud: int, bytesize: int, parity: str, stopbits: float)
-> [Link]:
return [Link](
port=port,
baudrate=baud,
bytesize=bytesize,
parity=parity_value(parity),
stopbits=stopbits_value(stopbits),
timeout=0,
write_timeout=2,
xonxoff=False,
rtscts=False,
dsrdtr=False,
)
def run_single_test(
ser: [Link],
mode: str,
port: str,
baud: int,
bytesize: int,
parity: str,
stopbits: float,
pattern_name: str,
tx: bytes,
repetition: 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)
written = [Link](tx)
[Link]()
if written != len(tx):
raise [Link](
f"Se escribieron {written} de {len(tx)} bytes"
)
rx, first_byte_s, elapsed_s = read_response(
ser, rx_window_s, end_silence_s
)
exact_echo = rx == tx
normal_similarity = similarity_bits(tx, rx)
best_shift, best_shift_similarity = shifted_byte_similarity(tx, rx)
inverse_similarity = similarity_bits(inverted(tx), rx)
classification = classify(
mode,
tx,
rx,
exact_echo,
normal_similarity,
best_shift_similarity,
inverse_similarity,
)
return Observation(
timestamp=[Link]().isoformat(timespec="milliseconds"),
mode=mode,
port=port,
baud=baud,
bytesize=bytesize,
parity=parity,
stopbits=stopbits,
pattern_name=pattern_name,
repetition=repetition,
tx_hex=hexstr(tx),
tx_len=len(tx),
rx_hex=hexstr(rx),
rx_len=len(rx),
exact_echo=exact_echo,
tx_found_in_rx=bool(tx and tx in rx),
rx_found_in_tx=bool(rx and rx in tx),
first_byte_ms=None if first_byte_s is None else round(first_byte_s * 1000,
3),
receive_window_ms=round(elapsed_s * 1000, 3),
bit_similarity=round(normal_similarity, 4),
best_shift=best_shift,
best_shift_similarity=round(best_shift_similarity, 4),
inverted_similarity=round(inverse_similarity, 4),
classification=classification,
cts=states["cts"],
dsr=states["dsr"],
ri=states["ri"],
cd=states["cd"],
)
def summarize_config(rows: list[Observation]) -> dict[str, object]:
total = len(rows)
received = [r for r in rows if r.rx_len > 0]
exact = [r for r in rows if r.exact_echo]
latencies = [r.first_byte_ms for r in rows if r.first_byte_ms is not None]
classes = Counter([Link] for r in rows)
return {
"baud": rows[0].baud,
"bytesize": rows[0].bytesize,
"parity": rows[0].parity,
"stopbits": rows[0].stopbits,
"tests": total,
"rx_tests": len(received),
"rx_rate": round(len(received) / total, 4) if total else 0,
"exact_echo_tests": len(exact),
"exact_echo_rate": round(len(exact) / total, 4) if total else 0,
"mean_bit_similarity": round(
[Link](r.bit_similarity for r in rows), 4
),
"mean_best_shift_similarity": round(
[Link](r.best_shift_similarity for r in rows), 4
),
"mean_inverted_similarity": round(
[Link](r.inverted_similarity for r in rows), 4
),
"median_first_byte_ms": (
round([Link](latencies), 3) if latencies else ""
),
"classifications": "; ".join(
f"{name}={count}" for name, count in classes.most_common()
),
}
def write_report(
path: Path,
mode: str,
port: str,
observations: list[Observation],
summaries: list[dict[str, object]],
) -> None:
with [Link]("w", encoding="utf-8") as f:
[Link]("DIAGNÓSTICO DE CAPA FÍSICA RS-232\n")
[Link]("=" * 90 + "\n\n")
[Link](f"Fecha: {[Link]().isoformat(timespec='seconds')}\n")
[Link](f"Modo: {mode}\n")
[Link](f"Puerto: {port}\n")
[Link](f"Ensayos: {len(observations)}\n\n")
[Link]("RESUMEN POR CONFIGURACIÓN\n")
[Link]("-" * 90 + "\n")
for row in summaries:
[Link](
f"{row['baud']:>6} "
f"{row['bytesize']}{row['parity']}{row['stopbits']:g} | "
f"RX={row['rx_tests']}/{row['tests']} "
f"({row['rx_rate']:.2f}) | "
f"eco exacto={row['exact_echo_tests']}/{row['tests']} "
f"({row['exact_echo_rate']:.2f}) | "
f"sim={row['mean_bit_similarity']:.3f} | "
f"sim desplazada={row['mean_best_shift_similarity']:.3f} | "
f"sim invertida={row['mean_inverted_similarity']:.3f} | "
f"lat={row['median_first_byte_ms']} ms\n"
)
[Link](f" {row['classifications']}\n")
[Link]("\n")
[Link]("OBSERVACIONES INDIVIDUALES\n")
[Link]("-" * 90 + "\n")
for obs in observations:
[Link](
f"{[Link]:>6} {[Link]}{[Link]}{[Link]:g} | "
f"{obs.pattern_name:<18} | rep={[Link]} | "
f"TX={obs.tx_hex} | "
f"RX={obs.rx_hex or '(vacío)'} | "
f"sim={obs.bit_similarity:.3f} | "
f"shift={obs.best_shift:+d}/{obs.best_shift_similarity:.3f} | "
f"inv={obs.inverted_similarity:.3f} | "
f"{[Link]}\n"
)
def parse_configs(text: Optional[str]) -> list[tuple[int, int, str, float]]:
if not text:
return DEFAULT_CONFIGS
configs = []
for raw in [Link](","):
item = [Link]().upper()
if not item:
continue
try:
speed_text, fmt = [Link]("-", 1)
baud = int(speed_text)
bytesize = int(fmt[0])
parity = fmt[1]
stopbits = float(fmt[2:])
if bytesize not in (5, 6, 7, 8):
raise ValueError
if parity not in ("N", "E", "O", "M", "S"):
raise ValueError
if stopbits not in (1.0, 1.5, 2.0):
raise ValueError
[Link]((baud, bytesize, parity, stopbits))
except (ValueError, IndexError) as exc:
raise [Link](
f"Configuración inválida: {item}. "
"Use 9600-8N1,19200-8E1, etc."
) from exc
if not configs:
raise [Link]("No hay configuraciones válidas.")
return configs
def build_parser() -> [Link]:
p = [Link](
description="Diagnóstico de capa física y UART RS-232."
)
p.add_argument("--port", default="COM5")
p.add_argument(
"--mode",
required=True,
choices=("disconnected", "loopback", "device"),
)
p.add_argument(
"--output-dir",
default=str(DEFAULT_OUTPUT_DIR),
)
p.add_argument(
"--configs",
default=None,
help="Ejemplo: 4800-8N1,9600-8N1,19200-8N1",
)
p.add_argument("--repetitions", type=int, default=3)
p.add_argument("--pre-tx", type=float, default=0.05)
p.add_argument("--rx-window", type=float, default=0.25)
p.add_argument("--end-silence", type=float, default=0.05)
p.add_argument("--pause", type=float, default=0.10)
p.add_argument("--yes", action="store_true")
return p
def main() -> int:
args = build_parser().parse_args()
try:
configs = parse_configs([Link])
except [Link] as exc:
print(f"ERROR: {exc}", file=[Link])
return 2
if [Link] < 1:
print("ERROR: --repetitions debe ser >= 1", file=[Link])
return 2
if [Link] == "device" and not [Link]:
print()
print("ADVERTENCIA")
print("El modo device enviará patrones binarios al equipo.")
print("Debe estar fuera de servicio clínico, sin paciente y en banco.")
if input('Escriba "BANCO" para continuar: ').strip() != "BANCO":
print("Cancelado.")
return 2
base_output = Path(args.output_dir)
base_output.mkdir(parents=True, exist_ok=True)
stamp = [Link]().strftime("%Y%m%d_%H%M%S")
run_dir = base_output / f"diagnostico_rs232_{[Link]}_{stamp}"
run_dir.mkdir(parents=True, exist_ok=True)
config_payload = {
"port": [Link],
"mode": [Link],
"output_dir": str(base_output),
"configs": configs,
"repetitions": [Link],
"pre_tx": args.pre_tx,
"rx_window": args.rx_window,
"end_silence": args.end_silence,
"pause": [Link],
"patterns": [
{"name": name, "hex": hexstr(pattern)}
for name, pattern in PATTERNS
],
}
(run_dir / "[Link]").write_text(
[Link](config_payload, indent=2, ensure_ascii=False),
encoding="utf-8",
)
raw_path = run_dir / "resultados_crudos.csv"
summary_path = run_dir / "resumen_configuraciones.csv"
report_path = run_dir / "[Link]"
observations = []
fieldnames = list(Observation.__dataclass_fields__.keys())
print("=" * 90)
print("DIAGNÓSTICO DE CAPA FÍSICA RS-232")
print("=" * 90)
print(f"Modo: {[Link]}")
print(f"Puerto: {[Link]}")
print(f"Salida: {run_dir}")
print()
try:
with raw_path.open("w", newline="", encoding="utf-8-sig") as raw_file:
writer = [Link](raw_file, fieldnames=fieldnames)
[Link]()
for i, (baud, bytesize, parity, stopbits) in enumerate(configs, 1):
print(
f"[{i}/{len(configs)}] "
f"{baud} {bytesize}{parity}{stopbits:g}"
)
try:
ser = open_serial(
[Link], baud, bytesize, parity, stopbits
)
except [Link] as exc:
print(
f" ERROR al abrir {[Link]}: {exc}",
file=[Link],
)
continue
try:
try:
[Link] = False
except (OSError, [Link]):
pass
try:
[Link] = False
except (OSError, [Link]):
pass
for pattern_name, pattern in PATTERNS:
for repetition in range(1, [Link] + 1):
obs = run_single_test(
ser,
[Link],
[Link],
baud,
bytesize,
parity,
stopbits,
pattern_name,
pattern,
repetition,
args.pre_tx,
args.rx_window,
args.end_silence,
)
[Link](obs)
[Link](asdict(obs))
raw_file.flush()
print(
f" {pattern_name:<18} "
f"{repetition}/{[Link]} | "
f"RX={obs.rx_hex or '(vacío)'} | "
f"sim={obs.bit_similarity:.3f} | "
f"inv={obs.inverted_similarity:.3f} | "
f"{[Link]}"
)
[Link]([Link])
finally:
[Link]()
except KeyboardInterrupt:
print("\nInterrumpido. Los datos registrados se conservaron.")
except (OSError, [Link]) as exc:
print(f"\nERROR de comunicación: {exc}", file=[Link])
grouped = {}
for obs in observations:
key = ([Link], [Link], [Link], [Link])
[Link](key, []).append(obs)
summaries = [
summarize_config(rows)
for _, rows in sorted([Link](), key=lambda x: x[0])
]
if summaries:
with summary_path.open("w", newline="", encoding="utf-8-sig") as f:
writer = [Link](f, fieldnames=list(summaries[0].keys()))
[Link]()
[Link](summaries)
else:
summary_path.write_text("", encoding="utf-8")
write_report(report_path, [Link], [Link], observations, summaries)
print()
print("=" * 90)
print("FINALIZADO")
print(f"Carpeta: {run_dir}")
print("Archivos:")
print(" resultados_crudos.csv")
print(" resumen_configuraciones.csv")
print(" [Link]")
print(" [Link]")
print("=" * 90)
return 0
if __name__ == "__main__":
raise SystemExit(main())