1. Python Orquestrador & Utilidades

La capa de Python actúa como el cerebro del sensor, gestionando el ciclo de vida de las capturas, la salud del sistema y la comunicación externa.

1.1. Núcleo de Control

Estos módulos gestionan la lógica principal y el flujo de trabajo de las mediciones de espectro.

Orquestrador Principal

Módulo Orquestador Principal.

Este módulo actúa como el motor central del sistema, coordinando la transición entre diferentes estados (IDLE, REALTIME, CAMPAIGN, KALIBRATING). Gestiona la descarga de configuraciones desde la API, la lógica de adquisición de datos en tiempo real y la programación de campañas mediante tareas cron.

class orchestrator.ManagedProc(name: str, argv: List[str], env: Dict[str, str], proc: subprocess.Popen, log_thread: threading.Thread)

Bases: object

argv: List[str]

env: Dict[str, str]

log_thread: Thread

name: str

proc: Popen

orchestrator.cleanup_stale_processes(match_terms: List[str], log: Logger, term_timeout: float = 2.0) → None

orchestrator.fetch_realtime_config(client)

Descarga y valida la configuración de tiempo real desde el servidor.

Realiza una petición GET al endpoint de configuración, calcula la latencia de red (delta_t) y mapea la respuesta JSON a objetos de configuración especializados para filtrado y demodulación.

Parámetros:

client (RequestClient) – Cliente HTTP para realizar la consulta.

Devuelve:

Un conjunto de tres elementos:

• dict: Configuración mapeada como diccionario (asdict). Vacío si falla.

• Response: Objeto de respuesta HTTP completo.

• int: Latencia de la petición en milisegundos (delta_t_ms).

Tipo del valor devuelto:

tuple

async orchestrator.main() → int

Punto de entrada principal del orquestador.

Inicializa los servicios base (Store, Client, Scheduler) y entra en un bucle infinito. Utiliza temporizadores (ElapsedTimer) para decidir cuándo consultar la API por nuevas configuraciones de tiempo real o campañas programadas.

Devuelve:

Código de salida del script.

Tipo del valor devuelto:

int

async orchestrator.run_campaigns_logic(client: RequestClient, store: ShmStore, scheduler: CronSchedulerCampaign) → int

Sincroniza y gestiona las campañas de medición programadas.

Consulta la lista de campañas pendientes en el servidor. Si hay campañas activas, ejecuta una calibración y pone al sistema en modo CAMPAIGN. Se mantiene monitoreando la ventana de tiempo hasta que no queden tareas pendientes.

Parámetros:

• client (RequestClient) – Cliente para comunicación con la API.

• store (ShmStore) – Almacén de persistencia.

• scheduler (CronSchedulerCampaign) – Gestor de tareas programadas en el sistema.

Devuelve:

Código de estado (0 éxito, 1 si el sistema no está IDLE o falla la red).

Tipo del valor devuelto:

int

async orchestrator.run_realtime_logic(client: RequestClient, store: ShmStore) → int

Gestiona el bucle de ejecución para el modo de tiempo real.

Establece una conexión ZMQ con el backend de procesamiento (DSP), adquiere espectros (con o sin demodulación/offset) y los sube a la API. El bucle se mantiene activo hasta que el servidor deja de enviar una configuración válida o se alcanza el tiempo de rotación.

Parámetros:

• client (RequestClient) – Cliente para comunicación con la API.

• store (ShmStore) – Almacén para guardar el delta de tiempo de red.

Devuelve:

Código de estado (0 para fin de ciclo, 1 si el sistema está ocupado).

Tipo del valor devuelto:

int

orchestrator.start_managed_process(*, name: str, argv: List[str], env: Dict[str, str], log: Logger, stale_match_terms: List[str] | None = None) → ManagedProc

orchestrator.stop_managed_process(mp: ManagedProc | None, log: Logger, timeout: float = 2.0) → None

Ejecutor de Campañas

Módulo Ejecutor de Campañas (Campaign Runner).

Este script realiza un ciclo único de adquisición de datos de radiofrecuencia (RF) asociado a una campaña programada. Se encarga de leer los parámetros de hardware desde la memoria compartida, coordinar la captura con el motor RF, intentar la subida de datos a la API y gestionar el almacenamiento local (Cola de reintentos e Histórico) según el estado del disco.

class campaign_runner.CampaignRunner

Controlador para la ejecución de una tarea de adquisición de campaña.

Atributos:

status_obj (StatusDevice): Herramienta para monitorear el estado del hardware (disco, etc). cli (RequestClient): Cliente para realizar peticiones HTTP a la API central. store (ShmStore): Interfaz de acceso a la memoria compartida del sistema. campaign_id (int/str): Identificador de la campaña actual obtenido de la persistencia.

async acquire_payload(rf_cfg: dict)

Gestiona la comunicación ZMQ con el motor RF para obtener la muestra.

Utiliza una estrategia de adquisición de campaña que realiza doble captura para eliminar el pico DC (spectral stitching).

Parámetros:

rf_cfg (dict) – Configuración de radiofrecuencia a aplicar en el hardware.

Devuelve:

El payload con los datos espectrales corregidos o None si falla.

Tipo del valor devuelto:

dict/None

async run() → int

Orquestador principal del flujo de ejecución del runner.

Pasos del flujo:

1. Carga de parámetros.

2. Adquisición de datos mediante ZMQ.

3. Intento de carga a la API.

4. Gestión de almacenamiento: Si falla la red, guarda en cola; si tiene éxito, gestiona el histórico y limpia el disco si es necesario.

Devuelve:

Código de salida (0 éxito, 1 fallo).

Tipo del valor devuelto:

int

1.2. Streaming y Multimedia

Gestión de audio de baja latencia para monitoreo remoto de señales demoduladas.

class server_webrtc.Publisher(loop, ws)

Maneja el pipeline de GStreamer para procesar y transmitir audio Opus vía WebRTC.

Esta clase inicializa un elemento “webrtcbin”, gestiona la generación de ofertas SDP, el intercambio de candidatos ICE y la inyección de frames de audio en el flujo.

Parámetros:

• loop (asyncio.AbstractEventLoop) – El bucle de eventos de asyncio para tareas asíncronas.

• ws (websockets.WebSocketClientProtocol) – La conexión activa del socket de señalización.

add_candidate(mline, cand)

on_bus(bus, msg)

on_ice_candidate(element, mline, candidate)

on_negotiation_needed(element)

on_offer_created(promise, element, _)

push_opus_frame(opus_bytes: bytes)

set_answer(sdp_text)

start()

stop()

async server_webrtc.main()

Punto de entrada principal que coordina el ciclo de vida de la aplicación.

Configura los manejadores de señales (SIGINT, SIGTERM), inicia el servidor TCP y ejecuta un bucle de reintento para la sesión de señalización en caso de fallos de red.

Devuelve:

Código de salida (0 para éxito, 1 para errores críticos).

Tipo del valor devuelto:

int

async server_webrtc.run_signaling_session()

Establece y mantiene la conexión con el servidor de señalización WebRTC.

Registra el dispositivo como “sensor”, instancia el Publisher y maneja los mensajes entrantes de tipo “answer” (respuesta SDP) y “candidate” (ICE).

Devuelve:

La función se ejecuta indefinidamente hasta un error o cierre de sesión.

Tipo del valor devuelto:

None

async server_webrtc.tcp_reader_task()

Servidor asíncrono que escucha conexiones TCP entrantes para recibir audio.

Lee los encabezados definidos en HDR_FMT (!IIIHH) para validar los paquetes y extrae el payload de audio para entregarlo al Publisher activo.

Muestra:

IncompleteReadError – Si la conexión se cierra antes de recibir el encabezado completo.

Nota

Pipeline de Audio: El flujo de datos sigue el camino: Captura SDR → Codificación Opus → Transporte TCP (puerto 9000) → GStreamer → Protocolo WebRTC para visualización en navegador.

1.3. Sincronización y Calibración

Módulos dedicados a garantizar que los datos recolectados sean precisos tanto en el dominio de la frecuencia como en el tiempo.

Herramientas de Precisión

Módulo	Descripción Funcional
kal_sync	Algoritmos de calibración de error en PPM (partes por millón) usando estaciones base GSM como referencia.
status	Monitoreo de telemetría: deriva de reloj NTP, uso de recursos y estado del hardware.

Módulo de Gestión de Estado del Dispositivo.

Este script se encarga de recopilar métricas del sistema, como la sincronización NTP, tiempos de calibración y deltas de tiempo, para construir un paquete de datos (payload) y enviarlo al servidor central a través de una API REST.

status.build_status_final_payload(store, device)

Construye el payload final con el estado completo del dispositivo.

Recopila información desde el almacenamiento persistente (ShmStore) y del sistema (NTP) para generar una instantánea (snapshot) del estado actual.

Parámetros:

• store (ShmStore) – Instancia del almacén de memoria compartida para consultar persistencia.

• device (StatusDevice) – Instancia del dispositivo para generar el formato del snapshot.

Devuelve:

Diccionario formateado con todas las métricas, MAC y timestamp actual.

Tipo del valor devuelto:

dict

status.main() → int

Punto de entrada principal del script.

Inicializa los clientes de comunicación, construye el payload de estado y realiza la petición POST hacia el endpoint configurado.

Devuelve:

Código de salida (0 para éxito, 1 para error).

Tipo del valor devuelto:

int

1.4. Infraestructura y Resiliencia

Módulos encargados de la robustez del sistema, manejo de configuraciones y persistencia de datos ante fallos de red.

1.4.1. Gestión de Configuración y Logs

Módulo de Configuración y Gestión de Entorno.

Centraliza las constantes de la API, rutas de directorios y la lógica de identificación del sensor. Incluye un sistema de logging especializado que minimiza el riesgo de corrupción de datos en sistemas embebidos.

cfg.API_URL = 'https://rsm.ane.gov.co:12443/api/sensor'

URL base de la API del sensor

class cfg.AtomicRotator(module_name: str, max_lines: int, max_files: int)

Manejador de archivos de log con rotación y escritura segura.

Monitorea el número de líneas escritas y gestiona un ciclo de vida de archivos (N archivos máximos) para evitar el llenado del almacenamiento.

flush()

write(data: str)

cfg.DEBUG = False

Controla si los mensajes DEBUG se habilitan en consola y archivo

cfg.DEVELOPMENT = False

Entorno de desarrollo (usa DUMMY_MAC)

class cfg.HandlerLevelFilter(allow_debug: bool, allow_info: bool)

Filtro explícito por severidad.

• DEBUG depende de allow_debug.

• INFO depende de allow_info.

• WARNING, ERROR, CRITICAL y excepciones siempre pasan.

filter(record: LogRecord) → bool

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

cfg.IPC_ADDR = 'ipc:///tmp/rf_engine'

Dirección del socket IPC para comunicación con el motor RF

cfg.PYTHON_ENV = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/sdr-spectrummonitoring-sensor/checkouts/latest/venv/bin/python3')

Ruta al ejecutable de Python dentro del entorno virtual

class cfg.SimpleFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class cfg.Tee(primary, manager: AtomicRotator | None)

flush()

write(data)

cfg.VERBOSE = False

Controla si los mensajes INFO se muestran en consola

cfg.get_mac() → str

Escanea las interfaces de red del sistema para obtener la MAC física.

Ignora interfaces virtuales (docker, loopback, tun) y prioriza “wlan” para asegurar una identificación única del hardware del sensor.

Devuelve:

Dirección MAC en formato “xx:xx:xx:xx:xx:xx”.

Tipo del valor devuelto:

str

cfg.get_time_ms() → int

Obtiene el tiempo actual en milisegundos.

Ajusta el timestamp de Unix al desfase horario de Colombia (UTC-5) para mantener consistencia con el servidor central.

Devuelve:

Timestamp en milisegundos.

Tipo del valor devuelto:

int

cfg.human_readable(ts_ms: int, target_tz: str = 'UTC') → str

Convierte un timestamp ms a cadena legible.

cfg.run_and_capture(func: Callable[[], int] | Callable[[], Coroutine[Any, Any, int]], num_files=10) → int

Wrapper de ejecución segura para puntos de entrada (main).

Configura automáticamente el logger del módulo, inicializa el rotador atómico y captura cualquier excepción no controlada o señal de interrupción (Ctrl+C) para cerrar el programa limpiamente.

Parámetros:

• func – Función o Corrutina a ejecutar.

• num_files – Cantidad máxima de archivos de log a conservar.

Devuelve:

Código de salida (0 para éxito, 1 para error).

Tipo del valor devuelto:

int

cfg.set_logger(rotator: AtomicRotator | None = None) → Logger

Configura logger asimétrico.

Consola: - DEBUG solo si DEBUG=True - INFO solo si VERBOSE=True - WARNING/ERROR/EXCEPTION siempre visibles

Archivo en Logs/: - DEBUG solo si DEBUG=True - INFO/WARNING/ERROR/EXCEPTION siempre visibles

1.4.2. Cola de Reintentos (Offline Storage)

Módulo de Gestión de Cola de Reintentos.

Este script se encarga de procesar archivos JSON almacenados localmente que no pudieron ser enviados previamente. Gestiona la lógica de reintentos, validación de archivos y limpieza de la cola de envío hacia la API central.

retry_queue.attempt_send(cli: RequestClient, payload: dict, url: str) → int

Intenta enviar un payload a la API una sola vez.

Realiza una petición POST y analiza la respuesta para determinar si el error es transitorio o permanente.

Parámetros:

• cli (RequestClient) – Instancia del cliente de peticiones HTTP.

• payload (dict) – Diccionario con los datos a enviar.

• url (str) – URL del endpoint de la API.

Devuelve:

Código de retorno basado en el resultado:

• RC_OK: Éxito (2xx).

• RC_NETWORK: Error transitorio de red o servidor (5xx).

• RC_JSON: Error de validación del cliente (4xx).

• RC_UNEXPECTED: Error inesperado o respuesta malformada.

Tipo del valor devuelto:

int

retry_queue.main() → int

Punto de entrada principal para el ejecutor de la cola de reintentos.

Inicializa el cliente HTTP y lanza el procesamiento de la cola.

Devuelve:

RC_OK si el proceso termina correctamente, RC_NETWORK si falla la inicialización.

Tipo del valor devuelto:

int

retry_queue.retry_queue(cli: RequestClient) → int

Procesa los archivos en el directorio de cola intentando reenviarlos.

Sigue un orden de antigüedad (los más viejos primero). Los archivos corruptos o enviados con éxito se eliminan. Si un archivo agota sus reintentos por error de red, el proceso se detiene inmediatamente para preservar el orden.

Parámetros:

cli (RequestClient) – Cliente para realizar los reintentos de envío.

Devuelve:

Siempre retorna RC_OK tras finalizar o detener el procesamiento. RC_IO: Si hay errores persistentes al listar el directorio.

Tipo del valor devuelto:

int

1.5. Paquetes de Bajo Nivel (Utils)

Funciones de utilidad general para manipulación de archivos, peticiones HTTP y estados del sistema.

Utilidades de Entrada/Salida

Módulo de Utilidades de E/S e Integridad de Datos.

Este módulo provee herramientas para el manejo seguro de archivos y persistencia:

1. Escritura Atómica: Evita la corrupción de archivos en caso de fallos.

2. ShmStore: Almacenamiento basado en RAM (/dev/shm) con bloqueo de archivos (file locking) para comunicación segura entre procesos.

3. Temporizadores: Control de flujo basado en tiempo.

class utils.io_util.ElapsedTimer

Temporizador simple de cuenta regresiva.

Permite verificar si ha transcurrido un intervalo de tiempo determinado sin bloquear el hilo de ejecución.

init_count(seconds: float)

Inicia la cuenta regresiva.

Parámetros:

seconds (float) – Segundos a esperar desde este momento.

time_elapsed() → bool

Verifica si el tiempo ya transcurrió.

Devuelve:

True si el tiempo actual superó el tiempo objetivo, False si no.

Tipo del valor devuelto:

bool

class utils.io_util.ShmStore(filename: str = 'persistent.json')

Almacenamiento de persistencia rápida en memoria compartida (RAM).

Utiliza el sistema de archivos /dev/shm de Linux para almacenar un objeto JSON. Es ideal para compartir variables de estado entre el motor de RF y los scripts de Python sin desgastar la tarjeta SD.

Atributos:

filepath (str): Ruta completa al archivo en la memoria compartida.

add_to_persistent(key: str, value: Any)

Actualiza una clave específica sin afectar al resto de los datos.

Parámetros:

• key (str) – Nombre de la clave.

• value (Any) – Valor a almacenar.

clear_persistent()

Limpia todo el almacenamiento, dejándolo como un objeto vacío {}.

consult_persistent(key: str) → Any | None

Consulta el valor de una clave.

Parámetros:

key (str) – Clave a buscar.

Devuelve:

El valor encontrado o None si la clave no existe.

Tipo del valor devuelto:

Any | None

update_from_dict(data_dict: dict)

Actualiza múltiples valores de forma atómica mediante un diccionario.

Parámetros:

data_dict (dict) – Conjunto de pares clave-valor a actualizar.

utils.io_util.atomic_write_bytes(target_path: Path, data: bytes) → None

Escribe datos en una ruta de forma atómica.

Para evitar que un archivo quede corrupto o a medias tras un fallo de energía o del sistema, esta función escribe primero en un archivo temporal y luego reemplaza el archivo destino en una sola operación del sistema operativo.

Parámetros:

• target_path (Path) – Ruta del archivo final.

• data (bytes) – Contenido binario a escribir.

Muestra:

Exception – Si ocurre un error durante la escritura, sincronización o reemplazo.

Cliente de API (Requests)

Utilidades de Comunicación (HTTP y ZMQ).

Este módulo centraliza las interacciones externas e internas del sensor. Incluye validadores de configuración de hardware, un cliente HTTP robusto con manejo de errores unificado y un controlador de sockets ZeroMQ para la comunicación entre procesos (IPC).

class utils.request_util.FilterConfig(start_freq_hz: int, end_freq_hz: int)

Configuración de filtrado digital para la señal de RF.

end_freq_hz: int

Frecuencia de fin en Hertz

start_freq_hz: int

Frecuencia de inicio en Hertz

class utils.request_util.RequestClient(base_url: str, mac_wifi: str = '', timeout: Tuple[float, float] = (5, 15), verbose: bool = False, logger=None)

Cliente HTTP ligero con códigos de retorno unificados.

Esta clase simplifica las peticiones requests inyectando automáticamente la dirección MAC en los endpoints y capturando excepciones comunes de red.

Códigos de Retorno (RC):

• 0: Éxito (Respuesta HTTP 2xx).

• 1: Error de red conocido (Timeout, Conexión, Error 4xx/5xx).

• 2: Error inesperado (Serialización, Excepciones críticas).

get(endpoint: str, params: Dict[str, Any] | None = None, headers: Dict[str, str] | None = None) → Tuple[int, Response | None]

Realiza una petición GET inyectando la MAC en la ruta.

post_json(endpoint: str, json_dict: Dict[str, Any], headers: Dict[str, str] | None = None) → Tuple[int, Response | None]

Envía un diccionario JSON mediante una petición POST.

class utils.request_util.ServerRealtimeConfig(method_psd: str, center_freq_hz: int, sample_rate_hz: int, rbw_hz: int, window: str, overlap: float, lna_gain: int, vga_gain: int, antenna_amp: bool, antenna_port: int, ppm_error: float, cooldown_request: float = 1.0, demodulation: str | None = None, filter: FilterConfig | None = None)

Configuración maestra de tiempo real enviada por el servidor.

Realiza validaciones automáticas mediante __post_init__ para asegurar que los parámetros solicitados por la nube sean compatibles con el hardware.

class utils.request_util.ZmqPairController(addr: str, is_server: bool = True, verbose: bool = False)

Controlador asíncrono para sockets ZeroMQ con flujo estricto request/reply.

Se utiliza para la comunicación Inter-Procesos (IPC) entre este código Python y el motor de procesamiento RF (C++/Rust/Python). Implementa un canal REQ/REP 1:1 con timeout y reciclado del socket para descartar cualquier reply tardío tras un timeout.

close()

Libera los recursos de ZeroMQ y limpia archivos temporales IPC.

async request(payload: dict) → dict | None

Envía un request y espera su reply dentro de la misma ventana de timeout.

async send_command(payload: dict)

Envía un comando JSON de forma asíncrona.

start()

Inicializa el contexto y prepara el socket.

async wait_for_data() → dict | None

Espera una respuesta y retorna None si vence timeout_ms.

Utilidades de Estado

Módulo de Estado del Dispositivo (Status Module).

Este módulo se encarga de recopilar métricas críticas del hardware, incluyendo: - Uso de CPU (por núcleo), RAM y Swap. - Ocupación de disco y temperatura del procesador. - Latencia de red (Ping) y extracción de logs recientes. - Metadatos de temporización (NTP, Calibración).

El módulo implementa mecanismos de reintento para garantizar la integridad de los datos ante posibles bloqueos de E/S del sistema operativo.

class utils.status_util.StatusDevice(disk_path: ~pathlib.Path = PosixPath('/'), logs_dir: ~pathlib.Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/sdr-spectrummonitoring-sensor/checkouts/latest/docs/Logs'), logger=<Logger utils.status_util (INFO)>)

Controlador para la consulta de métricas del sistema operativo.

Utiliza los sistemas de archivos virtuales /proc y /sys para obtener información del hardware sin necesidad de herramientas externas pesadas.

get_cpu_percent() → Dict[str, List[float]]

Calcula el uso de CPU leyendo /proc/stat.

Realiza dos lecturas de los contadores acumulativos (jiffies) con un intervalo de espera para calcular el diferencial de carga.

Devuelve:

Diccionario con la clave “cpu” y lista de porcentajes.

Tipo del valor devuelto:

Dict[str, List[float]]

get_disk() → dict

Calcula el espacio ocupado en disco mediante statvfs.

get_logs()

Extrae las últimas 10 líneas evitando la recursión de payloads.

get_ping_latency(ip: str) → Dict[str, float]

Mide la latencia de red hacia un host remoto.

Parámetros:

ip (str) – Dirección IP o dominio a pinguear.

Devuelve:

Tiempo de respuesta en milisegundos.

Tipo del valor devuelto:

Dict[str, float]

get_ram_swap_mb() → Dict[str, int]

Obtiene el uso actual de RAM y Swap desde /proc/meminfo.

Devuelve:

Megabytes en uso para RAM y Swap.

Tipo del valor devuelto:

Dict[str, int]

get_status_snapshot(delta_t_ms: int, last_kal_ms: int, last_ntp_ms: int, timestamp_ms: int, mac: str = '', ping_ip: str = '8.8.8.8') → Dict[str, Any]

Genera una captura completa del estado actual del sistema.

Coordina todas las subrutinas de recolección (CPU, RAM, Disco, Red, Logs) y empaqueta el resultado en el formato estricto definido por StatusPost.

Parámetros:

• delta_t_ms (int) – Latencia de procesamiento actual.

• last_kal_ms (int) – Tiempo desde la última calibración.

• last_ntp_ms (int) – Tiempo desde la última sincronía NTP.

• timestamp_ms (int) – Tiempo actual del sistema.

• mac (str) – Dirección MAC del sensor.

• ping_ip (str) – IP de destino para medir latencia de red.

Devuelve:

Snapshot completo serializado como diccionario.

Tipo del valor devuelto:

Dict[str, Any]

get_temp_c() → Dict[str, float]

Lee la temperatura del procesador desde thermal_zone.

Devuelve:

Temperatura en grados Celsius. -1.0 si falla.

Tipo del valor devuelto:

Dict[str, float]

get_total_disk() → dict

Calcula el tamaño total de la partición de disco.

get_total_ram_swap_mb() → Dict[str, int]

Obtiene las capacidades máximas de RAM y Swap.

class utils.status_util.StatusPost(mac: str, ram_mb: int, swap_mb: int, disk_mb: int, temp_c: float, total_ram_mb: int, total_swap_mb: int, total_disk_mb: int, delta_t_ms: int, ping_ms: float, timestamp_ms: int, last_kal_ms: int, last_ntp_ms: int, logs: str, cpu_loads: ~typing.List[float] = <factory>)

Estructura de datos para el reporte de estado del sensor.

Esta clase actúa como un DTO (Data Transfer Object) que valida y formatea las métricas para su envío a la API central. Maneja la conversión dinámica de listas de carga de CPU a claves individuales indexadas.

mac

Dirección MAC del dispositivo.

Type:

str

ram_mb

Memoria RAM en uso (MB).

Type:

int

swap_mb

Memoria Swap en uso (MB).

Type:

int

disk_mb

Espacio de disco en uso (MB).

Type:

int

temp_c

Temperatura actual en grados Celsius.

Type:

float

total_ram_mb

RAM total disponible.

Type:

int

total_swap_mb

Swap total disponible.

Type:

int

total_disk_mb

Capacidad total del disco.

Type:

int

delta_t_ms

Latencia de procesamiento interna.

Type:

int

ping_ms

Latencia de red hacia el servidor.

Type:

float

timestamp_ms

Tiempo Unix del reporte en milisegundos.

Type:

int

last_kal_ms

Timestamp de la última calibración.

Type:

int

last_ntp_ms

Timestamp de la última sincronización horaria.

Type:

int

logs

Fragmento de texto con los logs más recientes.

Type:

str

cpu_loads

Lista interna de cargas por núcleo.

Type:

List[float]

classmethod from_dict(data: Dict[str, Any])

Constructor personalizado para manejar claves de CPU dinámicas (cpu_0, cpu_1…).

Parámetros:

data (Dict[str, Any]) – Diccionario plano con métricas y claves cpu_n.

Devuelve:

Instancia de la clase con la lista cpu_loads poblada.

Tipo del valor devuelto:

StatusPost

to_dict() → Dict[str, Any]

Convierte la instancia a un diccionario plano para serialización JSON.

Aplana la lista cpu_loads de nuevo a claves individuales (cpu_0, cpu_1, etc.) para cumplir con el contrato de la API.

Devuelve:

Diccionario listo para ser enviado por HTTP.

Tipo del valor devuelto:

Dict[str, Any]