Poll Similarity Trajectory

from atomscale import Client
from atomscale.similarity import (
    iter_poll_trajectory,
    start_polling_trajectory_thread,
    aiter_poll_trajectory,
    start_polling_trajectory_task,
)

client = Client(api_key="YOUR_API_KEY")
source_id = "YOUR_DATA_OR_SAMPLE_ID"

def latest_similarity(df):
    """Extract the latest similarity score for deduplication."""
    if df.empty or "Similarity" not in df.columns:
        return None
    return df.iloc[-1]["Similarity"]


for result in iter_poll_trajectory(
    client,
    source_id=source_id,
    interval=5.0,
    last_n=10,
    distinct_by=latest_similarity,
    max_polls=20,
):
    if not result.empty and "Similarity" in result.columns:
        latest = result.iloc[-1]["Similarity"]
        print(f"Current similarity: {latest:.3f}")

def similarity_stabilized(df):
    """Stop when similarity exceeds 0.95."""
    if df.empty or "Similarity" not in df.columns:
        return False
    return df.iloc[-1]["Similarity"] > 0.95


for result in iter_poll_trajectory(
    client,
    source_id=source_id,
    interval=5.0,
    distinct_by=latest_similarity,
    until=similarity_stabilized,
):
    print(f"Similarity: {result.iloc[-1]['Similarity']:.3f}")

print("Growth stabilized!")

def on_trajectory_update(result):
    if not result.empty and "Similarity" in result.columns:
        similarity = result.iloc[-1]["Similarity"]
        if similarity > 0.95:
            print("Pattern stabilized - consider stopping growth")


stop_event = start_polling_trajectory_thread(
    client,
    source_id=source_id,
    interval=5.0,
    last_n=10,
    on_result=on_trajectory_update,
)

# Stop polling when done:
# stop_event.set()

import asyncio


async def monitor_trajectory():
    async for result in aiter_poll_trajectory(
        client,
        source_id=source_id,
        interval=5.0,
        last_n=10,
        max_polls=20,
    ):
        if not result.empty and "Similarity" in result.columns:
            print(f"Similarity: {result.iloc[-1]['Similarity']:.3f}")


asyncio.run(monitor_trajectory())

def handle_error(exc):
    print(f"Poll failed: {exc}")


for result in iter_poll_trajectory(
    client,
    source_id=source_id,
    interval=5.0,
    max_polls=10,
    on_error=handle_error,
):
    print(f"Received {len(result)} rows")