◇ OpenCyber Academy
Técnico · Capítulo 4 · Español

Capítulo 4 — Scripts y formatos de datos

Aprendizaje gratuito y abierto para quienes desean proteger a las personas y la tecnología. No se necesita una cuenta para estudiar ni consultar las respuestas.

Capítulo 4 — Scripts y formatos de datos

Mapa de la lección

Lección 04 publicada. Mapa, ejemplo, EX013.

flowchart TD
  N0["Resultados de aprendizaje y conexión"]
  N1["4.1 Formatos con los que se encontrará"]
  N2["4.2 Un orden de validación que falla de forma cerrad"]
  N3["4.3 Coordenadas del registro"]
  N4["4.5 Actividad práctica"]
  N5["Síntesis del capítulo"]
  N0 --> N1 --> N2 --> N3 --> N4 --> N5
  N5 --> EX["Ejemplo resuelto"]
  EX --> Q["Ejercicio EX013"]

Dibujo lógico

flowchart LR
  N0["Resultados de aprendizaje y conexión"]
  N1["4.1 Formatos con los que se encontrará"]
  N2["4.2 Un orden de validación que falla de forma cerrad"]
  N3["4.3 Coordenadas del registro"]
  N4["4.5 Actividad práctica"]
  N5["Síntesis del capítulo"]
  N0 --> N1 --> N2 --> N3 --> N4 --> N5 --> Q["EX013"]

[Resultados de aprendizaje y conexión]
    |
    v
[4.1 Formatos con los que se encontrará]
    |
    v
[4.2 Un orden de validación que falla de forma cerrad]
    |
    v
[4.3 Coordenadas del registro]
    |
    v
[4.5 Actividad práctica]
    |
    v
[Síntesis del capítulo]
    |
    v
[Ejemplo resuelto] --> [Ejercicio EX013]

Resultados de aprendizaje y conexión

Al terminar este capítulo debería ser capaz de:

Los scripts del técnico de los capítulos posteriores (decodificadores, auditores, detectores) leen todos datos que produjo otra persona. Este capítulo los vuelve estrictos con lo que aceptan, porque un script que «corrige» en silencio una entrada errónea produce respuestas equivocadas con total seguridad.

4.1 Formatos con los que se encontrará

4.2 Un orden de validación que falla de forma cerrada

Compruebe en un orden en el que cada paso pueda suponer que los anteriores se superaron:

  1. Estructura: el número correcto de campos. Si no lo es, no puede confiarse en nada más del registro: rechazar.
  2. Identidad y tipo: estación y etiqueta conocidas; la unidad que indica el esquema (un nivel en pies no es un nivel en metros); el valor se analiza como número.
  3. Finitud: float("nan") y float("inf") se analizan sin error, pero no son mediciones: rechazar.
  4. Tiempo: la marca de tiempo se analiza en el formato acordado; no está en el futuro más allá de una pequeña tolerancia.
  5. Verosimilitud: el valor está dentro del rango físico del instrumento y del proceso.

Rechace lo que no puede interpretarse (estructura errónea, unidad errónea, no es un número). Ponga en cuarentena lo que puede interpretarse pero es inverosímil (fuera de rango, hora futura): un fallo real de un sensor y un ataque real tienen ambos este aspecto, y un operador debe verlo en lugar de que se descarte en silencio. No fuerce nunca la conversión: convertir "12,5" en 12.5 o en 125 adivinando el separador decimal crea datos que nunca se midieron.

4.3 Coordenadas del registro

Todo hallazgo debe decir de dónde procede (archivo, línea y campo) para que otra persona pueda verificarlo. csv.reader.line_num cuenta las líneas físicas leídas hasta el momento, incluida la cabecera, de modo que la primera fila de datos es la línea 2. Los campos entrecomillados que contienen saltos de línea hacen que las líneas físicas difieran de los registros; line_num indica la línea en la que terminó el registro, que sigue siendo una coordenada utilizable. Cuando pegue un hallazgo en una incidencia, mantenga el número de línea junto al nombre de la estación, para que la siguiente persona pueda abrir la exportación e ir directamente a la fila. Si un campo entrecomillado ocupa varias líneas, indíquelo, porque la coordenada marca entonces el final del registro y no su comienzo.

4.4 Ejemplo resuelto: validar una exportación de estaciones (sintético)

Una exportación del historiador de Kestrel Water contiene una lectura por estación. La hora del análisis es 06:00Z del 12 de octubre de 2026. Lab 4 (guárdelo como lab04.py y ejecútelo con python3 lab04.py):

# Lab 4 - validate a telemetry export with record coordinates (offline, standard library only)
# Synthetic CSV export from Kestrel Water stations; "now" is 2026-10-12T06:00:00Z.
import csv, io, json, math
from datetime import datetime, timezone
export = """station,tag,value,unit,ts
KW-01,LEVEL_M,3.42,m,2026-10-12T05:58:00Z
KW-02,CL2_MGL,0.61,mg/L,2026-10-12T05:58:00Z
KW-03,LEVEL_M,,m,2026-10-12T05:58:00Z
KW-04,CL2_MGL,7.9,mg/L,2026-10-12T05:58:00Z
KW-05,FLOW_LS,"12,5",L/s,2026-10-12T05:58:00Z
KW-06,LEVEL_M,2.10,ft,2026-10-12T05:58:00Z
KW-07,LEVEL_M,2.95,m,2026-10-12 05:58
KW-08,LEVEL_M,nan,m,2026-10-12T05:58:00Z
KW-09,FLOW_LS,41.0,L/s,2026-10-12T07:15:00Z
KW-10,LEVEL_M,1.88,m,2026-10-12T05:58:00Z,extra
"""
SCHEMA = {"LEVEL_M": ("m", 0.0, 12.0), "CL2_MGL": ("mg/L", 0.0, 5.0), "FLOW_LS": ("L/s", 0.0, 400.0)}
now = datetime(2026, 10, 12, 6, 0, tzinfo=timezone.utc)
result = {"accept": [], "reject": [], "quarantine": []}
reader = csv.reader(io.StringIO(export))
header = next(reader)
for row in reader:
    where = f"line {reader.line_num}"
    if len(row) != len(header): result["reject"].append((where, f"{len(row)} fields, expected {len(header)}")); continue
    r = dict(zip(header, row))
    unit, lo, hi = SCHEMA[r["tag"]]
    if r["unit"] != unit: result["reject"].append((where, f"{r['station']} unit {r['unit']!r}, expected {unit!r}")); continue
    try: v = float(r["value"])
    except ValueError: result["reject"].append((where, f"{r['station']} value {r['value']!r} not a number")); continue
    if not math.isfinite(v): result["reject"].append((where, f"{r['station']} value not finite")); continue
    try:
        ts = datetime.strptime(r["ts"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
    except ValueError: result["reject"].append((where, f"{r['station']} timestamp {r['ts']!r} not RFC 3339 UTC")); continue
    if ts > now: result["quarantine"].append((where, f"{r['station']} timestamp {(ts - now).seconds // 60} min in the future")); continue
    if not lo <= v <= hi: result["quarantine"].append((where, f"{r['station']} {r['tag']}={v} outside {lo}-{hi}")); continue
    result["accept"].append((where, r["station"]))
for k in ("reject", "quarantine"):
    for where, why in result[k]: print(f"{k:<10} {where:<8} {why}")
print(json.dumps({k: len(v) for k, v in result.items()}))

Salida esperada (Python 3.13.5):

reject     line 4   KW-03 value '' not a number
reject     line 6   KW-05 value '12,5' not a number
reject     line 7   KW-06 unit 'ft', expected 'm'
reject     line 8   KW-07 timestamp '2026-10-12 05:58' not RFC 3339 UTC
reject     line 9   KW-08 value not finite
reject     line 11  6 fields, expected 5
quarantine line 5   KW-04 CL2_MGL=7.9 outside 0.0-5.0
quarantine line 10  KW-09 timestamp 75 min in the future
{"accept": 2, "reject": 6, "quarantine": 2}

Solo se aceptan dos de diez filas. Cada rechazo indica su línea: un valor vacío (línea 4), una coma decimal dentro de un campo entrecomillado (línea 6), un nivel en pies (línea 7), una marca de tiempo sin segundos ni desfase (línea 8), nan (línea 9) y un campo de más (línea 11). Las filas en cuarentena son las interesantes: un cloro de 7,9 mg/L está fuera del rango del instrumento (un fallo del sensor, un error de escalado o un valor manipulado), y la marca de tiempo de KW-09 está 75 minutos en el futuro, algo que en el capítulo 7 aprenderá a sospechar como un problema de reloj. Ambas se pasan a una persona con sus coordenadas.

El resumen JSON de la última línea es deliberadamente legible por máquina, para que una tarea de supervisión pueda alertar cuando suba el número de rechazos.

4.5 Actividad práctica

Añada al Lab 4 una comprobación de identidad de la estación: solo son válidas de KW-01 a KW-12. Añada después una fila KW-13,LEVEL_M,2.0,m,2026-10-12T05:58:00Z y confirme que se rechaza con su número de línea. Por último, haga que el analizador de marcas de tiempo acepte un desfase explícito +00:00 además de Z, sin aceptar marcas de tiempo sin zona. Rechace KW-13 con su número de línea y un motivo que nombre el conjunto permitido. Para las marcas de tiempo, acepte tanto Z como un desfase explícito de más cero, y siga rechazando una hora local sin zona. Vuelva a ejecutar el laboratorio y confirme que el número de rechazos del resumen JSON sube en uno por la nueva estación.

4.6 Errores comunes y límites

Ejercicios — capítulo 4

EX013 — Prediga antes de ejecutar. Objetivo: razonar sobre el validador. Entradas: el Lab 4 con tres filas nuevas: KW-11,LEVEL_M,12.0,m,2026-10-12T05:59:00Z; KW-12,LEVEL_M,inf,m,2026-10-12T05:59:00Z; KW-02,CL2_MGL,-0.0,mg/L,2026-10-12T05:59:00Z. Pasos: prediga cada veredicto y el nuevo resumen, y después ejecute. Evidencia esperada: predicciones, salida real y explicación de cualquier sorpresa. Rúbrica: 3 = aceptar, rechazar (no finito), aceptar; resumen {"accept": 4, "reject": 7, "quarantine": 2}; señala que −0.0 debería normalizarse; 2 = uno erróneo pero explicado; 1 = conjeturas; 0 = nada.

EX014 — ¿Rechazar o poner en cuarentena? Objetivo: aplicar la distinción. Entradas: (a) caudal −3,0 L/s; (b) unidad «M» en lugar de «m»; (c) nivel de 11,9 m en un depósito de 12 m; (d) marca de tiempo 4 s en el futuro. Pasos: decida y justifique. Evidencia esperada: cuatro veredictos. Rúbrica: 3 = (a) cuarentena (interpretable, inverosímil; podría ser un flujo inverso o un desfase); (b) rechazar, o normalizar solo si el esquema admite explícitamente unidades sin distinguir mayúsculas, sin adivinar; (c) aceptar, quizá con una alarma de nivel alto; (d) aceptar dentro de una tolerancia de reloj acordada; 2 = una débil; 1 = rechazarlas todas; 0 = nada.

EX015 — JSON con trampas. Objetivo: bastionar el análisis de JSON. Entradas: {"station": "KW-03", "value": NaN}. Pasos: muestre lo que devuelve json.loads por defecto y escriba el cambio de una línea que lo rechaza. Evidencia esperada: código y comportamiento. Rúbrica: 3 = por defecto devuelve un float nan; json.loads(s, parse_constant=lambda c: (_ for _ in ()).throw(ValueError(c))) o una función equivalente que lance una excepción; explica que la RFC 8259 no admite NaN; 2 = funciona sin explicación; 1 = intentado; 0 = nada.

EX016 — Resumen para la supervisión. Objetivo: diseñar un resultado legible por máquina. Entradas: el Lab 4. Pasos: amplíe el resumen JSON con recuentos por motivo y por estación, y proponga una condición de alerta sobre él. Evidencia esperada: un JSON de ejemplo y la condición. Rúbrica: 3 = claves estables, recuentos por motivo, una alerta como «cualquier cuarentena de CL2_MGL, o rechazos > 10 % de las filas en dos exportaciones consecutivas»; 2 = sin desglose por motivo; 1 = solo prosa; 0 = nada.

Síntesis del capítulo

Los datos operativos llegan mal formados, ambiguos y, a veces, hostiles. Analice con analizadores de verdad, valide en un orden que falle de forma cerrada, no fuerce nunca conversiones e informe de cada hallazgo con sus coordenadas. Rechace lo que no puede interpretar; ponga en cuarentena lo que puede interpretar pero no debería creer; y mantenga los resúmenes legibles por máquina para que el propio validador pueda supervisarse.

Siga estudiando — capítulo 04 de 18

← Capítulo 03 · Índice de capítulos del curso · Capítulo 05 →

Continúe con la siguiente lección escrita.

Respuestas desarrolladas

EX013 — Prediga antes de ejecutar. KW-11 12.0 → aceptar (rango inclusivo); KW-12 inf → rechazar value not finite (línea 13); KW-02 −0.0 → aceptar. Resumen {"accept": 4, "reject": 7, "quarantine": 2} (verificado ejecutándolo). Normalice −0.0 a 0.0 antes de almacenarlo.

EX014 — ¿Rechazar o poner en cuarentena? (a) Cuarentena: interpretable pero fuera de rango; flujo inverso, desfase del sensor o manipulación. (b) Rechazar, salvo que el esquema defina explícitamente las unidades sin distinguir mayúsculas; no adivine nunca. (c) Aceptar; una alarma de proceso de nivel alto es la herramienta del operador, no la del validador. (d) Aceptar si está dentro de una tolerancia de reloj acordada (por ejemplo, 5 s); en caso contrario, cuarentena como problema de reloj.

EX015 — JSON con trampas. json.loads('{"station": "KW-03", "value": NaN}') devuelve {'station': 'KW-03', 'value': nan}. Defina def no_const(c): raise ValueError(f"non-standard JSON constant {c}") y llame a json.loads(s, parse_constant=no_const); ahora lanza una excepción. La RFC 8259 no admite NaN ni Infinity.

EX016 — Resumen para la supervisión. Ejemplo: {"accept": 2, "reject": 6, "quarantine": 2, "by_reason": {"not a number": 2, "unit": 1, "timestamp": 1, "not finite": 1, "field count": 1, "range": 1, "future": 1}, "by_station": {"KW-03": 1, ...}}. Alerta: cualquier cuarentena de una etiqueta CL2_MGL, o rechazos por encima del 10 % de las filas en dos exportaciones consecutivas, o que falte por completo la exportación.