eyepopsdk.py

import logging
import os

from typing_extensions import deprecated

from eyepop import __version__
from eyepop.data.data_endpoint import DataEndpoint
from eyepop.data.data_syncify import SyncDataEndpoint
from eyepop.worker.worker_endpoint import WorkerEndpoint
from eyepop.worker.worker_syncify import SyncWorkerEndpoint

log = logging.getLogger('eyepop')
log.debug(f"EyePop SDK v{__version__} initializing...")

class EyePopSdk:
    """EyePop.ai Python SDK for Worker API."""

    @deprecated("use EyePopSdk.sync_worker() or EyePopSdk.async_worker() instead")
    @staticmethod
    def workerEndpoint(
            pop_id: str | None = None,
            session_uuid: str | None = None,
            secret_key: str | None = None,
            api_key: str | None = None,
            access_token: str | None = None,
            auto_start: bool = True,
            stop_jobs: bool = True,
            eyepop_url: str | None = None,
            job_queue_length: int = 1024,
            is_async: bool = False,
            is_local_mode: bool | None = None,
            request_tracer_max_buffer: int = 1204,
            dataset_uuid: str | None = None,
            pipeline_image: str | None = None,
            pipeline_version: str | None = None,
            **kwargs,
    ) -> WorkerEndpoint | SyncWorkerEndpoint:
        if is_async:
            return EyePopSdk.async_worker(
                pop_id=pop_id,
                session_uuid=session_uuid,
                secret_key=secret_key,
                api_key=api_key,
                access_token=access_token,
                auto_start=auto_start,
                stop_jobs=stop_jobs,
                eyepop_url=eyepop_url,
                job_queue_length=job_queue_length,
                is_local_mode=is_local_mode,
                request_tracer_max_buffer=request_tracer_max_buffer,
                dataset_uuid=dataset_uuid,
                pipeline_image=pipeline_image,
                pipeline_version=pipeline_version,
                **kwargs,
            )
        else:
            return EyePopSdk.sync_worker(
                pop_id=pop_id,
                session_uuid=session_uuid,
                secret_key=secret_key,
                api_key=api_key,
                access_token=access_token,
                auto_start=auto_start,
                stop_jobs=stop_jobs,
                eyepop_url=eyepop_url,
                job_queue_length=job_queue_length,
                is_local_mode=is_local_mode,
                request_tracer_max_buffer=request_tracer_max_buffer,
                dataset_uuid=dataset_uuid,
                pipeline_image=pipeline_image,
                pipeline_version=pipeline_version,
                **kwargs,
            )

    @staticmethod
    def sync_worker(
            pop_id: str | None = None,
            session_uuid: str | None = None,
            secret_key: str | None = None,
            api_key: str | None = None,
            access_token: str | None = None,
            auto_start: bool = True,
            stop_jobs: bool = True,
            eyepop_url: str | None = None,
            job_queue_length: int = 1024,
            is_local_mode: bool | None = None,
            request_tracer_max_buffer: int = 1204,
            dataset_uuid: str | None = None,
            pipeline_image: str | None = None,
            pipeline_version: str | None = None,
            **kwargs,
    ) -> SyncWorkerEndpoint:
        endpoint = EyePopSdk.async_worker(
            pop_id=pop_id,
            session_uuid=session_uuid,
            secret_key=secret_key,
            api_key=api_key,
            access_token=access_token,
            auto_start=auto_start,
            stop_jobs=stop_jobs,
            eyepop_url=eyepop_url,
            job_queue_length=job_queue_length,
            is_local_mode=is_local_mode,
            request_tracer_max_buffer=request_tracer_max_buffer,
            dataset_uuid=dataset_uuid,
            pipeline_image=pipeline_image,
            pipeline_version=pipeline_version,
            **kwargs,
        )
        return SyncWorkerEndpoint(endpoint)

    @staticmethod
    def async_worker(
            pop_id: str | None = None,
            session_uuid: str | None = None,
            secret_key: str | None = None,
            api_key: str | None = None,
            access_token: str | None = None,
            auto_start: bool = True,
            stop_jobs: bool = True,
            eyepop_url: str | None = None,
            job_queue_length: int = 1024,
            is_local_mode: bool | None = None,
            request_tracer_max_buffer: int = 1204,
            dataset_uuid: str | None = None,
            pipeline_image: str | None = None,
            pipeline_version: str | None = None,
            **kwargs,
    ) -> WorkerEndpoint:
        if is_local_mode is None:
            local_mode_env = os.getenv("EYEPOP_LOCAL_MODE", "")
            is_local_mode = local_mode_env.lower() in ("true", "yes")

        if pop_id is None:
            pop_id = os.getenv("EYEPOP_POP_ID", "transient")

        if session_uuid is None:
            session_uuid = os.getenv("EYEPOP_SESSION_UUID", None)

        has_any_auth_key = access_token is not None or secret_key is not None or api_key is not None

        if not has_any_auth_key and not is_local_mode:
            secret_key = os.getenv("EYEPOP_SECRET_KEY")
            api_key = os.getenv("EYEPOP_API_KEY")
            if secret_key is None and api_key is None:
                raise KeyError(
                    "At least one authentication method required: "
                    "EYEPOP_SECRET_KEY or EYEPOP_API_KEY or access_token"
                )

        if eyepop_url is None:
            eyepop_url = os.getenv("EYEPOP_URL")
            if eyepop_url is None:
                if is_local_mode:
                    eyepop_url = 'http://127.0.0.1:8080'
                elif api_key:
                    eyepop_url = "https://compute.eyepop.ai"
                else:
                    eyepop_url = "https://api.eyepop.ai"

        is_transient_pop = pop_id == "transient"

        if api_key and not is_transient_pop:
            raise ValueError(
                f"EYEPOP_API_KEY can only be used with transient pops. "
                f"Current pop_id: '{pop_id}'. Use EYEPOP_SECRET_KEY for named pops."
            )

        is_compute_url = eyepop_url and "https://compute" in eyepop_url.lower()
        if is_compute_url:
            if not api_key:
                raise ValueError(f"Compute API endpoint ({eyepop_url}) requires EYEPOP_API_KEY")
            if not is_transient_pop:
                raise ValueError(f"Compute API only supports transient mode. Current pop_id: '{pop_id}'")

        if is_local_mode and api_key is None:
            api_key = "<local api key>"

        assert eyepop_url
        log.debug(f"EyePop URL: {eyepop_url}")

        endpoint = WorkerEndpoint(
            secret_key=secret_key,
            access_token=access_token,
            api_key=api_key,
            pop_id=pop_id,
            session_uuid=session_uuid,
            auto_start=auto_start,
            stop_jobs=stop_jobs,
            eyepop_url=eyepop_url,
            job_queue_length=job_queue_length,
            request_tracer_max_buffer=request_tracer_max_buffer,
            dataset_uuid=dataset_uuid,
            pipeline_image=pipeline_image,
            pipeline_version=pipeline_version,
            session_opts=kwargs.get("session_opts"),
            session_headers=kwargs.get("session_headers"),
        )
        return endpoint

    """
    EyePop.ai Python SDK for Data API
    """

    @staticmethod
    def dataEndpoint(
        account_id: str | None = None,
        secret_key: str | None = None,
        access_token: str | None = None,
        api_key: str | None = None,
        eyepop_url: str | None = None,
        job_queue_length: int = 1024,
        is_async: bool = False,
        request_tracer_max_buffer: int = 1204,
        disable_ws: bool = True,
    ) -> DataEndpoint | SyncDataEndpoint:
        if access_token is None and secret_key is None and api_key is None:
            secret_key = os.getenv("EYEPOP_SECRET_KEY")
            api_key = os.getenv("EYEPOP_API_KEY")
            if secret_key is None and api_key is None:
                raise KeyError(
                    "At least one authentication method required: "
                    "EYEPOP_SECRET_KEY or EYEPOP_API_KEY or access_token"
                )

        if eyepop_url is None:
            eyepop_url = os.getenv("EYEPOP_URL")
            if eyepop_url is None:
                if api_key:
                    eyepop_url = "https://compute.eyepop.ai"
                else:
                    eyepop_url = "https://api.eyepop.ai"

        if account_id is None:
            account_id = os.getenv("EYEPOP_ACCOUNT_ID")

        endpoint = DataEndpoint(
            secret_key=secret_key,
            access_token=access_token,
            account_id=account_id,
            api_key=api_key,
            eyepop_url=eyepop_url,
            job_queue_length=job_queue_length,
            request_tracer_max_buffer=request_tracer_max_buffer,
            disable_ws=disable_ws,
        )

        if not is_async:
            return SyncDataEndpoint(endpoint)

        return endpoint

    from matplotlib.axes import Axes
    @staticmethod
    def plot(axes: Axes):
        from eyepop.visualize import EyePopPlot
        return EyePopPlot(axes)