Core

Shared building blocks of scikit-ai: media types, requests, backends and component bases.

AnimationRequest(image, motion_prompt=None, duration=2.0, fps=None, seed=None) dataclass

Ask a backend to turn an image into a short video that starts from it.

Parameters:

Name	Type	Description	Default
image	Image	The first frame of the video.	required
motion_prompt	str | None	A description of the motion.	None
duration	float	The length of the video in seconds.	2.0
fps	int | None	The number of frames per second, or None for the model's default.	None
seed	int | None	The seed for backends that support seeding.	None

Audio(samples, sample_rate)

Hold audio samples and their sample rate.

Parameters:

Name	Type	Description	Default
samples	NDArray[floating]	The audio samples, with shape (n_samples,) for mono or (n_samples, n_channels).	required
sample_rate	int	The number of samples per second.	required

Examples:

>>> import numpy as np
>>> from skai.core import Audio
>>> audio = Audio(np.zeros(8000), 16000)
>>> audio.duration
0.5

Source code in src/skai/core/_media.py

def __init__(self, samples: NDArray[np.floating], sample_rate: int) -> None:
    self.samples = np.asarray(samples)
    self.sample_rate = sample_rate

duration property

The length of the audio in seconds.

__repr__()

Return a short description of the audio.

Source code in src/skai/core/_media.py

def __repr__(self) -> str:
    """Return a short description of the audio."""
    return f'Audio(duration={self.duration:.2f}s, sample_rate={self.sample_rate})'

from_bytes(data) classmethod

Decode audio from encoded bytes.

Parameters:

Name	Type	Description	Default
data	bytes	The encoded audio, in any format that libsndfile reads.	required

Returns:

Type	Description
Self	The decoded audio.

Source code in src/skai/core/_media.py

@classmethod
def from_bytes(cls, data: bytes) -> Self:
    """Decode audio from encoded bytes.

    Args:
        data:
            The encoded audio, in any format that libsndfile reads.

    Returns:
        The decoded audio.
    """
    samples, sample_rate = sf.read(io.BytesIO(data), dtype='float32')
    return cls(samples, int(sample_rate))

resample(sample_rate)

Return the audio mixed to mono and resampled by linear interpolation.

Parameters:

Name	Type	Description	Default
sample_rate	int	The new number of samples per second.	required

Returns:

Type	Description
Audio	The resampled mono audio, or the mono audio itself when the rate is unchanged.

Source code in src/skai/core/_media.py

def resample(self, sample_rate: int) -> 'Audio':
    """Return the audio mixed to mono and resampled by linear interpolation.

    Args:
        sample_rate:
            The new number of samples per second.

    Returns:
        The resampled mono audio, or the mono audio itself when the rate is unchanged.
    """
    mono = self.to_mono()
    if mono.sample_rate == sample_rate:
        return mono
    times = np.arange(round(mono.duration * sample_rate)) / sample_rate
    source_times = np.arange(mono.samples.shape[0]) / mono.sample_rate
    return Audio(np.interp(times, source_times, mono.samples).astype(np.float32), sample_rate)

save(path)

Write the audio to a WAV file.

Parameters:

Name	Type	Description	Default
path	str | Path	The path of the file to write.	required

Returns:

Type	Description
Path	The path of the written file.

Source code in src/skai/core/_media.py

def save(self, path: str | Path) -> Path:
    """Write the audio to a WAV file.

    Args:
        path:
            The path of the file to write.

    Returns:
        The path of the written file.
    """
    path = Path(path)
    path.write_bytes(self.to_bytes())
    return path

to_bytes()

Encode the audio as WAV bytes.

Returns:

Type	Description
bytes	The WAV-encoded audio.

Source code in src/skai/core/_media.py

def to_bytes(self) -> bytes:
    """Encode the audio as WAV bytes.

    Returns:
        The WAV-encoded audio.
    """
    buffer = io.BytesIO()
    sf.write(buffer, self.samples, self.sample_rate, format='WAV')
    return buffer.getvalue()

to_mono()

Return the audio mixed down to one channel.

Returns:

Type	Description
Audio	The mono audio.

Source code in src/skai/core/_media.py

def to_mono(self) -> 'Audio':
    """Return the audio mixed down to one channel.

    Returns:
        The mono audio.
    """
    if self.samples.ndim == 1:
        return self
    return Audio(self.samples.mean(axis=1), self.sample_rate)

AudioGenerationRequest(prompt, duration, seed=None) dataclass

Ask a backend to generate music or a sound from a description.

Parameters:

Name	Type	Description	Default
prompt	str	The description of the audio.	required
duration	float	The length of the audio in seconds.	required
seed	int | None	The seed for backends that support seeding.	None

AudioOperationRequest(audio, stem='vocals', voice=None, seed=None) dataclass

Ask a backend to separate a stem of audio or to convert its voice.

Parameters:

Name	Type	Description	Default
audio	Audio	The audio.	required
stem	str	The stem to keep when separating, 'vocals' or 'accompaniment'.	'vocals'
voice	Audio | None	Reference audio of the target voice when converting.	None
seed	int | None	The seed for backends that support seeding.	None

AvatarRequest(avatar, audio, seed=None) dataclass

Ask a backend to animate a portrait so that it speaks an audio track.

Parameters:

Name	Type	Description	Default
avatar	Image	The portrait to animate.	required
audio	Audio	The speech the portrait says.	required
seed	int | None	The seed for backends that support seeding.	None

Backend(model=None, *, options=None, max_concurrency=1)

Bases: BaseEstimator

Run component tasks on a model provider.

A backend implements a task by overriding the method of the same name. The tasks a backend supports are the methods it overrides. Attributes that start with an underscore hold runtime state, such as a client, and are not pickled. A key of default_models is a task, or '<task>:<input kind>' for a default that depends on the input.

Parameters:

Name	Type	Description	Default
model	str | None	The model to use. The default model of the task is used when it is None.	None
options	dict | None	Extra keyword arguments passed to the provider on every call.	None
max_concurrency	int	The maximum number of requests that run at the same time.	1

Source code in src/skai/core/_backend.py

def __init__(self, model: str | None = None, *, options: dict | None = None, max_concurrency: int = 1) -> None:
    self.model = model
    self.options = options
    self.max_concurrency = max_concurrency

call_options property

The extra keyword arguments passed to the provider.

__getstate__()

Return the state to pickle without the runtime attributes.

Source code in src/skai/core/_backend.py

def __getstate__(self) -> dict[str, Any]:
    """Return the state to pickle without the runtime attributes."""
    state = super().__getstate__()
    return {key: value for key, value in state.items() if not key.startswith('_') or key == '_sklearn_version'}

__sklearn_clone__()

Clone the backend, sharing live objects such as clients, including those inside options.

Source code in src/skai/core/_backend.py

def __sklearn_clone__(self) -> Self:
    """Clone the backend, sharing live objects such as clients, including those inside `options`."""
    params = self.get_params(deep=False)
    cloned = {
        name: value if name in self.shared_params else clone(value, safe=False)
        for name, value in params.items()
        if name != 'options'
    }
    cloned['options'] = None if self.options is None else dict(self.options)
    return type(self)(**cloned)

animate_avatar(request)

Animate a portrait so that it speaks an audio track.

Parameters:

Name	Type	Description	Default
request	AvatarRequest	The request.	required

Returns:

Type	Description
Video	The video.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def animate_avatar(self, request: AvatarRequest) -> Video:
    """Animate a portrait so that it speaks an audio track.

    Args:
        request:
            The request.

    Returns:
        The video.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

animate_image(request)

Turn an image into a short video that starts from it.

Parameters:

Name	Type	Description	Default
request	AnimationRequest	The request.	required

Returns:

Type	Description
Video	The video.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def animate_image(self, request: AnimationRequest) -> Video:
    """Turn an image into a short video that starts from it.

    Args:
        request:
            The request.

    Returns:
        The video.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

classify_audio(request)

Score audio against the labels of a request.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def classify_audio(self, request: ClassificationRequest) -> dict[str, float]:
    """Score audio against the labels of a request.

    Args:
        request:
            The request.

    Returns:
        A score per label.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

classify_image(request)

Score an image against the labels of a request.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def classify_image(self, request: ClassificationRequest) -> dict[str, float]:
    """Score an image against the labels of a request.

    Args:
        request:
            The request.

    Returns:
        A score per label.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

classify_multimodal(request)

Score a multimodal sample against the labels of a request.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def classify_multimodal(self, request: ClassificationRequest) -> dict[str, float]:
    """Score a multimodal sample against the labels of a request.

    Args:
        request:
            The request.

    Returns:
        A score per label.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

classify_text(request)

Score a text against the labels of a request.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def classify_text(self, request: ClassificationRequest) -> dict[str, float]:
    """Score a text against the labels of a request.

    Args:
        request:
            The request.

    Returns:
        A score per label.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

classify_video(request)

Score sampled video frames against the labels of a request.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The request.	required

Returns:

Type	Description
dict[str, float]	A score per label.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def classify_video(self, request: ClassificationRequest) -> dict[str, float]:
    """Score sampled video frames against the labels of a request.

    Args:
        request:
            The request.

    Returns:
        A score per label.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

connect()

Prepare the backend to run requests, for example by creating a client.

Returns:

Type	Description
Self	The backend.

Source code in src/skai/core/_backend.py

def connect(self) -> Self:
    """Prepare the backend to run requests, for example by creating a client.

    Returns:
        The backend.
    """
    return self

convert_voice(request)

Say the speech of audio in another voice.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The request.	required

Returns:

Type	Description
Audio	The converted speech.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def convert_voice(self, request: AudioOperationRequest) -> Audio:
    """Say the speech of audio in another voice.

    Args:
        request:
            The request.

    Returns:
        The converted speech.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

detect_objects(request)

Find the boxes of the regions of an image.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The request.	required

Returns:

Type	Description
list[Region]	The regions, without masks.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def detect_objects(self, request: RegionRequest) -> list[Region]:
    """Find the boxes of the regions of an image.

    Args:
        request:
            The request.

    Returns:
        The regions, without masks.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

diarize(request)

Find who speaks when, with what they say.

Parameters:

Name	Type	Description	Default
request	DiarizationRequest	The request.	required

Returns:

Type	Description
Transcript	The transcript, with a speaker per segment.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def diarize(self, request: DiarizationRequest) -> Transcript:
    """Find who speaks when, with what they say.

    Args:
        request:
            The request.

    Returns:
        The transcript, with a speaker per segment.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

edit_image(request)

Edit an image with an instruction.

Parameters:

Name	Type	Description	Default
request	ImageEditRequest	The request.	required

Returns:

Type	Description
Image	The edited image.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def edit_image(self, request: ImageEditRequest) -> Image.Image:
    """Edit an image with an instruction.

    Args:
        request:
            The request.

    Returns:
        The edited image.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

embed_audio(request)

Embed audio.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def embed_audio(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed audio.

    Args:
        request:
            The request.

    Returns:
        The embedding vector.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

embed_image(request)

Embed an image.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def embed_image(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed an image.

    Args:
        request:
            The request.

    Returns:
        The embedding vector.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

embed_text(request)

Embed a text.

Parameters:

Name	Type	Description	Default
request	EmbeddingRequest	The request.	required

Returns:

Type	Description
NDArray[floating]	The embedding vector.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def embed_text(self, request: EmbeddingRequest) -> NDArray[np.floating]:
    """Embed a text.

    Args:
        request:
            The request.

    Returns:
        The embedding vector.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

enhance_audio(request)

Remove the noise of audio.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The request.	required

Returns:

Type	Description
Audio	The denoised audio.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def enhance_audio(self, request: AudioOperationRequest) -> Audio:
    """Remove the noise of audio.

    Args:
        request:
            The request.

    Returns:
        The denoised audio.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

extract_fields(request)

Extract named fields from a sample.

Parameters:

Name	Type	Description	Default
request	ExtractionRequest	The request.	required

Returns:

Type	Description
dict[str, Any]	The value of every field, or None for a field the sample does not contain.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def extract_fields(self, request: ExtractionRequest) -> dict[str, Any]:
    """Extract named fields from a sample.

    Args:
        request:
            The request.

    Returns:
        The value of every field, or `None` for a field the sample does not contain.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

find_moments(request)

Find the parts of audio or a video that match a description.

Parameters:

Name	Type	Description	Default
request	MomentRequest	The request.	required

Returns:

Type	Description
list[Moment]	The moments, in time order.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def find_moments(self, request: MomentRequest) -> list[Moment]:
    """Find the parts of audio or a video that match a description.

    Args:
        request:
            The request.

    Returns:
        The moments, in time order.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

fine_tune(request)

Train a model on prompts and their expected answers, and wait until it is ready.

Parameters:

Name	Type	Description	Default
request	FineTuneRequest	The request.	required

Returns:

Type	Description
str	The id of the trained model.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def fine_tune(self, request: FineTuneRequest) -> str:
    """Train a model on prompts and their expected answers, and wait until it is ready.

    Args:
        request:
            The request.

    Returns:
        The id of the trained model.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

generate_image(request)

Generate an image from a prompt.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The request.	required

Returns:

Type	Description
Image	The image.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def generate_image(self, request: ImageGenerationRequest) -> Image.Image:
    """Generate an image from a prompt.

    Args:
        request:
            The request.

    Returns:
        The image.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

generate_music(request)

Generate music from a description.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The request.	required

Returns:

Type	Description
Audio	The music.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def generate_music(self, request: AudioGenerationRequest) -> Audio:
    """Generate music from a description.

    Args:
        request:
            The request.

    Returns:
        The music.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

generate_sound(request)

Generate a sound from a description.

Parameters:

Name	Type	Description	Default
request	AudioGenerationRequest	The request.	required

Returns:

Type	Description
Audio	The sound.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def generate_sound(self, request: AudioGenerationRequest) -> Audio:
    """Generate a sound from a description.

    Args:
        request:
            The request.

    Returns:
        The sound.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

generate_text(request)

Produce text from a sample by following instructions.

Parameters:

Name	Type	Description	Default
request	TextGenerationRequest	The request.	required

Returns:

Type	Description
str	The text.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def generate_text(self, request: TextGenerationRequest) -> str:
    """Produce text from a sample by following instructions.

    Args:
        request:
            The request.

    Returns:
        The text.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

generate_video(request)

Generate a video from a prompt.

Parameters:

Name	Type	Description	Default
request	VideoGenerationRequest	The request.	required

Returns:

Type	Description
Video	The video.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def generate_video(self, request: VideoGenerationRequest) -> Video:
    """Generate a video from a prompt.

    Args:
        request:
            The request.

    Returns:
        The video.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

lip_sync(request)

Make a video speak new audio.

Parameters:

Name	Type	Description	Default
request	LipSyncRequest	The request.	required

Returns:

Type	Description
Video	The video.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def lip_sync(self, request: LipSyncRequest) -> Video:
    """Make a video speak new audio.

    Args:
        request:
            The request.

    Returns:
        The video.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

pick_thumbnail(request)

Pick the frame that best represents a video.

Parameters:

Name	Type	Description	Default
request	ThumbnailRequest	The request.	required

Returns:

Type	Description
int	The index of the frame.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def pick_thumbnail(self, request: ThumbnailRequest) -> int:
    """Pick the frame that best represents a video.

    Args:
        request:
            The request.

    Returns:
        The index of the frame.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

remove_background(request)

Remove the background of an image.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The request.	required

Returns:

Type	Description
Image	An RGBA image whose background is transparent.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def remove_background(self, request: ImageOperationRequest) -> Image.Image:
    """Remove the background of an image.

    Args:
        request:
            The request.

    Returns:
        An RGBA image whose background is transparent.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

resolve_model(task, input_kind=None)

Return the model that runs a task.

Parameters:

Name	Type	Description	Default
task	str	The name of the task.	required
input_kind	str | None	The input kind of the sample, which selects a default model of '<task>:<input kind>' first.	None

Returns:

Type	Description
str	The model parameter, or the default model of the task.

Raises:

Type	Description
UnsupportedBackendError	If no model is given and the task has no default model.

Source code in src/skai/core/_backend.py

def resolve_model(self, task: str, input_kind: str | None = None) -> str:
    """Return the model that runs a task.

    Args:
        task:
            The name of the task.

        input_kind:
            The input kind of the sample, which selects a default model of `'<task>:<input kind>'` first.

    Returns:
        The `model` parameter, or the default model of the task.

    Raises:
        UnsupportedBackendError:
            If no model is given and the task has no default model.
    """
    if self.model is not None:
        return self.model
    key = f'{task}:{input_kind}'
    if input_kind is not None and key in self.default_models:
        return self.default_models[key]
    if task not in self.default_models:
        msg = f'{type(self).__name__} has no default model for `{task}`. Set the `model` parameter.'
        raise UnsupportedBackendError(msg)
    return self.default_models[task]

run(task, requests)

Run a task on several requests, keeping their order.

Parameters:

Name	Type	Description	Default
task	str	The name of the task.	required
requests	list[Any]	The requests.	required

Returns:

Type	Description
list[Any]	The result of every request, or the exception it raised.

Source code in src/skai/core/_backend.py

def run(self, task: str, requests: list[Any]) -> list[Any]:
    """Run a task on several requests, keeping their order.

    Args:
        task:
            The name of the task.

        requests:
            The requests.

    Returns:
        The result of every request, or the exception it raised.
    """
    method = partial(_call, getattr(self, task))
    if self.max_concurrency <= 1 or len(requests) <= 1:
        return [method(request) for request in requests]
    with ThreadPoolExecutor(max_workers=self.max_concurrency) as executor:
        return list(executor.map(method, requests))

segment_image(request)

Find the masks of the regions of an image.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The request.	required

Returns:

Type	Description
list[Region]	The regions, with masks.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def segment_image(self, request: RegionRequest) -> list[Region]:
    """Find the masks of the regions of an image.

    Args:
        request:
            The request.

    Returns:
        The regions, with masks.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

separate_audio(request)

Separate a stem of audio.

Parameters:

Name	Type	Description	Default
request	AudioOperationRequest	The request.	required

Returns:

Type	Description
Audio	The stem.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def separate_audio(self, request: AudioOperationRequest) -> Audio:
    """Separate a stem of audio.

    Args:
        request:
            The request.

    Returns:
        The stem.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

supported_tasks() classmethod

Return the tasks that the backend implements.

Returns:

Type	Description
frozenset[str]	The names of the tasks.

Source code in src/skai/core/_backend.py

@classmethod
def supported_tasks(cls) -> frozenset[str]:
    """Return the tasks that the backend implements.

    Returns:
        The names of the tasks.
    """
    return frozenset(task for task in TASKS if getattr(cls, task) is not getattr(Backend, task))

supports(task, input_kind=None) classmethod

Return whether the backend runs a task on samples of an input kind.

Parameters:

Name	Type	Description	Default
task	str	The name of the task.	required
input_kind	str | None	The input kind of the samples, or None for any.	None

Returns:

Type	Description
bool	Whether the task is supported.

Source code in src/skai/core/_backend.py

@classmethod
def supports(cls, task: str, input_kind: str | None = None) -> bool:
    """Return whether the backend runs a task on samples of an input kind.

    Args:
        task:
            The name of the task.

        input_kind:
            The input kind of the samples, or `None` for any.

    Returns:
        Whether the task is supported.
    """
    return task in cls.supported_tasks()

synthesize_speech(request)

Speak a text.

Parameters:

Name	Type	Description	Default
request	SpeechRequest	The request.	required

Returns:

Type	Description
Audio	The speech.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def synthesize_speech(self, request: SpeechRequest) -> Audio:
    """Speak a text.

    Args:
        request:
            The request.

    Returns:
        The speech.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

transcribe(request)

Transcribe speech.

Parameters:

Name	Type	Description	Default
request	TranscriptionRequest	The request.	required

Returns:

Type	Description
Transcript	The transcript.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def transcribe(self, request: TranscriptionRequest) -> Transcript:
    """Transcribe speech.

    Args:
        request:
            The request.

    Returns:
        The transcript.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

transcribe_image(request)

Transcribe the text written in an image.

Parameters:

Name	Type	Description	Default
request	ImageTranscriptionRequest	The request.	required

Returns:

Type	Description
str	The text.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def transcribe_image(self, request: ImageTranscriptionRequest) -> str:
    """Transcribe the text written in an image.

    Args:
        request:
            The request.

    Returns:
        The text.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

upscale_image(request)

Upscale an image.

Parameters:

Name	Type	Description	Default
request	ImageOperationRequest	The request.	required

Returns:

Type	Description
Image	The upscaled image.

Raises:

Type	Description
NotImplementedError	If the backend does not support the task.

Source code in src/skai/core/_backend.py

def upscale_image(self, request: ImageOperationRequest) -> Image.Image:
    """Upscale an image.

    Args:
        request:
            The request.

    Returns:
        The upscaled image.

    Raises:
        NotImplementedError:
            If the backend does not support the task.
    """
    raise NotImplementedError

BaseClassifier

Bases: ClassifierMixin, BaseMediaEstimator

Assign labels with zero-shot or few-shot prediction.

__sklearn_tags__()

Declare that the classifier supports multilabel targets.

Source code in src/skai/core/_estimator.py

def __sklearn_tags__(self) -> Tags:
    """Declare that the classifier supports multilabel targets."""
    tags = super().__sklearn_tags__()
    tags.classifier_tags.multi_label = True
    return tags

fit(X=None, y=None)

Learn the classes and select the few-shot examples.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	The training samples. They can be None for zero-shot classification with labels.	None
y	ArrayLike | None	The targets, as labels or as a binary indicator matrix for multilabel classification. They can be None for zero-shot classification with labels.	None

Returns:

Type	Description
Self	The fitted classifier.

Raises:

Type	Description
ValueError	If adaptation='finetune' and no targets are given.

Source code in src/skai/core/_estimator.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Learn the classes and select the few-shot examples.

    Args:
        X:
            The training samples. They can be `None` for zero-shot classification with `labels`.

        y:
            The targets, as labels or as a binary indicator matrix for multilabel classification. They can be
            `None` for zero-shot classification with `labels`.

    Returns:
        The fitted classifier.

    Raises:
        ValueError:
            If `adaptation='finetune'` and no targets are given.
    """
    self._connect()
    samples = self._samples(X)
    targets = None
    if y is None:
        self._fit_labels()
    else:
        targets = self._fit_targets(samples, y)
    if self.adaptation == 'finetune':
        if targets is None:
            msg = 'Fine-tuning needs the targets `y`.'
            raise ValueError(msg)
        answers = [', '.join(self._label_names_of(target)) or NO_LABEL for target in targets]
        self._fine_tune(samples, answers)
        return self
    self.examples_ = self._select_examples(samples, targets)
    return self

predict(X)

Predict the labels of every sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray	The predicted labels, or a binary indicator matrix for multilabel classification. A failed sample is
NDArray	None, or a row of NaN for multilabel classification.

Source code in src/skai/core/_estimator.py

def predict(self, X: ArrayLike) -> NDArray:
    """Predict the labels of every sample.

    Args:
        X:
            The samples.

    Returns:
        The predicted labels, or a binary indicator matrix for multilabel classification. A failed sample is
        `None`, or a row of NaN for multilabel classification.
    """
    proba = self.predict_proba(X)
    failed = np.isnan(proba).any(axis=1)
    if self.multilabel_:
        indicator = (proba >= MULTILABEL_THRESHOLD).astype(int)
        return np.where(np.isnan(proba), np.nan, indicator) if failed.any() else indicator
    labels = self.classes_[np.nan_to_num(proba, nan=0.0).argmax(axis=1)]
    if not failed.any():
        return labels
    labels = labels.astype(object)
    labels[failed] = None
    return labels

predict_proba(X)

Score every sample against every class.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[float64]	An array of shape (n_samples, n_classes). Rows sum to one, except for multilabel classification where
NDArray[float64]	every column is an independent probability. A failed sample has a row of NaN.

Source code in src/skai/core/_estimator.py

def predict_proba(self, X: ArrayLike) -> NDArray[np.float64]:
    """Score every sample against every class.

    Args:
        X:
            The samples.

    Returns:
        An array of shape `(n_samples, n_classes)`. Rows sum to one, except for multilabel classification where
        every column is an independent probability. A failed sample has a row of NaN.
    """
    check_is_fitted(self, 'classes_')
    names = [str(label) for label in self.classes_]
    proba = np.full((len(self._samples(X)), len(names)), np.nan)
    for row, scores in enumerate(self._run_task(X)):
        if scores is not None:
            proba[row] = [scores.get(name, 0.0) for name in names]
    if self.multilabel_:
        return proba
    totals = proba.sum(axis=1, keepdims=True)
    proba = np.where(totals == 0, 1.0 / len(names), proba)
    return proba / np.where(totals == 0, 1.0, totals)

BaseEmbedder

Bases: TransformerMixin, BaseMediaEstimator

Map every sample to a fixed-size vector, learning centering and a projection from the samples in fit.

fit(X=None, y=None)

Embed the samples and learn their mean and, when needed, a projection.

With X, every sample is embedded, the mean is learned for centering, and a PCA projection to dimensions is learned when the backend cannot shorten embeddings. Without X, a small probe sample is embedded to find the number of dimensions, and nothing is centered or projected.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	The samples.	None
y	ArrayLike | None	Ignored.	None

Returns:

Type	Description
Self	The fitted embedder.

Source code in src/skai/core/_estimator.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Embed the samples and learn their mean and, when needed, a projection.

    With `X`, every sample is embedded, the mean is learned for centering, and a PCA projection to `dimensions`
    is learned when the backend cannot shorten embeddings. Without `X`, a small probe sample is embedded to find the
    number of dimensions, and nothing is centered or projected.

    Args:
        X:
            The samples.

        y:
            Ignored.

    Returns:
        The fitted embedder.
    """
    self._connect()
    embeddings = [embedding for embedding in self._embed(self._samples(X)) if embedding is not None]
    if embeddings:
        self._learn(np.stack(embeddings))
        return self
    probes = {
        'text': PROBE_TEXT,
        'image': Image.new('RGB', (PROBE_SIZE, PROBE_SIZE)),
        'audio': Audio(np.zeros(PROBE_SAMPLE_RATE, dtype=np.float32), PROBE_SAMPLE_RATE),
        'video': Video(np.zeros((PROBE_FRAMES, PROBE_SIZE, PROBE_SIZE, 3), dtype=np.uint8), fps=PROBE_FRAMES),
    }
    probe = self._embed([probes[self.input_kind]])[0]
    assert probe is not None
    self.n_features_out_ = probe[: self.dimensions].shape[0]
    self.mean_ = np.zeros(self.n_features_out_)
    self.components_ = None
    return self

get_feature_names_out(input_features=None)

Return the names of the embedding dimensions.

Parameters:

Name	Type	Description	Default
input_features	ArrayLike | None	Ignored.	None

Returns:

Type	Description
NDArray[object_]	The names <class name><index>.

Source code in src/skai/core/_estimator.py

def get_feature_names_out(self, input_features: ArrayLike | None = None) -> NDArray[np.object_]:
    """Return the names of the embedding dimensions.

    Args:
        input_features:
            Ignored.

    Returns:
        The names `<class name><index>`.
    """
    check_is_fitted(self, 'n_features_out_')
    prefix = type(self).__name__.lower()
    return np.asarray([f'{prefix}{index}' for index in range(self.n_features_out_)], dtype=object)

transform(X)

Embed every sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[float64]	An array of shape (n_samples, n_features_out_). A failed sample has a row of NaN.

Source code in src/skai/core/_estimator.py

def transform(self, X: ArrayLike) -> NDArray[np.float64]:
    """Embed every sample.

    Args:
        X:
            The samples.

    Returns:
        An array of shape `(n_samples, n_features_out_)`. A failed sample has a row of NaN.
    """
    results = self._embed(X)
    embeddings = np.full((len(results), self.n_features_out_), np.nan)
    for row, embedding in enumerate(results):
        if embedding is not None:
            embeddings[row] = self._to_vector(embedding)
    return embeddings

BaseExtractor

Bases: TransformerMixin, BaseMediaEstimator

Extract a record of declared fields from every sample, learning from reference records in fit.

predict returns one dict per sample, and transform one column per top-level field.

fit(X=None, y=None)

Resolve the backend and select example pairs of samples and their reference records.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	The training samples. They can be None to extract zero-shot.	None
y	ArrayLike | None	The reference record of every sample, as dicts keyed by field name.	None

Returns:

Type	Description
Self	The fitted extractor.

Raises:

Type	Description
ValueError	If X and y have different lengths.

Source code in src/skai/core/_estimator.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Resolve the backend and select example pairs of samples and their reference records.

    Args:
        X:
            The training samples. They can be `None` to extract zero-shot.

        y:
            The reference record of every sample, as dicts keyed by field name.

    Returns:
        The fitted extractor.

    Raises:
        ValueError:
            If `X` and `y` have different lengths.
    """
    self._connect()
    samples = self._samples(X) if y is not None else []
    references = list(y) if y is not None else []
    if len(samples) != len(references):
        msg = f'`X` has {len(samples)} samples but `y` has {len(references)} records.'
        raise ValueError(msg)
    self.examples_: tuple[tuple[Any, Any], ...] = ()
    if references and self.k_shot != 0:
        indices = _select_indices(self.k_shot, list(range(len(samples))), self.random_state, self.max_examples)
        self.examples_ = tuple((self._read(samples[index], index), dict(references[index])) for index in indices)
    return self

get_feature_names_out(input_features=None)

Return the top-level field names.

Parameters:

Name	Type	Description	Default
input_features	ArrayLike | None	Ignored.	None

Returns:

Type	Description
NDArray[object_]	The field names, in the order of fields.

Source code in src/skai/core/_estimator.py

def get_feature_names_out(self, input_features: ArrayLike | None = None) -> NDArray[np.object_]:
    """Return the top-level field names.

    Args:
        input_features:
            Ignored.

    Returns:
        The field names, in the order of `fields`.
    """
    check_is_fitted(self, 'backend_')
    return np.asarray(self._names(), dtype=object)

predict(X)

Extract the record of every sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[object_]	One dict per sample whose keys are the field names, or None for a failed sample.

Source code in src/skai/core/_estimator.py

def predict(self, X: ArrayLike) -> NDArray[np.object_]:
    """Extract the record of every sample.

    Args:
        X:
            The samples.

    Returns:
        One dict per sample whose keys are the field names, or `None` for a failed sample.
    """
    return to_object_array(self._run_task(X))

transform(X)

Extract the fields of every sample as columns.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[object_]	An object array of shape (n_samples, n_fields). A failed sample has a row of None.

Source code in src/skai/core/_estimator.py

def transform(self, X: ArrayLike) -> NDArray[np.object_]:
    """Extract the fields of every sample as columns.

    Args:
        X:
            The samples.

    Returns:
        An object array of shape `(n_samples, n_fields)`. A failed sample has a row of `None`.
    """
    names = self._names()
    records = self._run_task(X)
    table = np.empty((len(records), len(names)), dtype=object)
    for row, record in enumerate(records):
        table[row] = [None if record is None else record[name] for name in names]
    return table

BaseGenerator

Bases: TransformerMixin, BaseMediaEstimator

Generate one output per sample, returned by both predict and transform.

fit(X=None, y=None)

Validate the parameters and resolve the backend.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	Ignored. The component learns nothing from data.	None
y	ArrayLike | None	Ignored.	None

Returns:

Type	Description
Self	The fitted component.

Source code in src/skai/core/_estimator.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Validate the parameters and resolve the backend.

    Args:
        X:
            Ignored. The component learns nothing from data.

        y:
            Ignored.

    Returns:
        The fitted component.
    """
    self._connect()
    self._prepare()
    return self

predict(X)

Generate one output per sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray	The outputs, in input order.

Source code in src/skai/core/_estimator.py

def predict(self, X: ArrayLike) -> NDArray:
    """Generate one output per sample.

    Args:
        X:
            The samples.

    Returns:
        The outputs, in input order.
    """
    return self._finalize(self._run_task(X))

transform(X)

Generate one output per sample, so the component can be an intermediate pipeline step.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray	The outputs, in input order.

Source code in src/skai/core/_estimator.py

def transform(self, X: ArrayLike) -> NDArray:
    """Generate one output per sample, so the component can be an intermediate pipeline step.

    Args:
        X:
            The samples.

    Returns:
        The outputs, in input order.
    """
    return self.predict(X)

BaseMediaEstimator

Bases: BaseEstimator

Resolve the backend, batch samples and handle failed samples for media components.

A component declares its input kind and its operations as class variables. operations maps every operation to the backend task that runs it, or to None for an operation that runs locally without a backend. The first operation is the default, and the parameter named by operation_param selects one at fit. default_backends overrides default_backend for some operations.

__init_subclass__(**kwargs)

Validate the operation parameter of a component against its operations.

Source code in src/skai/core/_estimator.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """Validate the operation parameter of a component against its operations."""
    super().__init_subclass__(**kwargs)
    operations = getattr(cls, 'operations', None)
    if operations and cls.operation_param in inspect.signature(cls.__init__).parameters:
        constraint = [StrOptions(set(operations))]
        cls._parameter_constraints = {**cls._parameter_constraints, cls.operation_param: constraint}

__sklearn_tags__()

Declare the input type and the non-deterministic output of the component.

Source code in src/skai/core/_estimator.py

def __sklearn_tags__(self) -> Tags:
    """Declare the input type and the non-deterministic output of the component."""
    tags = super().__sklearn_tags__()
    tags.non_deterministic = True
    tags.input_tags.two_d_array = False
    tags.input_tags.string = self.input_kind == 'text'
    return tags

BaseMediaGenerator

Bases: BaseGenerator

Generate images, audio or video and return them in the requested output format.

Local operations need the video extra for videos, and no backend.

BaseSegmenter

Bases: BaseSupervisedGenerator

Split every sample into a list of region records, chosen with the by parameter.

Every region is a dict with at least label, score and media, the cropped piece of the sample.

BaseSupervisedGenerator

Bases: BaseGenerator

Generate outputs and learn from reference outputs given as y in fit.

fit(X, y) learns from the references. fit_transform(X, y), which pipelines use for intermediate steps, ignores y unless use_y=True, because y is then the target of the final step.

fit(X=None, y=None)

Learn from reference outputs.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	The training samples.	None
y	ArrayLike | None	The reference output of every sample, such as its summary or transcript. Without it, the component predicts zero-shot.	None

Returns:

Type	Description
Self	The fitted component.

Source code in src/skai/core/_estimator.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Learn from reference outputs.

    Args:
        X:
            The training samples.

        y:
            The reference output of every sample, such as its summary or transcript. Without it, the component
            predicts zero-shot.

    Returns:
        The fitted component.
    """
    return self._fit(X, None if getattr(self, 'use_y', 'auto') is False else y)

fit_transform(X, y=None, **fit_params)

Fit and return the outputs, ignoring y unless use_y=True.

Pipelines fit intermediate steps through this method and pass them the target of the final step, so y is not read as references by default.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required
y	ArrayLike | None	The references, used only when use_y=True.	None
**fit_params	Any	Ignored.	{}

Returns:

Type	Description
NDArray	The outputs, in input order.

Source code in src/skai/core/_estimator.py

def fit_transform(self, X: ArrayLike, y: ArrayLike | None = None, **fit_params: Any) -> NDArray:
    """Fit and return the outputs, ignoring `y` unless `use_y=True`.

    Pipelines fit intermediate steps through this method and pass them the target of the final step, so `y` is
    not read as references by default.

    Args:
        X:
            The samples.

        y:
            The references, used only when `use_y=True`.

        **fit_params:
            Ignored.

    Returns:
        The outputs, in input order.
    """
    return self._fit(X, y if getattr(self, 'use_y', 'auto') is True else None).transform(X)

ClassificationRequest(sample, labels, descriptions=dict(), instructions=None, examples=(), multi_label=False) dataclass

Ask a backend to score a sample against a set of labels.

Parameters:

Name	Type	Description	Default
sample	Sample	The text, image, audio, sampled video frames or multimodal sample to classify.	required
labels	tuple[str, ...]	The candidate labels.	required
descriptions	dict[str, str]	Descriptions of the labels, keyed by label.	dict()
instructions	str | None	Extra guidance on how to assign the labels.	None
examples	tuple[tuple[Any, tuple[str, ...]], ...]	Labeled examples, as pairs of a sample and its labels.	()
multi_label	bool	Whether a sample can have several labels.	False

DiarizationRequest(audio, language=None, vocabulary=()) dataclass

Ask a backend to find who speaks when, with what they say.

Parameters:

Name	Type	Description	Default
audio	Audio	The speech to diarize.	required
language	str | None	The language of the speech. It is detected when it is None.	None
vocabulary	tuple[str, ...]	Domain terms that the speech may contain.	()

EmbeddingRequest(sample, dimensions=None) dataclass

Ask a backend to embed a text, an image or audio.

Parameters:

Name	Type	Description	Default
sample	str | Image | Audio	The text, image or audio to embed.	required
dimensions	int | None	The number of dimensions of the embedding, for backends that can shorten embeddings.	None

ExtractionRequest(sample, fields, instructions=None, examples=()) dataclass

Ask a backend to extract named fields from a sample.

Parameters:

Name	Type	Description	Default
sample	Sample	The text, image, audio, sampled video frames or multimodal sample to extract from.	required
fields	tuple[tuple[str, Any], ...]	The field names and their declarations, in output order. A declaration is a description, a dict of subfields, or a one-element list of either for a list of values.	required
instructions	str | None	Extra guidance on how to extract the fields.	None
examples	tuple[tuple[Any, dict[str, Any]], ...]	Example pairs of a sample and its reference record.	()

FineTuneError

Bases: RuntimeError

Signal that a fine-tuning job failed or was cancelled.

FineTuneRequest(task, examples) dataclass

Ask a backend to train a model on prompts and their expected answers.

Parameters:

Name	Type	Description	Default
task	str	The task the model is trained for.	required
examples	tuple[tuple[Prompt, str], ...]	Pairs of a prompt and the answer the model must give.	required

ImageEditRequest(image, instruction, mask=None, seed=None) dataclass

Ask a backend to edit an image with an instruction.

Parameters:

Name	Type	Description	Default
image	Image	The image to edit.	required
instruction	str	The edit to apply.	required
mask	Image | None	The region to edit, as a grayscale image where white marks the region, or None for the whole image.	None
seed	int | None	The seed for backends that support seeding.	None

ImageGenerationRequest(prompt, size, style=None, negative_prompt=None, seed=None, reference_images=()) dataclass

Ask a backend to generate an image from a prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The description of the image.	required
size	tuple[int, int]	The width and height of the image in pixels.	required
style	str | None	A description of the visual style.	None
negative_prompt	str | None	A description of what the image must not show.	None
seed	int | None	The seed for backends that support seeding.	None
reference_images	tuple[Image, ...]	Images of a subject that the image must show consistently.	()

ImageOperationRequest(image, scale=2, seed=None) dataclass

Ask a backend to upscale an image or remove its background.

Parameters:

Name	Type	Description	Default
image	Image	The image.	required
scale	int	The factor of upscaling.	2
seed	int | None	The seed for backends that support seeding.	None

ImageTranscriptionRequest(image, markdown=False) dataclass

Ask a backend to transcribe the text written in an image.

Parameters:

Name	Type	Description	Default
image	Image	The image to read.	required
markdown	bool	Whether to keep the page structure as Markdown.	False

InvalidResponseError

Bases: ValueError

Signal that a backend returned a response that does not fit the task.

InvalidSampleError

Bases: TypeError

Signal that a sample has the wrong type for a component.

LipSyncRequest(video, audio, seed=None) dataclass

Ask a backend to make a video speak new audio.

Parameters:

Name	Type	Description	Default
video	Video	The video of a speaking person.	required
audio	Audio	The new speech.	required
seed	int | None	The seed for backends that support seeding.	None

MediaReadError

Bases: ValueError

Signal that a media sample cannot be read or decoded.

MissingCredentialError

Bases: RuntimeError

Signal that the environment variable holding a credential is not set.

MissingExtraError

Bases: ImportError

Signal that an optional dependency group is not installed.

Moment(start, end, score) dataclass

Hold one part of audio or a video that matches a description.

Parameters:

Name	Type	Description	Default
start	float	The start time in seconds.	required
end	float	The end time in seconds.	required
score	float	The confidence of the backend.	required

MomentRequest(sample, prompt) dataclass

Ask a backend to find the parts of audio or a video that match a description.

Parameters:

Name	Type	Description	Default
sample	Audio | VideoFrames	The audio or the sampled video frames.	required
prompt	str	A description of the moments to find.	required

MultiRequest

Bases: tuple

Hold several backend requests of one sample, whose results are returned together as a list.

Prompt(system, user, images=(), audio=(), json_output=False) dataclass

Hold the messages of a prompt for a chat or vision model.

Parameters:

Name	Type	Description	Default
system	str	The system message.	required
user	str	The user message. Images are referred to as <image 1>, <image 2> and so on, and audio as <audio 1>.	required
images	tuple[Image, ...]	The images of the user message, in the order they are referred to.	()
audio	tuple[Audio, ...]	The audio of the user message, in the order it is referred to.	()
json_output	bool	Whether the answer must be a JSON object.	False

PromptedBackend(model=None, *, options=None, max_concurrency=1)

Bases: Backend

Run analysis and text tasks by prompting a chat, vision or audio model.

A subclass implements complete, and inherits classification, text generation, field extraction, image transcription, object detection, moment finding and thumbnail picking for every input kind its models accept: text, images and video frames always, and audio only when the subclass sets accepts_audio. Video requests include the audio track only when it sets sends_video_audio.

Parameters:

Name	Type	Description	Default
model	str | None	The model to use. The default model of the task is used when it is None.	None
options	dict | None	Extra keyword arguments passed to the provider on every call.	None
max_concurrency	int	The maximum number of requests that run at the same time.	1

Source code in src/skai/core/_backend.py

def __init__(self, model: str | None = None, *, options: dict | None = None, max_concurrency: int = 1) -> None:
    self.model = model
    self.options = options
    self.max_concurrency = max_concurrency

classify_audio(request)

Score audio by asking an audio model for its labels.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score of one for every label the model names and zero for the others.

Source code in src/skai/core/_backend.py

def classify_audio(self, request: ClassificationRequest) -> dict[str, float]:
    """Score audio by asking an audio model for its labels.

    Args:
        request:
            The classification request.

    Returns:
        A score of one for every label the model names and zero for the others.
    """
    return self._classify('classify_audio', request)

classify_image(request)

Score an image by asking the vision model for its labels.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score of one for every label the model names and zero for the others.

Source code in src/skai/core/_backend.py

def classify_image(self, request: ClassificationRequest) -> dict[str, float]:
    """Score an image by asking the vision model for its labels.

    Args:
        request:
            The classification request.

    Returns:
        A score of one for every label the model names and zero for the others.
    """
    return self._classify('classify_image', request)

classify_multimodal(request)

Score a multimodal sample by asking the model for its labels.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score of one for every label the model names and zero for the others.

Source code in src/skai/core/_backend.py

def classify_multimodal(self, request: ClassificationRequest) -> dict[str, float]:
    """Score a multimodal sample by asking the model for its labels.

    Args:
        request:
            The classification request.

    Returns:
        A score of one for every label the model names and zero for the others.
    """
    return self._classify('classify_multimodal', request)

classify_text(request)

Score a text by asking the model for its labels.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score of one for every label the model names and zero for the others.

Source code in src/skai/core/_backend.py

def classify_text(self, request: ClassificationRequest) -> dict[str, float]:
    """Score a text by asking the model for its labels.

    Args:
        request:
            The classification request.

    Returns:
        A score of one for every label the model names and zero for the others.
    """
    return self._classify('classify_text', request)

classify_video(request)

Score sampled video frames by asking the vision model for their labels.

Parameters:

Name	Type	Description	Default
request	ClassificationRequest	The classification request.	required

Returns:

Type	Description
dict[str, float]	A score of one for every label the model names and zero for the others.

Source code in src/skai/core/_backend.py

def classify_video(self, request: ClassificationRequest) -> dict[str, float]:
    """Score sampled video frames by asking the vision model for their labels.

    Args:
        request:
            The classification request.

    Returns:
        A score of one for every label the model names and zero for the others.
    """
    return self._classify('classify_video', request)

complete(prompt, model) abstractmethod

Return the answer of a model to a prompt.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The prompt.	required
model	str	The model to prompt.	required

Returns:

Type	Description
str	The answer of the model.

Source code in src/skai/core/_backend.py

@abstractmethod
def complete(self, prompt: Prompt, model: str) -> str:
    """Return the answer of a model to a prompt.

    Args:
        prompt:
            The prompt.

        model:
            The model to prompt.

    Returns:
        The answer of the model.
    """

detect_objects(request)

Find the boxes of regions by asking the vision model for them as JSON.

Parameters:

Name	Type	Description	Default
request	RegionRequest	The region request.	required

Returns:

Type	Description
list[Region]	The regions, without masks.

Source code in src/skai/core/_backend.py

def detect_objects(self, request: RegionRequest) -> list[Region]:
    """Find the boxes of regions by asking the vision model for them as JSON.

    Args:
        request:
            The region request.

    Returns:
        The regions, without masks.
    """
    answer = self._prompt('detect_objects', build_detection_prompt(request), request.image)
    return parse_regions(answer, request.image.size)

extract_fields(request)

Extract fields by asking the model for a JSON object.

Parameters:

Name	Type	Description	Default
request	ExtractionRequest	The extraction request.	required

Returns:

Type	Description
dict[str, Any]	The value of every field, or None for a field the sample does not contain.

Source code in src/skai/core/_backend.py

def extract_fields(self, request: ExtractionRequest) -> dict[str, Any]:
    """Extract fields by asking the model for a JSON object.

    Args:
        request:
            The extraction request.

    Returns:
        The value of every field, or `None` for a field the sample does not contain.
    """
    prompt = build_extraction_prompt(request, include_audio=self.sends_video_audio_now())
    answer = self._prompt('extract_fields', prompt, request.sample)
    return parse_fields(answer, tuple(name for name, _ in request.fields))

find_moments(request)

Find moments by asking the model for their times as JSON.

Parameters:

Name	Type	Description	Default
request	MomentRequest	The moment request.	required

Returns:

Type	Description
list[Moment]	The moments, in time order.

Source code in src/skai/core/_backend.py

def find_moments(self, request: MomentRequest) -> list[Moment]:
    """Find moments by asking the model for their times as JSON.

    Args:
        request:
            The moment request.

    Returns:
        The moments, in time order.
    """
    prompt = build_moment_prompt(request, include_audio=self.sends_video_audio_now())
    answer = self._prompt('find_moments', prompt, request.sample)
    return parse_moments(answer, request.sample.duration)

generate_text(request)

Produce text from a sample by prompting the model with the instructions.

Parameters:

Name	Type	Description	Default
request	TextGenerationRequest	The text generation request.	required

Returns:

Type	Description
str	The text.

Source code in src/skai/core/_backend.py

def generate_text(self, request: TextGenerationRequest) -> str:
    """Produce text from a sample by prompting the model with the instructions.

    Args:
        request:
            The text generation request.

    Returns:
        The text.
    """
    prompt = build_text_generation_prompt(request, include_audio=self.sends_video_audio_now())
    return self._prompt('generate_text', prompt, request.sample).strip()

pick_thumbnail(request)

Pick the frame that best represents a video by asking the vision model.

Parameters:

Name	Type	Description	Default
request	ThumbnailRequest	The thumbnail request.	required

Returns:

Type	Description
int	The index of the frame.

Source code in src/skai/core/_backend.py

def pick_thumbnail(self, request: ThumbnailRequest) -> int:
    """Pick the frame that best represents a video by asking the vision model.

    Args:
        request:
            The thumbnail request.

    Returns:
        The index of the frame.
    """
    answer = self._prompt('pick_thumbnail', build_thumbnail_prompt(request), request.frames.frames[0])
    return parse_frame_number(answer, len(request.frames.frames))

sends_video_audio_now()

Return whether video requests include the audio track with the current settings.

Returns:

Type	Description
bool	The class flag sends_video_audio, unless a subclass makes it depend on its parameters.

Source code in src/skai/core/_backend.py

def sends_video_audio_now(self) -> bool:
    """Return whether video requests include the audio track with the current settings.

    Returns:
        The class flag `sends_video_audio`, unless a subclass makes it depend on its parameters.
    """
    return self.sends_video_audio

supported_tasks() classmethod

Return the tasks that the backend implements, without classify_audio when it does not accept audio.

Returns:

Type	Description
frozenset[str]	The names of the tasks.

Source code in src/skai/core/_backend.py

@classmethod
def supported_tasks(cls) -> frozenset[str]:
    """Return the tasks that the backend implements, without `classify_audio` when it does not accept audio.

    Returns:
        The names of the tasks.
    """
    tasks = super().supported_tasks()
    if cls.accepts_audio or cls.classify_audio is not PromptedBackend.classify_audio:
        return tasks
    return tasks - {'classify_audio'}

supports(task, input_kind=None) classmethod

Return whether the backend runs a task on samples of an input kind, refusing prompted audio without audio.

Parameters:

Name	Type	Description	Default
task	str	The name of the task.	required
input_kind	str | None	The input kind of the samples, or None for any.	None

Returns:

Type	Description
bool	Whether the task is supported.

Source code in src/skai/core/_backend.py

@classmethod
def supports(cls, task: str, input_kind: str | None = None) -> bool:
    """Return whether the backend runs a task on samples of an input kind, refusing prompted audio without audio.

    Args:
        task:
            The name of the task.

        input_kind:
            The input kind of the samples, or `None` for any.

    Returns:
        Whether the task is supported.
    """
    if not super().supports(task, input_kind):
        return False
    prompted = task in PROMPTED_TASKS and getattr(cls, task) is getattr(PromptedBackend, task)
    return not (prompted and input_kind == 'audio' and not cls.accepts_audio)

transcribe_image(request)

Transcribe the text of an image by prompting the vision model.

Parameters:

Name	Type	Description	Default
request	ImageTranscriptionRequest	The image transcription request.	required

Returns:

Type	Description
str	The text.

Source code in src/skai/core/_backend.py

def transcribe_image(self, request: ImageTranscriptionRequest) -> str:
    """Transcribe the text of an image by prompting the vision model.

    Args:
        request:
            The image transcription request.

    Returns:
        The text.
    """
    prompt = build_image_transcription_prompt(request)
    return self._prompt('transcribe_image', prompt, request.image).strip()

Region(label, score, box, mask=None) dataclass

Hold one region that a backend found in an image.

Parameters:

Name	Type	Description	Default
label	str	The kind of region.	required
score	float	The confidence of the backend.	required
box	tuple[int, int, int, int]	The (left, top, right, bottom) pixels of the region.	required
mask	Image | None	A grayscale image of the input size whose white pixels belong to the region, or None for boxes.	None

RegionRequest(image, labels=(), prompt=None) dataclass

Ask a backend to find regions of an image, as boxes or masks.

Parameters:

Name	Type	Description	Default
image	Image	The image to search.	required
labels	tuple[str, ...]	The kinds of regions to find.	()
prompt	str | None	A description of the regions to find, used when labels is empty.	None

SampleError

Bases: RuntimeError

Signal that processing one sample failed.

SampleFailureWarning

Bases: UserWarning

Warn that some samples failed and were replaced by missing values.

Segment(start, end, text, speaker=None) dataclass

Hold one timed piece of a transcript.

Parameters:

Name	Type	Description	Default
start	float	The start time in seconds.	required
end	float	The end time in seconds.	required
text	str	The transcribed text.	required
speaker	str | None	The label of the speaker, or None when speakers are not labeled.	None

SpeechRequest(text, voice=None, language=None, seed=None) dataclass

Ask a backend to speak a text.

Parameters:

Name	Type	Description	Default
text	str	The text to speak.	required
voice	str | Audio | None	A voice name of the backend, or reference audio of the voice to clone.	None
language	str | None	The language of the text.	None
seed	int | None	The seed for backends that support seeding.	None

TextGenerationRequest(sample, instructions, examples=(), max_length=None, glossary=()) dataclass

Ask a backend to produce text from a sample by following instructions.

Parameters:

Name	Type	Description	Default
sample	Sample	The text, image, audio, sampled video frames or multimodal sample.	required
instructions	str	What to produce.	required
examples	tuple[tuple[Any, str], ...]	Example pairs of a sample and its reference output.	()
max_length	int | None	The maximum number of words of the output, or None for no limit.	None
glossary	tuple[tuple[str, str], ...]	Pairs of a term and its required rendering.	()

ThumbnailRequest(frames) dataclass

Ask a backend to pick the frame that best represents a video.

Parameters:

Name	Type	Description	Default
frames	VideoFrames	The sampled frames of the video.	required

Transcript(text, segments=()) dataclass

Hold a transcript and its timed segments.

Parameters:

Name	Type	Description	Default
text	str	The full transcribed text.	required
segments	tuple[Segment, ...]	The timed segments or words.	()

TranscriptionRequest(audio, language=None, timestamps=None, vocabulary=()) dataclass

Ask a backend to transcribe speech.

Parameters:

Name	Type	Description	Default
audio	Audio	The speech to transcribe.	required
language	str | None	The language of the speech. It is detected when it is None.	None
timestamps	Timestamps	The granularity of the timestamps, or None for plain text.	None
vocabulary	tuple[str, ...]	Domain terms that the speech may contain, given to the model as context for spelling.	()

UnsupportedBackendError

Bases: ValueError

Signal that a backend is unknown or cannot run a task.

UnusedFitDataWarning

Bases: UserWarning

Warn that the backend cannot use something the component learned in fit.

Video(frames=None, fps=None, audio=None, *, data=None, subtitles=None)

Hold a video as decoded frames or as encoded MP4 bytes.

Either frames and fps or data must be given. The other representation is computed when it is first needed, which requires the video extra.

Parameters:

Name	Type	Description	Default
frames	NDArray[uint8] | None	The RGB frames, with shape (n_frames, height, width, 3) and type uint8.	None
fps	float | None	The number of frames per second.	None
audio	Audio | None	The audio track.	None
data	bytes | None	The MP4-encoded video.	None
subtitles	str | None	Subtitles in SRT format, written next to the video by save.	None

Raises:

Type	Description
ValueError	If neither frames with fps nor data are given.

Examples:

>>> import numpy as np
>>> from skai.core import Video
>>> video = Video(np.zeros((50, 64, 64, 3), dtype=np.uint8), fps=25)
>>> video.duration
2.0

Source code in src/skai/core/_media.py

def __init__(
    self,
    frames: NDArray[np.uint8] | None = None,
    fps: float | None = None,
    audio: Audio | None = None,
    *,
    data: bytes | None = None,
    subtitles: str | None = None,
) -> None:
    if data is None and (frames is None or fps is None):
        msg = 'Provide either `frames` and `fps`, or the encoded `data`.'
        raise ValueError(msg)
    self._frames = frames
    self._fps = fps
    self._audio = audio
    self._data = data
    self.subtitles = subtitles

audio property

The audio track of the video.

duration property

The length of the video in seconds.

fps property

The number of frames per second.

frames property

The RGB frames of the video.

__repr__()

Return a short description of the video.

Source code in src/skai/core/_media.py

def __repr__(self) -> str:
    """Return a short description of the video."""
    if self._frames is None:
        return f'Video(encoded_bytes={len(self._data or b"")})'
    return f'Video(duration={self.duration:.2f}s, fps={self._fps})'

audio_track()

Return the audio track mixed to mono, or silence as long as the video when it has no track.

Returns:

Type	Description
Audio	The mono audio track.

Source code in src/skai/core/_media.py

def audio_track(self) -> Audio:
    """Return the audio track mixed to mono, or silence as long as the video when it has no track.

    Returns:
        The mono audio track.
    """
    if self.audio is None:
        return Audio(np.zeros(round(self.duration * SILENCE_SAMPLE_RATE), dtype=np.float32), SILENCE_SAMPLE_RATE)
    return self.audio.to_mono()

resize(size)

Return a copy whose frames are resized to a width and height.

Parameters:

Name	Type	Description	Default
size	tuple[int, int]	The width and height in pixels.	required

Returns:

Type	Description
Video	The resized video, or the video itself when it already has that size.

Source code in src/skai/core/_media.py

def resize(self, size: tuple[int, int]) -> 'Video':
    """Return a copy whose frames are resized to a width and height.

    Args:
        size:
            The width and height in pixels.

    Returns:
        The resized video, or the video itself when it already has that size.
    """
    if (self.frames.shape[2], self.frames.shape[1]) == tuple(size):
        return self
    frames = np.stack([np.asarray(Image.fromarray(frame).resize(size)) for frame in self.frames])
    return Video(frames, self.fps, self.audio)

save(path)

Write the video to an MP4 file, and its subtitles to an SRT file of the same name when it has them.

Parameters:

Name	Type	Description	Default
path	str | Path	The path of the file to write.	required

Returns:

Type	Description
Path	The path of the written video file.

Source code in src/skai/core/_media.py

def save(self, path: str | Path) -> Path:
    """Write the video to an MP4 file, and its subtitles to an SRT file of the same name when it has them.

    Args:
        path:
            The path of the file to write.

    Returns:
        The path of the written video file.
    """
    path = Path(path)
    path.write_bytes(self.to_bytes())
    if self.subtitles is not None:
        path.with_suffix('.srt').write_text(self.subtitles, encoding='utf-8')
    return path

to_bytes()

Encode the video as MP4 bytes.

Returns:

Type	Description
bytes	The MP4-encoded video.

Source code in src/skai/core/_media.py

def to_bytes(self) -> bytes:
    """Encode the video as MP4 bytes.

    Returns:
        The MP4-encoded video.
    """
    if self._data is None:
        self._data = _encode_video(self.frames, self.fps, self._audio)
    return self._data

trim(seconds)

Return a copy cut to at most a number of seconds.

Parameters:

Name	Type	Description	Default
seconds	float	The maximum length in seconds.	required

Returns:

Type	Description
Video	The trimmed video, or the video itself when it is not longer.

Source code in src/skai/core/_media.py

def trim(self, seconds: float) -> 'Video':
    """Return a copy cut to at most a number of seconds.

    Args:
        seconds:
            The maximum length in seconds.

    Returns:
        The trimmed video, or the video itself when it is not longer.
    """
    if self.duration <= seconds:
        return self
    frames = self.frames[: max(1, round(seconds * self.fps))]
    audio = self.audio
    if audio is not None:
        audio = Audio(audio.samples[: round(seconds * audio.sample_rate)], audio.sample_rate)
    return Video(frames, self.fps, audio)

VideoFrames(frames, timestamps, duration, audio=None) dataclass

Hold frames sampled from a video, their times and the audio track.

Parameters:

Name	Type	Description	Default
frames	tuple[Image, ...]	The sampled frames as PIL images.	required
timestamps	tuple[float, ...]	The time of every frame in seconds.	required
duration	float	The length of the video in seconds.	required
audio	Audio | None	The mono audio track of the video, or None when it has none.	None

VideoGenerationRequest(prompt, duration, size, fps=None, style=None, negative_prompt=None, seed=None) dataclass

Ask a backend to generate a video from a prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The description of the video.	required
duration	float	The length of the video in seconds.	required
size	tuple[int, int]	The width and height of the video in pixels.	required
fps	int | None	The number of frames per second, or None for the model's default.	None
style	str | None	A description of the visual style.	None
negative_prompt	str | None	A description of what the video must not show.	None
seed	int | None	The seed for backends that support seeding.	None

build_image_prompt(request)

Build the text prompt of an image generation model from a request.

Parameters:

Name	Type	Description	Default
request	ImageGenerationRequest	The image generation request.	required

Returns:

Type	Description
str	The prompt, with the style appended.

Source code in src/skai/core/_prompts.py

def build_image_prompt(request: ImageGenerationRequest) -> str:
    """Build the text prompt of an image generation model from a request.

    Args:
        request:
            The image generation request.

    Returns:
        The prompt, with the style appended.
    """
    return f'{request.prompt}. Style: {request.style}' if request.style else request.prompt

build_task_prompt(task, request)

Build the prompt that a prompted backend sends for a task and a request.

Parameters:

Name	Type	Description	Default
task	str	The name of the task.	required
request	ClassificationRequest | TextGenerationRequest | ExtractionRequest	The request of the task.	required

Returns:

Type	Description
Prompt	The prompt.

Raises:

Type	Description
ValueError	If the task is not answered by prompting.

Source code in src/skai/core/_prompts.py

def build_task_prompt(task: str, request: ClassificationRequest | TextGenerationRequest | ExtractionRequest) -> Prompt:
    """Build the prompt that a prompted backend sends for a task and a request.

    Args:
        task:
            The name of the task.

        request:
            The request of the task.

    Returns:
        The prompt.

    Raises:
        ValueError:
            If the task is not answered by prompting.
    """
    builders: dict[str, Callable[[Any], Prompt]] = {
        'classify_text': build_classification_prompt,
        'classify_image': build_classification_prompt,
        'classify_audio': build_classification_prompt,
        'classify_video': build_classification_prompt,
        'classify_multimodal': build_classification_prompt,
        'generate_text': build_text_generation_prompt,
        'extract_fields': build_extraction_prompt,
    }
    if task not in builders:
        msg = f'The task `{task}` is not answered by prompting.'
        raise ValueError(msg)
    return builders[task](request)

check_size(size, owner)

Check that a size parameter is a pair of positive integers.

Parameters:

Name	Type	Description	Default
size	object	The value of the size parameter.	required
owner	str	The name of the component, used in the error message.	required

Returns:

Type	Description
tuple[int, int]	The width and height.

Raises:

Type	Description
InvalidParameterError	If the size is not a (width, height) pair of positive integers.

Source code in src/skai/core/_utils.py

def check_size(size: object, owner: str) -> tuple[int, int]:
    """Check that a size parameter is a pair of positive integers.

    Args:
        size:
            The value of the `size` parameter.

        owner:
            The name of the component, used in the error message.

    Returns:
        The width and height.

    Raises:
        InvalidParameterError:
            If the size is not a `(width, height)` pair of positive integers.
    """
    if not (
        isinstance(size, tuple)
        and len(size) == SIZE_LENGTH
        and all(isinstance(side, Integral) and int(side) > 0 for side in size)
    ):
        msg = f'The `size` parameter of {owner} must be a (width, height) pair of positive integers. Got {size}.'
        raise InvalidParameterError(msg)
    return int(size[0]), int(size[1])

collage(images, columns=None, tile=256)

Lay images out in a grid of square tiles.

Parameters:

Name	Type	Description	Default
images	list[Image]	The images.	required
columns	int | None	The number of columns, the square root of the number of images rounded up when it is None.	None
tile	int	The side of every tile in pixels.	256

Returns:

Type	Description
Image	The collage, with black where there are no images.

Source code in src/skai/core/_local.py

def collage(images: list[Image.Image], columns: int | None = None, tile: int = 256) -> Image.Image:
    """Lay images out in a grid of square tiles.

    Args:
        images:
            The images.

        columns:
            The number of columns, the square root of the number of images rounded up when it is `None`.

        tile:
            The side of every tile in pixels.

    Returns:
        The collage, with black where there are no images.
    """
    columns = columns or max(1, math.ceil(math.sqrt(len(images))))
    rows = max(1, math.ceil(len(images) / columns))
    canvas = Image.new('RGB', (columns * tile, rows * tile))
    for index, image in enumerate(images):
        thumbnail = image.convert('RGB')
        thumbnail.thumbnail((tile, tile))
        row, column = divmod(index, columns)
        left = column * tile + (tile - thumbnail.width) // 2
        top = row * tile + (tile - thumbnail.height) // 2
        canvas.paste(thumbnail, (left, top))
    return canvas

concatenate_audio(tracks, gap=0.0)

Join audio tracks in order, mixed to mono at the rate of the first one.

Parameters:

Name	Type	Description	Default
tracks	list[Audio]	The tracks to join.	required
gap	float	The seconds of silence between consecutive tracks.	0.0

Returns:

Type	Description
Audio	The joined audio.

Source code in src/skai/core/_local.py

def concatenate_audio(tracks: list[Audio], gap: float = 0.0) -> Audio:
    """Join audio tracks in order, mixed to mono at the rate of the first one.

    Args:
        tracks:
            The tracks to join.

        gap:
            The seconds of silence between consecutive tracks.

    Returns:
        The joined audio.
    """
    rate = _rate_of(tracks)
    silence = np.zeros(round(gap * rate), dtype=np.float32)
    parts: list[NDArray[np.float32]] = []
    for position, track in enumerate(tracks):
        if position and silence.size:
            parts.append(silence)
        parts.append(track.resample(rate).samples.astype(np.float32))
    return Audio(np.concatenate(parts) if parts else np.zeros(0, dtype=np.float32), rate)

concatenate_video(clips)

Join video clips in order, at the size and frame rate of the first one.

Parameters:

Name	Type	Description	Default
clips	list[Video]	The clips to join.	required

Returns:

Type	Description
Video	The joined video, with the joined audio tracks when any clip has one.

Raises:

Type	Description
ValueError	If no clips are given.

Source code in src/skai/core/_local.py

def concatenate_video(clips: list[Video]) -> Video:
    """Join video clips in order, at the size and frame rate of the first one.

    Args:
        clips:
            The clips to join.

    Returns:
        The joined video, with the joined audio tracks when any clip has one.

    Raises:
        ValueError:
            If no clips are given.
    """
    if not clips:
        msg = 'Concatenation needs at least one clip.'
        raise ValueError(msg)
    first = clips[0]
    height, width = first.frames.shape[1:3]
    fps = first.fps
    frames = []
    for clip in clips:
        resized = clip.resize((width, height))
        count = max(1, round(clip.duration * fps))
        indices = np.minimum((np.arange(count) * clip.fps / fps).astype(int), resized.frames.shape[0] - 1)
        frames.append(resized.frames[indices])
    audio = None
    if any(clip.audio is not None for clip in clips):
        audio = concatenate_audio([_audio_of(clip) for clip in clips])
    return Video(np.concatenate(frames), fps, audio)

crop_to_aspect(video, aspect_ratio, centers)

Crop a video to an aspect ratio, following horizontal subject centers over time.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
aspect_ratio	float	The width divided by the height of the output.	required
centers	list[float]	The horizontal center of the subject, from 0 to 1, at evenly spaced times. An empty list centers the crop.	required

Returns:

Type	Description
Video	The cropped video.

Source code in src/skai/core/_local.py

def crop_to_aspect(video: Video, aspect_ratio: float, centers: list[float]) -> Video:
    """Crop a video to an aspect ratio, following horizontal subject centers over time.

    Args:
        video:
            The video.

        aspect_ratio:
            The width divided by the height of the output.

        centers:
            The horizontal center of the subject, from 0 to 1, at evenly spaced times. An empty list centers the crop.

    Returns:
        The cropped video.
    """
    frames = video.frames
    count, height, width = frames.shape[:3]
    crop_width = min(width, round(height * aspect_ratio))
    crop_height = min(height, round(crop_width / aspect_ratio))
    times = np.linspace(0, 1, num=max(1, len(centers)))
    positions = np.interp(np.linspace(0, 1, num=count), times, centers or [0.5])
    top = (height - crop_height) // 2
    cropped = []
    for frame, center in zip(frames, positions, strict=True):
        left = int(np.clip(round(center * width - crop_width / 2), 0, width - crop_width))
        cropped.append(frame[top : top + crop_height, left : left + crop_width])
    return Video(np.stack(cropped), video.fps, video.audio)

cut_audio(audio, start, end)

Return the part of audio between two times.

Parameters:

Name	Type	Description	Default
audio	Audio	The audio.	required
start	float	The start time in seconds.	required
end	float	The end time in seconds.	required

Returns:

Type	Description
Audio	The part of the audio.

Source code in src/skai/core/_local.py

def cut_audio(audio: Audio, start: float, end: float) -> Audio:
    """Return the part of audio between two times.

    Args:
        audio:
            The audio.

        start:
            The start time in seconds.

        end:
            The end time in seconds.

    Returns:
        The part of the audio.
    """
    first, last = round(start * audio.sample_rate), round(end * audio.sample_rate)
    return Audio(audio.samples[first:last], audio.sample_rate)

cut_video(video, start, end)

Return the part of a video between two times, with its audio.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
start	float	The start time in seconds.	required
end	float	The end time in seconds.	required

Returns:

Type	Description
Video	The part of the video, with at least one frame.

Source code in src/skai/core/_local.py

def cut_video(video: Video, start: float, end: float) -> Video:
    """Return the part of a video between two times, with its audio.

    Args:
        video:
            The video.

        start:
            The start time in seconds.

        end:
            The end time in seconds.

    Returns:
        The part of the video, with at least one frame.
    """
    count = video.frames.shape[0]
    first = min(count - 1, max(0, round(start * video.fps)))
    last = max(first + 1, min(count, round(end * video.fps)))
    audio = None if video.audio is None else cut_audio(video.audio, first / video.fps, last / video.fps)
    return Video(video.frames[first:last], video.fps, audio)

draw_seeds(random_state, count)

Draw the seeds that components send to backends.

Parameters:

Name	Type	Description	Default
random_state	int | RandomState | None	The random_state parameter of the component.	required
count	int	The number of seeds.	required

Returns:

Type	Description
list[int]	The seeds, reproducible for an integer random_state.

Source code in src/skai/core/_utils.py

def draw_seeds(random_state: int | np.random.RandomState | None, count: int) -> list[int]:
    """Draw the seeds that components send to backends.

    Args:
        random_state:
            The `random_state` parameter of the component.

        count:
            The number of seeds.

    Returns:
        The seeds, reproducible for an integer `random_state`.
    """
    return [int(seed) for seed in check_random_state(random_state).randint(0, MAX_SEED, size=count)]

encode_png(image)

Encode an image as PNG bytes.

Parameters:

Name	Type	Description	Default
image	Image	The image to encode.	required

Returns:

Type	Description
bytes	The PNG-encoded image.

Source code in src/skai/core/_io.py

def encode_png(image: Image.Image) -> bytes:
    """Encode an image as PNG bytes.

    Args:
        image:
            The image to encode.

    Returns:
        The PNG-encoded image.
    """
    buffer = io.BytesIO()
    image.save(buffer, format='PNG')
    return buffer.getvalue()

extract_audio(video)

Return the mono audio track of a video, or silence as long as the video.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required

Returns:

Type	Description
Audio	The audio track.

Source code in src/skai/core/_local.py

def extract_audio(video: Video) -> Audio:
    """Return the mono audio track of a video, or silence as long as the video.

    Args:
        video:
            The video.

    Returns:
        The audio track.
    """
    return video.audio_track()

find_terms(text)

Find the terms of a text that look like names, products or technical terms.

A term is a token of two or more characters that contains a digit, has an upper-case letter after its first character, or starts with an upper-case letter in the middle of a sentence.

Parameters:

Name	Type	Description	Default
text	str	The text to search.	required

Returns:

Type	Description
list[str]	The terms in order of first appearance, with repeats.

Examples:

>>> from skai.core import find_terms
>>> find_terms('We deploy on Kubernetes with gRPC and Python 3.12.')
['Kubernetes', 'gRPC', 'Python', '3.12']

Source code in src/skai/core/_terms.py

def find_terms(text: str) -> list[str]:
    """Find the terms of a text that look like names, products or technical terms.

    A term is a token of two or more characters that contains a digit, has an upper-case letter after its first
    character, or starts with an upper-case letter in the middle of a sentence.

    Args:
        text:
            The text to search.

    Returns:
        The terms in order of first appearance, with repeats.

    Examples:
        >>> from skai.core import find_terms
        >>> find_terms('We deploy on Kubernetes with gRPC and Python 3.12.')
        ['Kubernetes', 'gRPC', 'Python', '3.12']
    """
    terms = []
    for sentence in SENTENCE_END.split(text):
        for position, match in enumerate(TOKEN.finditer(sentence)):
            if _is_term(match.group(), starts_sentence=position == 0):
                terms.append(match.group())
    return terms

frame_at(video, at)

Return the frame of a video at a time.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
at	float	The time in seconds, clamped to the video.	required

Returns:

Type	Description
Image	The frame.

Source code in src/skai/core/_local.py

def frame_at(video: Video, at: float) -> Image.Image:
    """Return the frame of a video at a time.

    Args:
        video:
            The video.

        at:
            The time in seconds, clamped to the video.

    Returns:
        The frame.
    """
    index = min(video.frames.shape[0] - 1, max(0, int(at * video.fps)))
    return Image.fromarray(video.frames[index])

import_optional(module, extra)

Import a module that belongs to an optional dependency group.

Parameters:

Name	Type	Description	Default
module	str	The name of the module to import.	required
extra	str	The name of the optional dependency group that provides the module.	required

Returns:

Type	Description
ModuleType	The imported module.

Raises:

Type	Description
MissingExtraError	If the module is not installed.

Source code in src/skai/core/_utils.py

def import_optional(module: str, extra: str) -> ModuleType:
    """Import a module that belongs to an optional dependency group.

    Args:
        module:
            The name of the module to import.

        extra:
            The name of the optional dependency group that provides the module.

    Returns:
        The imported module.

    Raises:
        MissingExtraError:
            If the module is not installed.
    """
    try:
        return importlib.import_module(module)
    except ModuleNotFoundError as error:
        msg = f'Module `{module}` is required. Install it with `pip install scikit-ai[{extra}]`.'
        raise MissingExtraError(msg) from error

input_kind_of(sample)

Return the input kind of a read sample.

Parameters:

Name	Type	Description	Default
sample	object	A text, image, audio, sampled video frames or multimodal sample.	required

Returns:

Type	Description
str	One of 'text', 'image', 'audio', 'video' and 'multimodal'.

Source code in src/skai/core/_backend.py

def input_kind_of(sample: object) -> str:
    """Return the input kind of a read sample.

    Args:
        sample:
            A text, image, audio, sampled video frames or multimodal sample.

    Returns:
        One of `'text'`, `'image'`, `'audio'`, `'video'` and `'multimodal'`.
    """
    if isinstance(sample, str):
        return 'text'
    if isinstance(sample, Audio):
        return 'audio'
    if isinstance(sample, VideoFrames | Video):
        return 'video'
    if isinstance(sample, dict):
        return 'multimodal'
    return 'image'

list_backends(task=None, predicate=None, input_kind=None)

List the names of the registered backends.

Parameters:

Name	Type	Description	Default
task	str | None	Keep only the backends that support this task.	None
predicate	Callable[[type[Backend]], bool] | None	Keep only the backend classes for which this function returns True.	None
input_kind	str | None	Keep only the backends that run the task on samples of this input kind.	None

Returns:

Type	Description
list[str]	The sorted backend names.

Examples:

>>> from skai.core import list_backends
>>> 'dummy' in list_backends('generate_text')
True

Source code in src/skai/core/_registry.py

def list_backends(
    task: str | None = None,
    predicate: Callable[[type[Backend]], bool] | None = None,
    input_kind: str | None = None,
) -> list[str]:
    """List the names of the registered backends.

    Args:
        task:
            Keep only the backends that support this task.

        predicate:
            Keep only the backend classes for which this function returns `True`.

        input_kind:
            Keep only the backends that run the task on samples of this input kind.

    Returns:
        The sorted backend names.

    Examples:
        >>> from skai.core import list_backends
        >>> 'dummy' in list_backends('generate_text')
        True
    """
    _load_entry_points()
    return sorted(
        name
        for name, backend_class in BACKENDS.items()
        if (task is None or backend_class.supports(task, input_kind))
        and (predicate is None or predicate(backend_class))
    )

mix_audio(tracks, *, volumes=None, duck_under=None, duck_for=None, duck_level=0.25, threshold=0.01)

Mix named tracks, lowering some of them while another one is not silent.

Parameters:

Name	Type	Description	Default
tracks	dict[str, Audio]	The tracks to mix, by name.	required
volumes	dict[str, float] | None	The gain of every track, one when it is missing.	None
duck_under	list[str] | None	The names of the tracks to lower.	None
duck_for	str | None	The name of the track whose sound lowers the others.	None
duck_level	float	The gain of the lowered tracks while duck_for sounds.	0.25
threshold	float	The RMS level above which duck_for counts as sounding.	0.01

Returns:

Type	Description
Audio	A mono track as long as the longest one, clipped to [-1, 1].

Source code in src/skai/core/_local.py

def mix_audio(
    tracks: dict[str, Audio],
    *,
    volumes: dict[str, float] | None = None,
    duck_under: list[str] | None = None,
    duck_for: str | None = None,
    duck_level: float = 0.25,
    threshold: float = 0.01,
) -> Audio:
    """Mix named tracks, lowering some of them while another one is not silent.

    Args:
        tracks:
            The tracks to mix, by name.

        volumes:
            The gain of every track, one when it is missing.

        duck_under:
            The names of the tracks to lower.

        duck_for:
            The name of the track whose sound lowers the others.

        duck_level:
            The gain of the lowered tracks while `duck_for` sounds.

        threshold:
            The RMS level above which `duck_for` counts as sounding.

    Returns:
        A mono track as long as the longest one, clipped to `[-1, 1]`.
    """
    rate = _rate_of(list(tracks.values()))
    mono = {name: track.resample(rate).samples.astype(np.float64) for name, track in tracks.items()}
    length = max((samples.shape[0] for samples in mono.values()), default=0)
    volumes = volumes or {}
    gains = np.ones(length)
    if duck_for is not None and duck_for in mono and duck_under:
        sounding = _envelope(_fit_length(mono[duck_for], length), rate) > threshold
        weight = np.convolve(sounding.astype(float), np.ones(max(1, round(SMOOTHING_SECONDS * rate))), mode='same')
        gains = 1.0 - (1.0 - duck_level) * np.clip(weight / max(1, round(SMOOTHING_SECONDS * rate)), 0, 1)
    mixed = np.zeros(length)
    for name, samples in mono.items():
        track = _fit_length(samples, length).astype(np.float64) * volumes.get(name, 1.0)
        mixed += track * gains if duck_under and name in duck_under else track
    return Audio(np.clip(mixed, -1.0, 1.0).astype(np.float32), rate)

mux(video, audio)

Replace the soundtrack of a video, cut or padded to the length of the video.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
audio	Audio	The new soundtrack.	required

Returns:

Type	Description
Video	The video with the new soundtrack.

Source code in src/skai/core/_local.py

def mux(video: Video, audio: Audio) -> Video:
    """Replace the soundtrack of a video, cut or padded to the length of the video.

    Args:
        video:
            The video.

        audio:
            The new soundtrack.

    Returns:
        The video with the new soundtrack.
    """
    mono = audio.to_mono()
    samples = _fit_length(mono.samples, round(video.duration * mono.sample_rate))
    return Video(video.frames, video.fps, Audio(samples, mono.sample_rate), subtitles=video.subtitles)

place_audio(clips, length=0.0)

Place audio clips at their start times on one silent track.

Parameters:

Name	Type	Description	Default
clips	list[tuple[float, Audio]]	Pairs of a start time in seconds and a clip. Overlapping clips are added.	required
length	float	The minimum length of the track in seconds.	0.0

Returns:

Type	Description
Audio	A mono track at the rate of the first clip, long enough for every clip.

Source code in src/skai/core/_local.py

def place_audio(clips: list[tuple[float, Audio]], length: float = 0.0) -> Audio:
    """Place audio clips at their start times on one silent track.

    Args:
        clips:
            Pairs of a start time in seconds and a clip. Overlapping clips are added.

        length:
            The minimum length of the track in seconds.

    Returns:
        A mono track at the rate of the first clip, long enough for every clip.
    """
    rate = _rate_of([clip for _, clip in clips])
    mono = [(round(start * rate), clip.resample(rate).samples.astype(np.float64)) for start, clip in clips]
    size = max([round(length * rate), *(start + samples.shape[0] for start, samples in mono)])
    track = np.zeros(size)
    for start, samples in mono:
        track[start : start + samples.shape[0]] += samples
    return Audio(np.clip(track, -1.0, 1.0).astype(np.float32), rate)

read_audio(sample, index=0)

Read audio from a path, a URL, bytes, an Audio object or a (samples, sample_rate) tuple.

Parameters:

Name	Type	Description	Default
sample	object	The audio to read.	required
index	int	The position of the sample in the input.	0

Returns:

Type	Description
Audio	The audio.

Raises:

Type	Description
MediaReadError	If the sample cannot be read as audio.

Source code in src/skai/core/_io.py

def read_audio(sample: object, index: int = 0) -> Audio:
    """Read audio from a path, a URL, bytes, an `Audio` object or a `(samples, sample_rate)` tuple.

    Args:
        sample:
            The audio to read.

        index:
            The position of the sample in the input.

    Returns:
        The audio.

    Raises:
        MediaReadError:
            If the sample cannot be read as audio.
    """
    if isinstance(sample, Audio):
        return sample
    if isinstance(sample, tuple) and len(sample) == AUDIO_TUPLE_LENGTH:
        samples, sample_rate = sample
        return Audio(np.asarray(samples, dtype=np.float32), int(sample_rate))
    if not isinstance(sample, str | bytes | os.PathLike):
        msg = f'Sample {index}: cannot read audio from {type(sample).__name__}.'
        raise MediaReadError(msg)
    try:
        return Audio.from_bytes(_read_source(sample, index))
    except sf.LibsndfileError as error:
        msg = f'Sample {index}: cannot decode the audio ({error}).'
        raise MediaReadError(msg) from error

read_env(name)

Read the value of the environment variable that holds a credential.

Parameters:

Name	Type	Description	Default
name	str	The name of the environment variable.	required

Returns:

Type	Description
str	The value of the environment variable.

Raises:

Type	Description
MissingCredentialError	If the environment variable is not set or is empty.

Source code in src/skai/core/_utils.py

def read_env(name: str) -> str:
    """Read the value of the environment variable that holds a credential.

    Args:
        name:
            The name of the environment variable.

    Returns:
        The value of the environment variable.

    Raises:
        MissingCredentialError:
            If the environment variable is not set or is empty.
    """
    value = os.environ.get(name)
    if not value:
        msg = f'Set the environment variable `{name}` to the credential of the backend.'
        raise MissingCredentialError(msg)
    return value

read_image(sample, index=0)

Read an image from a path, a URL, bytes, a PIL image or a NumPy array.

Parameters:

Name	Type	Description	Default
sample	object	The image to read.	required
index	int	The position of the sample in the input.	0

Returns:

Type	Description
Image	The image.

Raises:

Type	Description
MediaReadError	If the sample cannot be read as an image.

Source code in src/skai/core/_io.py

def read_image(sample: object, index: int = 0) -> Image.Image:
    """Read an image from a path, a URL, bytes, a PIL image or a NumPy array.

    Args:
        sample:
            The image to read.

        index:
            The position of the sample in the input.

    Returns:
        The image.

    Raises:
        MediaReadError:
            If the sample cannot be read as an image.
    """
    if isinstance(sample, Image.Image):
        return sample
    if isinstance(sample, np.ndarray):
        try:
            return Image.fromarray(sample)
        except (TypeError, ValueError) as error:
            msg = f'Sample {index}: cannot convert an array of shape {sample.shape} to an image ({error}).'
            raise MediaReadError(msg) from error
    if not isinstance(sample, str | bytes | os.PathLike):
        msg = f'Sample {index}: cannot read an image from {type(sample).__name__}.'
        raise MediaReadError(msg)
    try:
        image = Image.open(io.BytesIO(_read_source(sample, index)))
        image.load()
    except (UnidentifiedImageError, OSError) as error:
        msg = f'Sample {index}: cannot decode the image ({error}).'
        raise MediaReadError(msg) from error
    return image

read_input(kind, sample, index, n_frames=None)

Read one sample of an input kind.

Parameters:

Name	Type	Description	Default
kind	InputKind	The input kind.	required
sample	object	The sample.	required
index	int	The position of the sample in the input.	required
n_frames	int | None	The number of frames to sample from videos, or None to keep videos whole.	None

Returns:

Type	Description
object	The read sample.

Source code in src/skai/core/_io.py

def read_input(kind: InputKind, sample: object, index: int, n_frames: int | None = None) -> object:
    """Read one sample of an input kind.

    Args:
        kind:
            The input kind.

        sample:
            The sample.

        index:
            The position of the sample in the input.

        n_frames:
            The number of frames to sample from videos, or `None` to keep videos whole.

    Returns:
        The read sample.
    """
    if kind == 'text':
        return check_text(sample, index)
    if kind == 'image':
        return read_image(sample, index)
    if kind == 'audio':
        return read_audio(sample, index)
    if kind == 'video':
        return read_video(sample, index) if n_frames is None else read_video_frames(sample, index, n_frames)
    return read_multimodal(sample, index, n_frames)

read_multimodal(sample, index=0, n_frames=None, required=())

Read a multimodal sample, a dict of named texts, images, audio, videos or transcripts.

Strings are read as texts, unless they are URLs or paths of existing files with a media extension.

Parameters:

Name	Type	Description	Default
sample	object	The dict, or a pandas row, to read.	required
index	int	The position of the sample in the input.	0
n_frames	int | None	The number of frames to sample from every video, or None to keep videos whole.	None
required	tuple[str, ...]	The keys the sample must have.	()

Returns:

Type	Description
dict[str, Any]	The sample, with every part read by its own kind.

Raises:

Type	Description
InvalidSampleError	If the sample is not a dict or misses a required key.

Source code in src/skai/core/_io.py

def read_multimodal(
    sample: object,
    index: int = 0,
    n_frames: int | None = None,
    required: tuple[str, ...] = (),
) -> dict[str, Any]:
    """Read a multimodal sample, a dict of named texts, images, audio, videos or transcripts.

    Strings are read as texts, unless they are URLs or paths of existing files with a media extension.

    Args:
        sample:
            The dict, or a pandas row, to read.

        index:
            The position of the sample in the input.

        n_frames:
            The number of frames to sample from every video, or `None` to keep videos whole.

        required:
            The keys the sample must have.

    Returns:
        The sample, with every part read by its own kind.

    Raises:
        InvalidSampleError:
            If the sample is not a dict or misses a required key.
    """
    if hasattr(sample, 'to_dict') and not isinstance(sample, dict):
        sample = sample.to_dict()
    if not isinstance(sample, dict):
        msg = f'Sample {index} must be a dict of named media, but got {type(sample).__name__}.'
        raise InvalidSampleError(msg)
    missing = [key for key in required if key not in sample]
    if missing:
        msg = f'Sample {index} has no `{missing[0]}` key. The keys {list(required)} are required.'
        raise InvalidSampleError(msg)
    return {str(key): _read_part(value, index, n_frames) for key, value in sample.items()}

read_video(sample, index=0)

Read a video from a path, a URL, bytes or a Video object.

Parameters:

Name	Type	Description	Default
sample	object	The video to read.	required
index	int	The position of the sample in the input.	0

Returns:

Type	Description
Video	The video.

Raises:

Type	Description
MediaReadError	If the sample cannot be read as a video.

Source code in src/skai/core/_io.py

def read_video(sample: object, index: int = 0) -> Video:
    """Read a video from a path, a URL, bytes or a `Video` object.

    Args:
        sample:
            The video to read.

        index:
            The position of the sample in the input.

    Returns:
        The video.

    Raises:
        MediaReadError:
            If the sample cannot be read as a video.
    """
    if isinstance(sample, Video):
        return sample
    if not isinstance(sample, str | bytes | os.PathLike):
        msg = f'Sample {index}: cannot read a video from {type(sample).__name__}.'
        raise MediaReadError(msg)
    return Video(data=_read_source(sample, index))

read_video_frames(sample, index, n_frames)

Read a video and sample evenly spaced frames of it.

Parameters:

Name	Type	Description	Default
sample	object	The video to read, or frames that were already sampled.	required
index	int	The position of the sample in the input.	required
n_frames	int	The number of frames to sample.	required

Returns:

Type	Description
VideoFrames	The sampled frames.

Raises:

Type	Description
MediaReadError	If the video cannot be decoded.

Source code in src/skai/core/_io.py

def read_video_frames(sample: object, index: int, n_frames: int) -> VideoFrames:
    """Read a video and sample evenly spaced frames of it.

    Args:
        sample:
            The video to read, or frames that were already sampled.

        index:
            The position of the sample in the input.

        n_frames:
            The number of frames to sample.

    Returns:
        The sampled frames.

    Raises:
        MediaReadError:
            If the video cannot be decoded.
    """
    if isinstance(sample, VideoFrames):
        return sample
    video = read_video(sample, index)
    try:
        return sample_frames(video, n_frames)
    except (OSError, ValueError) as error:
        msg = f'Sample {index}: cannot decode the video ({error}).'
        raise MediaReadError(msg) from error

register(name, backend_class)

Register a backend class under a name, so the name can be used in backend strings.

Parameters:

Name	Type	Description	Default
name	str	The provider name, used as '<name>' or '<name>:<model>'.	required
backend_class	type[Backend]	The backend class.	required

Raises:

Type	Description
TypeError	If backend_class is not a subclass of Backend.

Examples:

>>> from skai.core import Backend, register
>>> class EchoBackend(Backend):
...     def generate_text(self, request):
...         return request.sample
>>> register('echo', EchoBackend)

Source code in src/skai/core/_registry.py

def register(name: str, backend_class: type[Backend]) -> None:
    """Register a backend class under a name, so the name can be used in `backend` strings.

    Args:
        name:
            The provider name, used as `'<name>'` or `'<name>:<model>'`.

        backend_class:
            The backend class.

    Raises:
        TypeError:
            If `backend_class` is not a subclass of `Backend`.

    Examples:
        >>> from skai.core import Backend, register
        >>> class EchoBackend(Backend):
        ...     def generate_text(self, request):
        ...         return request.sample
        >>> register('echo', EchoBackend)
    """
    if not (isinstance(backend_class, type) and issubclass(backend_class, Backend)):
        msg = f'The backend class of `{name}` must be a subclass of `Backend`.'
        raise TypeError(msg)
    BACKENDS[name] = backend_class

sample_frames(video, n_frames)

Sample evenly spaced frames of a video, starting with the first frame.

A video with fewer frames than n_frames gives all of its frames.

Parameters:

Name	Type	Description	Default
video	Video	The video to sample.	required
n_frames	int	The number of frames to sample.	required

Returns:

Type	Description
VideoFrames	The sampled frames, their times, the length of the video, and its mono audio track when it has one.

Source code in src/skai/core/_io.py

def sample_frames(video: Video, n_frames: int) -> VideoFrames:
    """Sample evenly spaced frames of a video, starting with the first frame.

    A video with fewer frames than `n_frames` gives all of its frames.

    Args:
        video:
            The video to sample.

        n_frames:
            The number of frames to sample.

    Returns:
        The sampled frames, their times, the length of the video, and its mono audio track when it has one.
    """
    frames = video.frames
    count = min(n_frames, frames.shape[0])
    indices = np.arange(count) * frames.shape[0] // count
    images = tuple(Image.fromarray(frames[index]) for index in indices)
    timestamps = tuple(float(index / video.fps) for index in indices)
    audio = None if video.audio is None else video.audio_track()
    return VideoFrames(images, timestamps, video.duration, audio)

shot_spans(video, threshold=0.5)

Find the shots of a video from the histogram distance between consecutive frames.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
threshold	float	The total variation distance between the color histograms of consecutive frames above which a cut is found, from 0 to 1.	0.5

Returns:

Type	Description
list[tuple[float, float]]	The (start, end) seconds of every shot, covering the video in order.

Source code in src/skai/core/_local.py

def shot_spans(video: Video, threshold: float = 0.5) -> list[tuple[float, float]]:
    """Find the shots of a video from the histogram distance between consecutive frames.

    Args:
        video:
            The video.

        threshold:
            The total variation distance between the color histograms of consecutive frames above which a cut is
            found, from 0 to 1.

    Returns:
        The `(start, end)` seconds of every shot, covering the video in order.
    """
    frames = video.frames
    histograms = []
    for frame in frames:
        counts = [np.histogram(frame[..., channel], bins=HISTOGRAM_BINS, range=(0, 256))[0] for channel in range(3)]
        histogram = np.concatenate(counts).astype(float)
        histograms.append(histogram / histogram.sum())
    cuts = [
        index
        for index in range(1, len(histograms))
        if 0.5 * np.abs(histograms[index] - histograms[index - 1]).sum() > threshold
    ]
    bounds = [0, *cuts, frames.shape[0]]
    return [(start / video.fps, end / video.fps) for start, end in pairwise(bounds)]

sound_spans(audio, threshold=0.01, min_silence=0.3)

Find the parts of audio between silences.

Parameters:

Name	Type	Description	Default
audio	Audio	The audio.	required
threshold	float	The RMS level below which audio counts as silent.	0.01
min_silence	float	The shortest silence in seconds that separates two parts.	0.3

Returns:

Type	Description
list[tuple[float, float]]	The (start, end) seconds of every part with sound, in order.

Source code in src/skai/core/_local.py

def sound_spans(audio: Audio, threshold: float = 0.01, min_silence: float = 0.3) -> list[tuple[float, float]]:
    """Find the parts of audio between silences.

    Args:
        audio:
            The audio.

        threshold:
            The RMS level below which audio counts as silent.

        min_silence:
            The shortest silence in seconds that separates two parts.

    Returns:
        The `(start, end)` seconds of every part with sound, in order.
    """
    mono = audio.to_mono()
    window = max(1, round(ENVELOPE_SECONDS * mono.sample_rate))
    count = mono.samples.shape[0] // window
    if count == 0:
        return []
    blocks = np.asarray(mono.samples[: count * window], dtype=np.float64).reshape(count, window)
    sounding = np.sqrt((blocks**2).mean(axis=1)) > threshold
    spans: list[tuple[int, int]] = []
    start = None
    for index, value in enumerate([*sounding.tolist(), False]):
        if value and start is None:
            start = index
        elif not value and start is not None:
            spans.append((start, index))
            start = None
    merged: list[tuple[int, int]] = []
    gap = min_silence * mono.sample_rate / window
    for span in spans:
        if merged and span[0] - merged[-1][1] < gap:
            merged[-1] = (merged[-1][0], span[1])
        else:
            merged.append(span)
    seconds = window / mono.sample_rate
    return [(start * seconds, min(mono.duration, end * seconds)) for start, end in merged]

subtitle(video, transcript, burn=True)

Show a transcript as subtitles, burned into the frames or attached in SRT format.

Parameters:

Name	Type	Description	Default
video	Video	The video.	required
transcript	Any	The transcript, in any form that to_cues accepts.	required
burn	bool	Whether to draw the subtitles on the frames. Otherwise they are attached as subtitles.	True

Returns:

Type	Description
Video	The subtitled video.

Source code in src/skai/core/_local.py

def subtitle(video: Video, transcript: Any, burn: bool = True) -> Video:
    """Show a transcript as subtitles, burned into the frames or attached in SRT format.

    Args:
        video:
            The video.

        transcript:
            The transcript, in any form that `to_cues` accepts.

        burn:
            Whether to draw the subtitles on the frames. Otherwise they are attached as `subtitles`.

    Returns:
        The subtitled video.
    """
    cues = to_cues(transcript, video.duration)
    if not burn:
        return Video(video.frames, video.fps, video.audio, subtitles=to_srt(cues))
    frames = video.frames.copy()
    font = ImageFont.load_default()
    for index in range(frames.shape[0]):
        time = index / video.fps
        texts = [text for start, end, text in cues if start <= time < end]
        if not texts:
            continue
        image = Image.fromarray(frames[index])
        draw = ImageDraw.Draw(image)
        lines = textwrap.wrap(' '.join(texts), SUBTITLE_WIDTH)
        line_height = draw.textbbox((0, 0), 'Ag', font=font)[3] + 2
        top = image.height - round(SUBTITLE_MARGIN * image.height) - line_height * len(lines)
        for number, line in enumerate(lines):
            width = draw.textlength(line, font=font)
            position = ((image.width - width) / 2, top + number * line_height)
            draw.text(position, line, fill='white', font=font, stroke_width=1, stroke_fill='black')
        frames[index] = np.asarray(image)
    return Video(frames, video.fps, video.audio, subtitles=to_srt(cues))

text_spans(text, by, chunk_size=1000)

Split a text into paragraphs, sentences or fixed-size chunks.

Parameters:

Name	Type	Description	Default
text	str	The text.	required
by	str	'paragraph', 'sentence' or 'chunk'.	required
chunk_size	int	The number of characters of a chunk.	1000

Returns:

Type	Description
list[tuple[int, int]]	The (start, end) character offsets of every piece. Chunks cover the text exactly.

Source code in src/skai/core/_local.py

def text_spans(text: str, by: str, chunk_size: int = 1000) -> list[tuple[int, int]]:
    """Split a text into paragraphs, sentences or fixed-size chunks.

    Args:
        text:
            The text.

        by:
            `'paragraph'`, `'sentence'` or `'chunk'`.

        chunk_size:
            The number of characters of a chunk.

    Returns:
        The `(start, end)` character offsets of every piece. Chunks cover the text exactly.
    """
    if by == 'chunk':
        return [(start, min(start + chunk_size, len(text))) for start in range(0, len(text), chunk_size)]
    pattern = PARAGRAPH if by == 'paragraph' else SENTENCE
    return [(match.start(), match.end()) for match in pattern.finditer(text) if match.group().strip()]

to_chat_messages(prompt)

Turn a prompt into chat messages in the OpenAI format, with images as data URLs and audio as WAV.

Parameters:

Name	Type	Description	Default
prompt	Prompt	The prompt.	required

Returns:

Type	Description
list[dict[str, Any]]	The system and user messages.

Source code in src/skai/core/_prompts.py

def to_chat_messages(prompt: Prompt) -> list[dict[str, Any]]:
    """Turn a prompt into chat messages in the OpenAI format, with images as data URLs and audio as WAV.

    Args:
        prompt:
            The prompt.

    Returns:
        The system and user messages.
    """
    content: list[dict[str, Any]] = [{'type': 'text', 'text': prompt.user}]
    content += [{'type': 'image_url', 'image_url': {'url': to_data_url(image)}} for image in prompt.images]
    content += [
        {'type': 'input_audio', 'input_audio': {'data': base64.b64encode(audio.to_bytes()).decode(), 'format': 'wav'}}
        for audio in prompt.audio
    ]
    return [{'role': 'system', 'content': prompt.system}, {'role': 'user', 'content': content}]

to_cues(transcript, duration)

Turn a transcript into timed subtitle cues.

Parameters:

Name	Type	Description	Default
transcript	Any	A string shown for the whole duration, a Transcript, or a list of dicts or segments with start, end and text.	required
duration	float	The length of the video in seconds.	required

Returns:

Type	Description
list[tuple[float, float, str]]	The cues as (start, end, text).

Source code in src/skai/core/_local.py

def to_cues(transcript: Any, duration: float) -> list[tuple[float, float, str]]:
    """Turn a transcript into timed subtitle cues.

    Args:
        transcript:
            A string shown for the whole duration, a `Transcript`, or a list of dicts or segments with `start`, `end`
            and `text`.

        duration:
            The length of the video in seconds.

    Returns:
        The cues as `(start, end, text)`.
    """
    if isinstance(transcript, Transcript):
        transcript = list(transcript.segments) or transcript.text
    if isinstance(transcript, str):
        return [(0.0, duration, transcript)] if transcript.strip() else []
    cues = []
    for item in transcript:
        values = item if isinstance(item, dict) else vars(item)
        cues.append((float(values['start']), float(values['end']), str(values.get('text', ''))))
    return cues

to_data_url(image)

Encode an image as a PNG data URL.

Parameters:

Name	Type	Description	Default
image	Image	The image to encode.	required

Returns:

Type	Description
str	The data URL.

Source code in src/skai/core/_io.py

def to_data_url(image: Image.Image) -> str:
    """Encode an image as a PNG data URL.

    Args:
        image:
            The image to encode.

    Returns:
        The data URL.
    """
    return f'data:image/png;base64,{base64.b64encode(encode_png(image.convert("RGB"))).decode()}'

to_object_array(items)

Pack items into a one-dimensional object array without broadcasting them.

Parameters:

Name	Type	Description	Default
items	Iterable[Any]	The items to pack.	required

Returns:

Type	Description
NDArray[object_]	The object array that holds the items.

Source code in src/skai/core/_io.py

def to_object_array(items: Iterable[Any]) -> NDArray[np.object_]:
    """Pack items into a one-dimensional object array without broadcasting them.

    Args:
        items:
            The items to pack.

    Returns:
        The object array that holds the items.
    """
    items = list(items)
    array = np.empty(len(items), dtype=object)
    array[:] = items
    return array

to_samples(X, rows=False)

Convert a container of samples into a list of samples.

Parameters:

Name	Type	Description	Default
X	Any	A list, tuple, NumPy array, pandas Series or single-column DataFrame of samples, or None.	required
rows	bool	Whether every row of a DataFrame is one multimodal sample, a dict keyed by column.	False

Returns:

Type	Description
list[Any]	The samples.

Raises:

Type	Description
ValueError	If X is two-dimensional with more than one column and rows is False.

Source code in src/skai/core/_io.py

def to_samples(X: Any, rows: bool = False) -> list[Any]:
    """Convert a container of samples into a list of samples.

    Args:
        X:
            A list, tuple, NumPy array, pandas Series or single-column DataFrame of samples, or `None`.

        rows:
            Whether every row of a DataFrame is one multimodal sample, a dict keyed by column.

    Returns:
        The samples.

    Raises:
        ValueError:
            If `X` is two-dimensional with more than one column and `rows` is `False`.
    """
    if X is None:
        return []
    if rows and hasattr(X, 'to_dict') and hasattr(X, 'columns'):
        return [{str(key): value for key, value in row.items()} for row in X.to_dict('records')]
    if hasattr(X, 'to_numpy') and not isinstance(X, np.ndarray):
        X = X.to_numpy()
    if not isinstance(X, np.ndarray):
        return list(X)
    if X.ndim == TABLE_NDIM and X.dtype.kind in 'USO':
        if X.shape[1] != 1:
            msg = f'Expected one column of samples, but got an array with {X.shape[1]} columns.'
            raise ValueError(msg)
        return list(X[:, 0])
    return list(X)

to_srt(cues)

Write subtitle cues in SRT format.

Parameters:

Name	Type	Description	Default
cues	list[tuple[float, float, str]]	The cues as (start, end, text).	required

Returns:

Type	Description
str	The SRT text.

Source code in src/skai/core/_local.py

def to_srt(cues: list[tuple[float, float, str]]) -> str:
    """Write subtitle cues in SRT format.

    Args:
        cues:
            The cues as `(start, end, text)`.

    Returns:
        The SRT text.
    """
    blocks = [
        f'{number}\n{_timestamp(start)} --> {_timestamp(end)}\n{text}\n'
        for number, (start, end, text) in enumerate(cues, start=1)
    ]
    return '\n'.join(blocks)