omixom-data

Cliente Python de la Omixom Data API v3: lectura de series de mediciones y réplica incremental para mantener una copia siempre al día.

pip install omixom-data

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

Mantener una copia al día

El caso típico es replicar las mediciones de un grupo de equipos y seguirlas en el tiempo. El Feed encapsula todo el protocolo (bootstrap paginado, cursores, coherencia del snapshot) y lo entrega en batches: cada batch es una página de trabajo acotado que trae sus eventos y el estado que la deja atrás.

from omixom_data import Client, MeasurementDeleted

with Client(token="...") as client:
    feed = client.feed([30125, 30126])  # toda la historia de cada equipo

    for batch in feed.batches():  # bootstrap + cambios, hasta estar al día
        for event in batch.events:
            if isinstance(event, MeasurementDeleted):
                store.delete(event.station, event.module, event.time)
            else:
                store.upsert(event.station, event.module, event.time, event.value)
        save(batch.state.model_dump_json())  # checkpoint alineado a la página

El contrato es aplicar primero, guardar después (idealmente ambos en una transacción del store propio). Con ese orden, cortar el proceso en cualquier punto (incluso a mitad de un backfill de días) deja como peor caso una página re-aplicada, y aplicar eventos es idempotente por diseño: MeasurementUpserted deja la copia en ese valor, MeasurementDeleted la quita, re-aplicar no cambia nada. Cada batch persistido es progreso ganado; break entre batches es siempre seguro.

Para volver a sincronizar (el próximo poll o la próxima corrida del proceso):

from omixom_data import Client, FeedState

with Client(token="...") as client:
    feed = client.resume(FeedState.model_validate_json(saved))
    for batch in feed.batches():  # solo lo que falta desde el checkpoint
        apply_all(batch.events)
        save(batch.state.model_dump_json())

batches() es re-llamable: cada llamada avanza hasta quedar al día y la siguiente retoma desde ahí, así que el loop de un daemon es batches() + esperar el intervalo deseado. El avance es por módulo (el estado guarda un cursor por serie) y los requests van por estación: el bootstrap de varios equipos avanza round-robin para solapar sus límites de uso.

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 equipo se replica desde su fecha de instalación. Con date_from solo, la réplica sigue recibiendo lo nuevo; con ambas, la ventana queda cerrada pero las correcciones a datos de esa ventana siguen llegando. modules acota a esos módulos, de cualquiera de las estaciones dadas (una estación sin módulos seleccionados no genera requests); el conjunto replicado queda fijado al crear el feed, así que un sensor instalado después no se suma solo (ver abajo). categories filtra qué tipo de dato replicar; con ese filtro, un punto corregido hacia una categoría no seleccionada llega como borrado (salió de la vista replicada).

Sumar módulos a un feed existente

Un sensor instalado después de crear el feed se suma con add_modules, sobre un feed nuevo o retomado; sin lista de módulos entran todos los del equipo que falten:

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

for batch in feed.batches():  # el nuevo hace su bootstrap; el resto sigue donde estaba
    apply_all(batch.events)
    save(batch.state.model_dump_json())

Devuelve los ids agregados (los ya rastreados se omiten) y rechaza con ValueError un módulo que no pertenece al equipo. También sirve para sumar una estación nueva al feed. El alta queda persistida con el state del próximo batch, así que si el proceso corta antes, la próxima corrida tiene que volver a llamarlo.

Lecturas puntuales

Para consultas de una sola vez, sin réplica:

client.stations()  # equipos 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 camina todas las páginas por adentro repitiendo el cursor de la primera, así el resultado entero es un corte coherente de la base aunque haya escrituras concurrentes. Para materializar la serie, list(client.series(...)).

Debajo de eso está el acceso crudo página por página (series_page, changes_page), donde el manejo del cursor queda a cargo del caller: para paginar de forma coherente hay que repetir el request con date_from igual al next_from recibido y cursor igual al cursor recibido.

Errores y límites de uso

Los errores de la API llegan como excepciones tipadas bajo OmixomDataError: AuthenticationError, NotFoundError, InvalidRequestError, RateLimitedError y ServerError. Ante un 429 el cliente espera lo que indique Retry-After y reintenta solo; Client(..., wait_on_rate_limit=False) desactiva la espera y levanta RateLimitedError con el tiempo sugerido en retry_after.

Desarrollo

uv sync
uv run tox        # style + tests + cobertura