octoai package

Subpackages

• octoai.clients package
  • Submodules
  • octoai.clients.asset_orch module
    • Asset
    • AssetOrchestrator
      • AssetOrchestrator.create()
      • AssetOrchestrator.delete()
      • AssetOrchestrator.get()
      • AssetOrchestrator.list()
      • AssetOrchestrator.wait_for_ready()
    • CheckpointData
    • FileData
    • LoraData
    • ModelData
    • TextualInversionData
    • VAEData
  • octoai.clients.fine_tuning module
    • FineTuningClient
      • FineTuningClient.cancel()
      • FineTuningClient.create()
      • FineTuningClient.delete()
      • FineTuningClient.get()
      • FineTuningClient.list()
  • octoai.clients.image_gen module
    • Engine
      • Engine.CONTROLNET_SD
      • Engine.CONTROLNET_SDXL
      • Engine.SD
      • Engine.SDXL
      • Engine.SSD
      • Engine.to_name()
    • ImageGenerateResponse
      • ImageGenerateResponse.images
      • ImageGenerateResponse.removed_for_safety
    • ImageGenerator
      • ImageGenerator.generate()
  • Module contents

Submodules

octoai.chat module

Entities and APIs for working with text generation models.

Instead of using these classes directly, developers should use the octoai.client.Client class. For example:

client = octoai.client.Client() completion = client.chat.completions.create(…)

class octoai.chat.Chat(client: Client)

Bases: object

Chat API for text generation models.

completions: Completions

class octoai.chat.ChatCompletion(*, id: str, choices: List[Choice], created: int, model: str, object: typing_extensions.Literal[chat.completion, chat.completion.chunk] | None = None, system_fingerprint: str | None = None, usage: CompletionUsage | None = None)

Bases: BaseModel

A text completion response.

choices: List[Choice]

created: int

id: str

model: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'choices': FieldInfo(annotation=List[octoai.chat.Choice], required=True), 'created': FieldInfo(annotation=int, required=True), 'id': FieldInfo(annotation=str, required=True), 'model': FieldInfo(annotation=str, required=True), 'object': FieldInfo(annotation=Union[Literal['chat.completion', 'chat.completion.chunk'], NoneType], required=False), 'system_fingerprint': FieldInfo(annotation=Union[str, NoneType], required=False), 'usage': FieldInfo(annotation=Union[CompletionUsage, NoneType], required=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

object: typing_extensions.Literal[chat.completion, chat.completion.chunk] | None

system_fingerprint: str | None

usage: CompletionUsage | None

class octoai.chat.Choice(*, index: int, message: ChatMessage = None, delta: ChoiceDelta = None, finish_reason: typing_extensions.Literal[stop, length, tool_calls, content_filter, function_call] | None = None)

Bases: BaseModel

A single choice in a text completion response.

delta: ChoiceDelta

finish_reason: typing_extensions.Literal[stop, length, tool_calls, content_filter, function_call] | None

index: int

message: ChatMessage

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'delta': FieldInfo(annotation=ChoiceDelta, required=False), 'finish_reason': FieldInfo(annotation=Union[Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'], NoneType], required=False), 'index': FieldInfo(annotation=int, required=True), 'message': FieldInfo(annotation=ChatMessage, required=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class octoai.chat.ChoiceDelta(*, content: str | None = None, role: typing_extensions.Literal[system, user, assistant, tool] | None = None)

Bases: BaseModel

Contents for streaming text completion responses.

content: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'content': FieldInfo(annotation=Union[str, NoneType], required=False), 'role': FieldInfo(annotation=Union[Literal['system', 'user', 'assistant', 'tool'], NoneType], required=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

role: typing_extensions.Literal[system, user, assistant, tool] | None

class octoai.chat.CompletionUsage(*, completion_tokens: int, prompt_tokens: int, total_tokens: int)

Bases: BaseModel

Usage statistics for a text completion response.

completion_tokens: int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'completion_tokens': FieldInfo(annotation=int, required=True), 'prompt_tokens': FieldInfo(annotation=int, required=True), 'total_tokens': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

prompt_tokens: int

total_tokens: int

class octoai.chat.Completions(client: Client)

Bases: object

Text completions API.

client: Client

create(*, messages: List[ChatMessage], model: str | TextModel, frequency_penalty: float | None = 0.0, max_tokens: int | None = None, presence_penalty: float | None = 0.0, response_format: ChatCompletionResponseFormat | None = None, stop: str | None = None, stream: bool | None = False, temperature: float | None = 1.0, top_p: float | None = 1.0) → ChatCompletion | Iterable[ChatCompletion]

Create a chat completion with a text generation model.

Parameters:

• messages – Required. A list of messages to use as context for the completion.

• model – Required. The model to use for the completion. Supported models are listed in the octoai.chat.TextModel enum.

• frequency_penalty – Positive values make it less likely that the model repeats tokens several times in the completion. Valid values are between -2.0 and 2.0.

• max_tokens – The maximum number of tokens to generate.

• presence_penalty – Positive values make it less likely that the model repeats tokens in the completion. Valid values are between -2.0 and 2.0.

• response_format – An object specifying the format that the model must output.

• stop – A list of sequences where the model stops generating tokens.

• stream – Whether to return a generator that yields partial message deltas as they become available, instead of waiting to return the entire response.

• temperature – Sampling temperature. A value between 0 and 2. Higher values make the model more creative by sampling less likely tokens.

• top_p – The cumulative probability of the most likely tokens to use. Use temperature or top_p but not both.

endpoint: str = 'https://text.octoai.run/v1/chat/completions'

class octoai.chat.TextModel(value)

Bases: str, Enum

List of available text models.

CODELLAMA_13B_INSTRUCT = 'codellama-13b-instruct'

CODELLAMA_34B_INSTRUCT = 'codellama-34b-instruct'

CODELLAMA_70B_INSTRUCT = 'codellama-70b-instruct'

CODELLAMA_7B_INSTRUCT = 'codellama-7b-instruct'

LLAMA_2_13B_CHAT = 'llama-2-13b-chat'

LLAMA_2_70B_CHAT = 'llama-2-70b-chat'

MISTRAL_7B_INSTRUCT = 'mistral-7b-instruct'

MIXTRAL_8X7B_INSTRUCT = 'mixtral-8x7b-instruct'

to_name()

Return the name of the model.

octoai.chat.get_model_list() → List[str]

Return a list of available text models.

octoai.client module

Client used to infer from endpoints.

class octoai.client.Client(token: str | None = None, config_path: str | None = None, timeout: float = 900.0, verify_ssl: bool = True, secure_link: bool = False)

Bases: object

A class that allows inferences from existing endpoints.

Parameters:

• token (str, optional) – api token, defaults to None

• config_path (str, optional) – path to ‘/.octoai/config.yaml’. Installed in ~, defaults to None and will check home path

• timeout (float, optional) – seconds before request times out, defaults to 900.0

• verify_ssl (bool, optional) – verify SSL certificates, defaults to True

• secure_link (bool, optional) – use secure link for inferences, defaults to False

Sets various headers. Gets auth token from environment if none is provided.

get_future_result(future: InferenceFuture) → Dict[str, Any] | None

Return the result of an inference.

This class will raise any errors if the status code is >= 400.

Parameters:

future (InferenceFuture) – Future from Client.infer_async()

Raises:

OctoAIClientError

Raises:

OctoAIServerError

Returns:

None if future is not ready, or dict of the response.

Return type:

Dict[str, Any], optional

health_check(endpoint_url: str, timeout: float = 900.0) → int

Check health of an endpoint using a get request. Try until timeout.

Parameters:

• endpoint_url (str) – URL as a str starting with https permitting get requests.

• timeout (float) – Seconds before request times out, defaults to 900.

Returns:

status code from get request. 200 means ready.

Return type:

int

infer(endpoint_url: str, inputs: Mapping[str, Any]) → Mapping[str, Any]

Send a request to the given endpoint URL with inputs as request body.

Parameters:

• endpoint_url (str) – target endpoint

• inputs (Mapping[str, Any]) – inputs for target endpoint

Raises:

• OctoAIServerError – server-side failures (unreachable, etc)

• OctoAIClientError – client-side failures (throttling, unset token)

Returns:

outputs from endpoint

Return type:

Mapping[str, Any]

infer_async(endpoint_url: str, inputs: Mapping[str, Any]) → InferenceFuture

Execute an inference in the background on the server.

InferenceFuture allows you to query status and get results once it’s ready.

Parameters:

• endpoint_url (str) – url to post inference request

• inputs (Mapping[str, Any]) – inputs to send to endpoint

infer_stream(endpoint_url: str, inputs: Mapping[str, Any], map_fn: Callable | None = None) → Iterator[dict]

Stream text event response body for supporting endpoints.

This is an alternative to loading all response body into memory at once.

Parameters:

• endpoint_url (str) – target endpoint

• inputs (Callable) – inputs for target endpoint such as a prompt and other parameters

• inputs – function to map over each response

Returns:

Yields a dict that contains the server response data.

Return type:

Iterator[dict]

is_future_ready(future: InferenceFuture) → bool

Return whether the future’s result has been computed.

This class will raise any errors if the status code is >= 400.

Parameters:

future (InferenceFuture) – Future from Client.infer_async()

Raises:

OctoAIClientError

Raises:

OctoAIServerError

Returns:

True if able to use Client.get_future_result()

class octoai.client.InferenceFuture(*, response_id: str, poll_url: str)

Bases: BaseModel

Response class for endpoints that support server side async inferences.

Parameters:

• response_id (str) – Unique identifier for inference

• poll_url (str) – URL to poll status of inference.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'poll_url': FieldInfo(annotation=str, required=True), 'response_id': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

poll_url: str

response_id: str

octoai.errors module

OctoAI errors.

exception octoai.errors.OctoAIAssetReadyTimeoutError(message: str, caused_by: Exception | None = None)

Bases: OctoAIError

Raise when asset was not ready to use by the timeout deadline.

exception octoai.errors.OctoAIClientError(message: str, caused_by: Exception | None = None)

Bases: OctoAIError

Raise when the client returns a 4xx error.

exception octoai.errors.OctoAIError(message: str, caused_by: Exception | None = None)

Bases: Exception

Base class for all OctoAI errors.

Parameters:

• message (str) – error message

• caused_by (:class:Exception, optional) – Exception source

exception octoai.errors.OctoAIServerError(message: str, caused_by: Exception | None = None)

Bases: OctoAIError

Raise when the server returns a 5xx error.

exception octoai.errors.OctoAIValidationError(message: str, caused_by: Exception | None = None)

Bases: OctoAIError

Raise when inputs are unable to be validated.

octoai.server module

Server for OctoAI endpoints created with the octoai CLI.

Developers that want to create an endpoint should not use this module directly. Instead, they should use the octoai command-line interface, which directs them to implement the octoai.service.Service class and use the octoai CLI to help build and deploy their endpoint.

class octoai.server.InferenceRequest(response_pipe: Connection, method: str, inputs: Any)

Bases: NamedTuple

Class for returning inference results.

inputs: Any

Alias for field number 2

method: str

Alias for field number 1

response_pipe: Connection

Alias for field number 0

class octoai.server.InferenceResponse(inference_time_ms: float, outputs: Any)

Bases: NamedTuple

Class for returning inference results.

inference_time_ms: float

Alias for field number 0

outputs: Any

Alias for field number 1

class octoai.server.Server(service: Service, async_enable: bool = True)

Bases: object

Server for OctoAI endpoints created with the octoai CLI.

Developers that want to create an endpoint should not use this class directly. Instead, they should use the octoai command-line interface, which directs them to implement the octoai.service.Service class and use the octoai CLI to help build and deploy their endpoint.

class State(value)

Bases: Enum

Describes the states of Server.

LAUNCH_PREDICT_LOOP = 'LAUNCH_PREDICT_LOOP'

RUNNING = 'RUNNING'

SETUP_SERVICE = 'SETUP_SERVICE'

SHUTTING_DOWN = 'SHUTTING_DOWN'

STOPPED = 'STOPPED'

UNINITIALIZED = 'UNINITIALIZED'

get_api_schema() → Dict[str, Any]

Return the Open API schema for the underlying service.

get_usage_examples(format: DocumentationFormat) → MustacheGenerationResult

Return the Mustache generation result for the underlying service.

property is_running

True when this server instance can serve a request.

prepare_for_serving()

Context manager that should surround all serving.

This is intended to be used as an ASGI application’s lifetime handler.

run(port: int, timeout_keep_alive: int)

Run the server exposing the underlying service.

property state

Get the status of this server.

store_assets() → None

Run service store assets.

class octoai.server.ServiceMethod(value)

Bases: Enum

Class for distinguishing route implementations in the request queue.

INFER_FORM_DATA = 'infer_form_data'

INFER_JSON = 'infer'

octoai.server.is_store_assets_needed(ctx) → bool

Check if store-assets step is required.

octoai.server.load_service(module_name: str, class_name: str | None = None) → Service

Load a class from service implementation.

octoai.server.maybe_set_volume_environment_variables(ctx) → None

Set volume environment variables if needed.

octoai.service module

Contains the base interface that OctoAI endpoints should implement.

Developers that want to create an endpoint should implement the Service class in this module as directed by the octoai command-line interface.

class octoai.service.ResponseAnalytics(*, inference_time_ms: float, performance_time_ms: float)

Bases: BaseModel

Additional analytics metadata.

inference_time_ms: float

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'inference_time_ms': FieldInfo(annotation=float, required=True, description='Inference execution time (without pauses)'), 'performance_time_ms': FieldInfo(annotation=float, required=True, description='Inference execution time (including pauses)')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

performance_time_ms: float

class octoai.service.Service

Bases: ABC

The base interface that OctoAI endpoints should implement.

Developers that want to create an endpoint should implement this class as directed by the octoai command-line interface.

abstract infer(**kwargs: Any) → Any

Perform inference.

on_server_shutdown() → None

Perform any necessary cleanup when the server stops.

on_server_startup() → None

Perform any necessary initialization when the server starts.

This method is separate from setup() because setup() can be called outside the serving context to include weights when building the image.

setup() → None

Perform service initialization.

A common operation to include here is loading weights and making those available to the infer() method in a member variable.

store_assets() → None

Download model assets.

octoai.service.find_additional_endpoints(service: Service) → Mapping[str, Tuple[Callable, str]]

Find additional methods that should be exposed as endpoints.

octoai.service.implements_form_data(service: Service) → bool

Check if infer_form_data() is implemented.

octoai.service.inspect_input_parameters(parameters: Mapping[str, Parameter]) → Type[BaseModel]

Create Pydnatic input mode from the provided signature.

octoai.service.inspect_input_types(method: Callable) → Type[BaseModel]

Create Pydantic input model from inference method signature.

octoai.service.inspect_output_types(method: Callable) → Type[BaseModel]

Create Pydantic output model from inference method signature.

octoai.service.path(path: str)

Specify the path for a service method.

octoai.service.transform_form_data_signature(service: Service) → Signature

Add FastAPI Form() defaults to ‘str’ fields and check compatibility.

octoai.service.volume_path() → str

Get mounted volume path in docker.

Returns:

Docker path.

octoai.types module

Type definitions to help communicate with endpoints.

These type definitions help with routine format conversions that are necessary when transferring binary files (such as images or audio) over HTTP. These type definitions can be useful both when creating endpoints (implementing octoai.service.Service as directed by the octoai command-line interface) and when communicating with live endpoints using the client SDK.

class octoai.types.Audio(data: bytes | MediaUrl)

Bases: Media

Audio helpers for models that accept or return audio.

The Audio class is a wrapper for either binary audio content or a URL reference to an audio file. When an instance contains binary audio data, it is stored and transferred over HTTP as a base64-encoded string. When an instance contains a URL reference, the URL is stored and transferred over HTTP as a string. The URL representation can be more advantageous when the audio file is already hosted on a server, as it avoids the overhead of base64 encoding and decoding.

The Audio class contains additional convenience methods to read and write audio files in common formats.

data: bytes | MediaUrl

classmethod from_numpy(data: _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], sample_rate: int, format='WAV') → Audio

Create from a numpy array.

The first dimension of the array should represent audio frames (samples), while the second dimension should represent audio channels.

A file format and a sample rate are needed since the audio data is serialized to a binary audio file. The default format is “WAV”, and the sample rate is required.

Parameters:

• data (ArrayLike) – numpy array with audio data (frames x channels)

• sample_rate (int) – samples per second taken to create signal

• format (str, optional) – target format, defaults to “WAV”

Returns:

Audio object

Return type:

Audio

is_valid()

Check if this is a valid audio.

Returns:

True if it’s valid, false if not.

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'data': FieldInfo(annotation=Union[bytes, MediaUrl], required=True, description='URL reference or base64-encoded data')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

to_numpy() → Tuple[_SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], int]

Convert to numpy array.

Returns:

numpy array representation (frames x channels)

Return type:

Tuple[ArrayLike, int]

class octoai.types.File(file: BinaryIO, *, size: int | None = None, filename: str | None = None, headers: Headers | None = None)

Bases: UploadFile

File class for file uploads in form-data endpoints.

file: BinaryIO

filename: str | None

headers: Headers

size: int | None

class octoai.types.Image(data: bytes | MediaUrl)

Bases: Media

Image helpers for models that accept or return images.

The Image class is a wrapper for either binary image content or a URL reference to an image file. When an instance contains binary image data, it is stored and transferred over HTTP as a base64-encoded string. When an instance contains a URL reference, the URL is stored and transferred over HTTP as a string. The URL representation can be more advantageous when the image file is already hosted on a server, as it avoids the overhead of base64 encoding and decoding.

The Image class contains additional convenience methods to read and write image files in common formats.

data: bytes | MediaUrl

classmethod from_pil(image_pil: <module 'PIL.Image' from '/home/runner/work/nimbus/nimbus/octoai-python-sdk/.venv/lib/python3.9/site-packages/PIL/Image.py'>, format='JPEG') → Image

Create from Pillow image object.

A file format is required since the Pillow image object is serialized to a binary image file. The default is “JPEG”.

Parameters:

• image_pil (PIL.Image) – image in PIL format

• format (str, optional) – target file format, defaults to “JPEG”

Returns:

Image object

Return type:

Image

is_valid()

Check if this is a valid image.

Returns:

True if valid, False if invalid

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'data': FieldInfo(annotation=Union[bytes, MediaUrl], required=True, description='URL reference or base64-encoded data')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

to_pil() → <module 'PIL.Image' from '/home/runner/work/nimbus/nimbus/octoai-python-sdk/.venv/lib/python3.9/site-packages/PIL/Image.py'>

Convert to PIL Image.

Returns:

Pillow image object

Return type:

PIL.Image

class octoai.types.Media(data: bytes | MediaUrl)

Bases: BaseModel

Base class for media files.

The Media class is a wrapper for either binary media content or a URL reference to a media file. When an instance contains binary media data, it is stored and transferred over HTTP as a base64-encoded string. When an instance contains a URL reference, the URL is stored and transferred over HTTP as a string. The URL representation can be more advantageous when the media file is already hosted on a server, as it avoids the overhead of base64 encoding and decoding.

The Media class is the base class for the Image, Audio, and Video classes, which contain additional convenience methods to read and write media files in common formats. These subclasses should be used instead of the Media in most cases.

data: bytes | MediaUrl

default_download(file_name: str)

Download media URL to a file using the default settings.

Override this to change the download behavior.

Parameters:

file_name (str) – path to local file

download(follow_redirects, file_name)

Download media URL to local file.

Parameters:

• follow_redirects (bool) – whether to follow redirects

• file_name (str) – path to local file

classmethod from_base64(b64: bytes) → Media

Create from base64-encoded data, such as that returned from an HTTP call.

See also Media.from_endpoint_response().

Parameters:

b64 (bytes) – base64-encoded data representing a media file

Returns:

Media instance

Return type:

Media

classmethod from_endpoint_response(resp_dict: Dict[str, Any], key: str) → Media

Create from an endpoint response, such as an endpoint that produces media.

Parameters:

• resp_dict (Dict[str, Any]) – a response from an OctoAI endpoint that produces media

• key – the key name in the response that contains the media

Returns:

Media object

Return type:

Media

classmethod from_file(file_name: str) → Media

Create media from local file.

Parameters:

file_name (str) – path to local media file

Raises:

ValueError – file_name not found at provided path

Returns:

Media object

Return type:

Media

classmethod from_url(url: str, b64=False, follow_redirects=False) → Media

Create media from URL.

Parameters:

• url (str) – URL to media file

• b64 (bool, optional) – whether to base64-encode the media file, defaults to False

• follow_redirects (bool, optional) – whether to follow redirects, defaults to False

Return type:

Media

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'data': FieldInfo(annotation=Union[bytes, MediaUrl], required=True, description='URL reference or base64-encoded data')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

to_base64() → bytes

Convert media to base64-encoded bytes.

Returns:

media as base64-encoded bytes

Return type:

bytes

to_bytes() → bytes

Convert media to bytes.

Returns:

media as bytes

Return type:

bytes

to_file(file_name: str)

Write media to local file.

Parameters:

file_name (str) – path to local file

class octoai.types.MediaUrl(*, url: Url)

Bases: BaseModel

URL reference to a media file.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'url': FieldInfo(annotation=Url, required=True, metadata=[UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

url: Url

url2str(val) → str

Serialize HttpUrl to str.

class octoai.types.Video(data: bytes | MediaUrl)

Bases: Media

Video helpers for models that accept or return video.

The Video class is a wrapper for either binary video content or a URL reference to a video file. When an instance contains binary video data, it is stored and transferred over HTTP as a base64-encoded string. When an instance contains a URL reference, the URL is stored and transferred over HTTP as a string. The URL representation can be more advantageous when the video file is already hosted on a server, as it avoids the overhead of base64 encoding and decoding.

The Video class contains additional convenience methods to read and write video files in common formats.

data: bytes | MediaUrl

classmethod from_numpy(video_frames: List[_SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]], codec='libx264', pixel_format='yuv420p', fps=30) → Video

Create from NumPy video frames.

Parameters:

• video_frames (List[ArrayLike]) – list of NumPy arrays representing video frames

• codec (str, optional) – FFMPEG video codec, defaults to ‘libx264’

• pixel_format (str, optional) – FFMPEG pixel format, defaults to ‘yuv420p’

• fps (int, optional) – frames per second, defaults to 30

Returns:

Video object

Return type:

Video

is_valid() → bool

Check if this is a valid video.

Returns:

True if valid, False if invalid

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'data': FieldInfo(annotation=Union[bytes, MediaUrl], required=True, description='URL reference or base64-encoded data')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

to_numpy() → _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]

Convert to a NumPy array.

Returns:

NumPy array representing video frames

Return type:

ArrayLike

to_numpy_iterator() → Tuple[Iterator[_SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]], str]

Convert to a NumPy iterator.

Returns:

tuple of iterator of NumPy arrays representing video frames and the file name of the temporary file. Users of this method should delete the temporary file when the iterator is no longer needed to avoid filling up the disk.

Return type:

Tuple[Iterator[ArrayLike], str]

octoai.utils module

Various utilities.

octoai.utils.retry(fn: Callable[[], Response], retry_count: int = 5, interval: float = 1.0, exp_backoff: float = 2.0, jitter: float = 1.0, maximum_backoff: float = 30.0) → Response

Retry an HTTP request with exponential backoff and jitter.

Parameters:

• fn (Callable[[], httpx.Response]) – function to call

• retry_count (int, optional) – number of times to retry, defaults to 5

• interval (float, optional) – duration to wait before retry, defaults to 1.0

• exp_backoff (float, optional) – exponent to increase interval each try, defaults to 2.0

• jitter (float, optional) – , defaults to 1.0

• maximum_backoff (float, optional) – max duration to wait, defaults to 30.0

Raises:

OctoAIClientError – occurs when a client error is thrown.

Returns:

response from api server

Return type:

httpx.Response

Module contents

Initializes the octoai module.