streaming_progress.py

"""VynFi Streaming — real-time progress monitoring for generation jobs."""

import os
import sys

import vynfi

client = vynfi.VynFi(api_key=os.environ["VYNFI_API_KEY"])

# Submit an async generation job
print("Submitting generation job...")
job = client.jobs.generate_config(
    config={
        "sector": "retail",
        "country": "US",
        "accountingFramework": "us_gaap",
        "rows": 5000,
        "companies": 5,
        "periods": 3,
        "periodLength": "monthly",
        "processModels": ["o2c", "p2p"],
        "exportFormat": "json",
        "fraudPacks": [],
        "fraudRate": 0.0,
    }
)
print(f"Job {job.id} submitted ({job.credits_reserved} credits reserved)")
print(f"Estimated duration: {job.estimated_duration_seconds}s")
print()

# Stream real-time progress events via SSE
print("Streaming progress:")
for event in client.jobs.stream(job.id):
    event_type = event.get("event", "unknown")
    data = event.get("data", {})

    if event_type == "progress":
        pct = data.get("progress", 0)
        stage = data.get("stage", "")
        bar = "=" * int(pct / 2) + ">" + " " * (50 - int(pct / 2))
        sys.stdout.write(f"\r  [{bar}] {pct}% — {stage}")
        sys.stdout.flush()

    elif event_type == "complete":
        print(f"\n\nJob completed!")
        print(f"  Status: {data.get('status')}")
        print(f"  Credits used: {data.get('credits_used')}")
        print(f"  Rows generated: {data.get('rows_generated')}")
        break

    elif event_type == "error":
        print(f"\n\nJob failed: {data.get('error')}")
        sys.exit(1)

# Download the results
print("\nDownloading archive...")
archive = client.jobs.download_archive(job.id)
print(archive)

summary = archive.summary()
for category, count in summary["categories"].items():
    print(f"  {category}: {count} files")