omixom-data

Cliente Python de la Omixom Data API v3 para la consulta de series de mediciones y la réplica incremental que mantiene una copia local sincronizada.

pip install omixom-data

Requiere Python 3.11 o superior y un token de acceso de la red Omixom.

Réplica incremental

El caso de uso principal es replicar las mediciones de un conjunto de estaciones y mantener esa réplica actualizada en el tiempo. El Feed encapsula el protocolo completo (bootstrap paginado, cursores, coherencia del snapshot) y entrega el resultado en batches, donde cada batch contiene una página de eventos junto con el estado que queda tras aplicarla.

from omixom_data import Client, MeasurementDeleted, MeasurementUpserted, ModuleMoved

with Client(token="...") as client:
    feed = client.feed([30125, 30126])  # la historia completa de cada estación

    for batch in feed.batches():  # bootstrap y cambios pendientes
        for event in batch.events:
            match event:
                case MeasurementUpserted():
                    store.upsert(event.station, event.module, event.time, event.value)
                case MeasurementDeleted():
                    store.delete(event.station, event.module, event.time)
                case ModuleMoved():
                    store.move(event.module, event.previous_station, event.station)
        save(batch.state.model_dump_json())  # checkpoint alineado a la página

El contrato es aplicar primero y persistir el estado después, idealmente ambos dentro de una transacción del almacenamiento propio. Con ese orden, una interrupción en cualquier punto (incluso durante un backfill extenso) tiene como peor caso una página re-aplicada, y aplicar eventos es idempotente (MeasurementUpserted fija el valor, MeasurementDeleted lo elimina, ModuleMoved pasa las mediciones del módulo a otra estación y re-aplicarlos no altera el resultado). Cada batch persistido consolida el avance; interrumpir el recorrido entre batches es siempre seguro.

Sincronización continua

Un proceso de sincronización continua reúne las piezas anteriores, retomando el estado persistido si existe (client.resume), procesando los batches pendientes y esperando el intervalo de consulta. El estado se guarda serializado en el mismo store que recibe los datos, y cada batch se confirma en una única transacción que abarca los eventos y el checkpoint.

import time

from omixom_data import (
    Client,
    FeedEvent,
    FeedState,
    MeasurementDeleted,
    MeasurementUpserted,
    ModuleMoved,
)

POLL_INTERVAL = 300  # segundos
STATIONS = [30125, 30126]


def apply(event: FeedEvent) -> None:
    match event:
        case MeasurementUpserted():
            store.upsert(event.station, event.module, event.time, event.value)
        case MeasurementDeleted():
            store.delete(event.station, event.module, event.time)
        case ModuleMoved():
            store.move(event.module, event.previous_station, event.station)


with Client(token="...") as client:
    saved = store.load_state()  # el JSON guardado en la corrida anterior, o None
    if saved is not None:
        feed = client.resume(FeedState.model_validate_json(saved))
    else:
        feed = client.feed(STATIONS)

    while True:
        for batch in feed.batches():  # solo lo pendiente desde el último checkpoint
            with store.transaction():  # eventos y checkpoint juntos, todo o nada
                for event in batch.events:
                    apply(event)
                store.save_state(batch.state.model_dump_json())
        time.sleep(POLL_INTERVAL)

Con esa transacción por batch el checkpoint queda exactamente alineado con lo aplicado, porque una falla a mitad de un batch revierte todo y la corrida siguiente retoma desde el checkpoint anterior, sin estados intermedios. Si el estado no puede vivir en el store y se guarda aparte (por ejemplo, en un archivo JSON local, escrito de forma atómica vía un temporal y Path.replace), la garantía se relaja, y una interrupción entre aplicar y guardar implica re-aplicar la última página en la corrida siguiente, lo que no altera el resultado por la idempotencia de los eventos.

batches() puede invocarse repetidamente sobre el mismo feed, ya que cada llamada avanza hasta alcanzar el estado actual y la siguiente retoma desde ese punto, por lo que el ciclo del proceso es invocarlo y esperar el intervalo, manteniendo el Client abierto para reutilizar las conexiones. El bootstrap se pide por estación, con varias estaciones avanzando round-robin para distribuir los límites de uso, mientras que el seguimiento de cambios usa un único cursor para todo el feed y pide en cada request los cambios de la lista explícita de módulos seguidos, de todas sus estaciones (POST /api/v3/measurements/changes), de modo que un poll sin atraso cuesta un solo request. El estado que guardó una versión anterior del cliente (version 1, con un cursor por módulo) se sigue leyendo, ya que FeedState.model_validate_json lo migra al formato actual al cargarlo. Las versiones anteriores del cliente seguían los cambios con una ruta por estación que la API ya no ofrece, así que dejan de funcionar y hay que actualizarlas.

Acotar el rango

from datetime import UTC, datetime

feed = client.feed([30125], date_from=datetime(2023, 1, 1, tzinfo=UTC))  # desde 2023
feed = client.feed(
    [30125],
    date_from=datetime(2023, 1, 1, tzinfo=UTC),
    date_to=datetime(2024, 1, 1, tzinfo=UTC),  # 2023 completo
)

Sin fechas, cada estación se replica desde su fecha de instalación. Con date_from únicamente, la réplica continúa recibiendo los datos nuevos; con ambas fechas la ventana queda cerrada, aunque las correcciones sobre datos de esa ventana siguen llegando. modules restringe la réplica a esos módulos, de cualquiera de las estaciones indicadas (una estación sin módulos seleccionados no genera requests); el conjunto replicado queda fijado al crear el feed, por lo que un sensor instalado con posterioridad no se incorpora automáticamente (ver la sección siguiente). categories filtra el tipo de dato a replicar; con ese filtro, un punto corregido hacia una categoría no seleccionada se recibe como borrado, dado que salió de la vista replicada.

Incorporar módulos a un feed existente

Un módulo instalado después de crear el feed se incorpora con add_modules, sobre un feed nuevo o retomado; sin lista de módulos se incorporan todos los de la estación que aún no estén rastreados.

feed = client.resume(FeedState.model_validate_json(saved))
feed.add_modules(30125)  # o add_modules(30125, [4812]) para un módulo puntual

for batch in feed.batches():  # el módulo nuevo hace su bootstrap; el resto continúa
    apply_all(batch.events)
    save(batch.state.model_dump_json())

El método devuelve los ids incorporados (los ya rastreados se omiten) y levanta ValueError ante un módulo que no pertenece a la estación. Sirve también para incorporar una estación nueva al feed. El alta queda persistida con el state del batch siguiente; si el proceso se interrumpe antes, la corrida siguiente debe repetir la llamada.

El seguimiento pide solo los módulos que el feed rastrea y que ya completaron su bootstrap. Un módulo que el feed no rastrea (un sensor nuevo, una estación que no está en el feed o un módulo fuera de modules) entra a la réplica recién con su bootstrap, que trae su historia completa; a partir de ahí sus cambios siguen llegando con el resto.

Módulos fuera del alcance

Un módulo seguido sale del alcance del token cuando la cuenta pierde el acceso a su estación, cuando el módulo se deshabilita o cuando deja de existir. La API no distingue esos casos y lo devuelve en missing. El feed no se corta: el módulo queda stale, su copia se congela en el último cursor en que estuvo cubierto y el resto de los módulos sigue avanzando. Cada batch lista en batch.stale los módulos del feed que quedaron fuera del alcance, y state.modules[id].stale lo registra en el estado.

for batch in feed.batches():
    apply_all(batch.events)
    save(batch.state.model_dump_json())
    if batch.stale:
        log.warning("módulos fuera del alcance: %s", batch.stale)

El seguimiento sigue pidiendo los módulos stale, así que la API avisa cuando vuelven sin requests de más. La excepción es un módulo que no terminó su bootstrap: cada llamada a batches() consulta el detalle de su estación para saber si volvió. Cuando un módulo vuelve, el batch de fase recovery trae sus cambios desde el cursor en que quedó, junto con los demás que volvieron, hasta alcanzar el cursor del feed. Ahí vuelve al seguimiento común. Lo mismo vale para un módulo cuya estación sale del alcance antes de terminar su bootstrap: el bootstrap espera y sigue desde donde quedó.

Un módulo stale no tiene tope de tiempo y el feed nunca lo da de baja solo. Si el consumidor decide que no va a volver, feed.remove_modules([id]) deja de seguirlo, y la copia de ese módulo queda a cargo del consumidor.

Módulos que cambian de estación

Las mediciones pertenecen al módulo, así que un módulo que pasa a otra estación de la cuenta se lleva su historia, y la API informa en cada cambio la estación actual del módulo. Cuando esa estación difiere de la registrada, el feed emite ModuleMoved antes del primer cambio de la estación nueva y registra el módulo ahí, sin cortar el seguimiento ni repetir el bootstrap. El consumidor pasa la copia del módulo de previous_station a station; con un store indexado por módulo el evento no requiere más que actualizar la estación.

El aviso llega con el primer cambio del módulo después del traslado: hasta entonces la copia sigue en la estación anterior, que es igual de correcta porque no hubo cambios. Un módulo que se traslada antes de terminar su bootstrap queda stale, porque el detalle de su estación ya no lo lista. Para retomarlo, feed.remove_modules([id]) y feed.add_modules(estación_nueva, [id]), que repite su bootstrap en la estación nueva; la copia parcial de la anterior queda a cargo del consumidor.

Lecturas puntuales

Las lecturas puntuales resuelven consultas únicas, sin réplica de por medio.

client.stations()  # estaciones accesibles con el token
client.station(30125)  # fecha de instalación y módulos

for point in client.series(  # la serie completa de una ventana, en streaming
    30125,
    date_from=datetime(2024, 1, 1, tzinfo=UTC),
    date_to=datetime(2024, 2, 1, tzinfo=UTC),
):
    print(point.module, point.time, point.value)

series recorre internamente todas las páginas repitiendo el cursor de la primera, de modo que el resultado completo es un corte coherente de la base aun ante escrituras concurrentes. Para materializar la serie completa, list(client.series(...)).

Por debajo existe el acceso crudo página por página (series_page de una estación y account_changes_page de una lista de módulos de cualquier estación), donde el manejo del cursor queda a cargo del caller. Para paginar la serie de forma coherente se repite el request con date_from igual al next_from recibido y cursor igual al cursor recibido. Para los cambios se repite el request con el cursor recibido mientras has_more venga en true, y un módulo en missing no avanza con ese cursor.

Errores y límites de uso

Los errores de la API se levantan como excepciones tipadas bajo OmixomDataError (AuthenticationError, NotFoundError, InvalidRequestError, RateLimitedError y ServerError). Ante un 429 el cliente espera el tiempo indicado en Retry-After y reintenta automáticamente; Client(..., wait_on_rate_limit=False) desactiva esa espera y levanta RateLimitedError con el tiempo sugerido en retry_after.