Odyssey Client

Odyssey(api_key: str, **kwargs)

from odyssey import Odyssey

client = Odyssey(api_key="ody_your_api_key_here")

async def connect(
    on_connected: Callable[[], None] | None = None,
    on_disconnected: Callable[[], None] | None = None,
    on_video_frame: Callable[[VideoFrame], None] | None = None,
    on_stream_started: Callable[[str], None] | None = None,
    on_stream_ended: Callable[[], None] | None = None,
    on_interact_acknowledged: Callable[[str], None] | None = None,
    on_stream_error: Callable[[str, str], None] | None = None,
    on_error: Callable[[Exception, bool], None] | None = None,
    on_status_change: Callable[[ConnectionStatus, str | None], None] | None = None,
) -> None

try:
    await client.connect(
        on_video_frame=lambda frame: process_frame(frame),
        on_stream_error=lambda reason, msg: print(f"Stream error: {reason} - {msg}"),
        on_status_change=lambda status, msg: print(f"Status: {status.value}"),
    )
except OdysseyAuthError:
    print("Invalid API key")
except OdysseyConnectionError as e:
    print(f"Connection failed: {e}")

async def disconnect() -> None

await client.disconnect()

async def start_stream(
    prompt: str = "",
    portrait: bool = True,
    image: str | bytes | Image.Image | np.ndarray | None = None,
    image_path: str | None = None  # deprecated
) -> str

try:
    stream_id = await client.start_stream("A cat", portrait=True)
    print(f"Stream started: {stream_id}")
except OdysseyStreamError as e:
    print(f"Failed to start stream: {e}")

# Image-to-video examples
await client.connect(on_video_frame=process_frame)

# Using a file path
stream_id = await client.start_stream(
    prompt="A cat",
    portrait=False,
    image="/path/to/image.jpg"
)

# Using PIL Image
from PIL import Image
pil_image = Image.open("/path/to/image.jpg")
stream_id = await client.start_stream(prompt="A cat", image=pil_image)

# Using bytes
with open("/path/to/image.jpg", "rb") as f:
    image_bytes = f.read()
stream_id = await client.start_stream(prompt="A cat", image=image_bytes)

async def interact(prompt: str) -> str

try:
    ack_prompt = await client.interact("Pet the cat")
    print(f"Interaction acknowledged: {ack_prompt}")
except OdysseyStreamError as e:
    print(f"Failed to interact: {e}")

async def end_stream() -> None

await client.end_stream()

async def get_recording(stream_id: str) -> Recording

recording = await client.get_recording("stream-123")
if recording.video_url:
    print(f"Video: {recording.video_url}")
    print(f"Duration: {recording.duration_seconds}s")

async def list_stream_recordings(
    limit: int | None = None,
    offset: int | None = None
) -> StreamRecordingsList

result = await client.list_stream_recordings(limit=10)
for rec in result.recordings:
    print(f"{rec.stream_id}: {rec.duration_seconds}s ({rec.width}x{rec.height})")
print(f"Total: {result.total}")

async def simulate(
    script: list[dict],
    portrait: bool = True
) -> SimulationJobInfo

job = await client.simulate(
    script=[
        {"timestamp_ms": 0, "start": {"prompt": "A cat sitting on a windowsill"}},
        {"timestamp_ms": 3000, "interact": {"prompt": "The cat stretches"}},
        {"timestamp_ms": 6000, "interact": {"prompt": "The cat yawns"}},
        {"timestamp_ms": 9000, "end": {}}
    ],
    portrait=True
)
print(f"Simulation started: {job.job_id}")

async def get_simulate_status(job_id: str) -> SimulationJobDetail

status = await client.get_simulate_status(job.job_id)
print(f"Status: {status.status}")
if status.status == "completed":
    for stream in status.streams:
        print(f"Stream: {stream.stream_id}")

async def list_simulations(
    limit: int | None = None,
    offset: int | None = None
) -> SimulationJobsList

result = await client.list_simulations(limit=10)
for sim in result.jobs:
    print(f"{sim.job_id}: {sim.status}")
print(f"Total: {result.total}")

async def cancel_simulation(job_id: str) -> None

await client.cancel_simulation(job.job_id)
print("Simulation cancelled")

@property
def is_connected(self) -> bool

@property
def current_status(self) -> ConnectionStatus

@property
def current_session_id(self) -> str | None