Poll Time Series Updates

from atomscale import Client
from atomscale.timeseries import (
    iter_poll,
    aiter_poll,
    start_polling_thread,
    start_polling_task,
)

client = Client(api_key="YOUR_API_KEY")
data_id = "YOUR_STREAMING_DATA_ID"


def latest_timestamp(df):
    """Extract the last timestamp for deduplication."""
    if df.empty or "timestamp" not in df.columns:
        return None
    return df.iloc[-1]["timestamp"]

for idx, result in enumerate(
    iter_poll(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=10,
        distinct_by=latest_timestamp,
        max_polls=3,
    ),
    start=1,
):
    print(f"Poll {idx}: latest timestamp -> {latest_timestamp(result)}")
    print(result.tail())

collected = []


def on_result(result):
    print(f"Thread received {len(result)} rows")
    collected.append(result)


stop_event = start_polling_thread(
    client,
    data_id=data_id,
    interval=10.0,
    last_n=10,
    max_polls=5,
    distinct_by=latest_timestamp,
    on_result=on_result,
)

# Later, to stop polling:
# stop_event.set()

import asyncio


async def stream_updates():
    async for result in aiter_poll(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=10,
        distinct_by=latest_timestamp,
        max_polls=3,
    ):
        print(f"Async poll received {len(result)} rows")
        print(result.tail())


asyncio.run(stream_updates())

async def handle_async(result):
    print(f"Task handler received {len(result)} rows")


async def main():
    task = start_polling_task(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=5,
        max_polls=3,
        distinct_by=latest_timestamp,
        on_result=handle_async,
    )
    await task


asyncio.run(main())

def latest_timestamp(df):
    if df.empty or "timestamp" not in df.columns:
        return None
    return df.iloc[-1]["timestamp"]