ModelDefinitions

Represents a batch of ML model metadata records.

Parameters

Field	Type	Description
model_names	list[str]	Required. Model names, length N.
model_formats	list[str]	Required. Model serialization formats, length N.
visibilities	list[str]	Required. Model visibility levels, length N.
model_statuses	list[str]	Required. Model statuses, length N.
model_descriptions	list[str | None] | None	Optional. Per-model descriptions, length N.
model_inputs	list[list[ModelTensorDefinition | dict] | None] | None	Optional. Per-model input tensor signatures, length N. Dict entries are converted via ModelTensorDefinition.from_dict.
model_outputs	list[list[ModelTensorDefinition | dict] | None] | None	Optional. Per-model output tensor signatures, same shape as model_inputs.
created_ats	list[datetime | None] | None	Optional. Per-model creation timestamps, length N.
updated_ats	list[datetime | None] | None	Optional. Per-model last-update timestamps, length N.

Raises

Exception	Condition
TypeError	A required field isn't a list of the expected element type, or an optional field isn't None/a list of the expected type.
ValueError	Any provided field's length doesn't match len(model_names).

Attributes

Attribute	Type	Description
model_names	np.ndarray	Shape (N,), object dtype.
model_formats	np.ndarray	Shape (N,), object dtype.
visibilities	np.ndarray	Shape (N,), object dtype.
model_statuses	np.ndarray	Shape (N,), object dtype.
model_descriptions	np.ndarray | None	Shape (N,), or None if every record omitted it.
model_inputs	np.ndarray | None	Length-N ragged object array; each element is a list[ModelTensorDefinition] or None, or the whole attribute is None if every record omitted it.
model_outputs	np.ndarray | None	Same shape/semantics as model_inputs.
created_ats	list[datetime] | None	Plain list (not an np.ndarray), length N, or None if every record omitted it.
updated_ats	list[datetime] | None	Plain list, length N, or None if every record omitted it.

All properties allocate a fresh array/list on each access — none are defensive views into mutable internal state, so there's no separate setter to validate against.

Methods

Method	Description
ModelDefinitions.coerce(value, name="value")	Classmethod. Returns value unchanged if it's already a ModelDefinitions; expands a dict of constructor keyword arguments via cls(**value). Raises TypeError otherwise.

Operators

Operation	Behavior
len(defs)	Number of records, N.
defs[i] (int)	A length-1 ModelDefinitions sub-batch. Negative indices are supported; raises IndexError if out of range.
defs[i:j] (slice)	A ModelDefinitions sub-batch.
defs[mask] (boolean np.ndarray)	A ModelDefinitions sub-batch of the matched rows. Raises ValueError if mask isn't length N.
a == b	True if other is a ModelDefinitions with equal fields, in the same order.

Not hashable (__hash__ is None) — despite being immutable after construction (no data-style setter), ModelDefinitions is deliberately excluded from hashing.

Serialization

to_pyarrow() uses list-of-structs (one shared offsets buffer for the whole batch) rather than struct-of-lists, which is measurably faster to (de)serialize and marginally smaller:

ListArray length 1
└── values: StructArray length N
        ├── model_name:        string
        ├── model_format:      string
        ├── visibility:        string
        ├── model_status:      string
        ├── model_description: string (null if omitted)
        ├── model_input:       list<struct<name, canonical_name, dtype, shape:list<int32>>> (null if omitted)
        ├── model_output:      same type as model_input (null if omitted)
        ├── created_at:        timestamp[ns, tz=UTC] (null if omitted)
        └── updated_at:        timestamp[ns, tz=UTC] (null if omitted)

One ModelDefinitions object serializes as one Arrow row, with all N records packed into that row's single list element.

Visualization

datatypes.visualize(definitions, entity_path=...) logs the batch as a rerun markdown table, one row per record. model_description is omitted from the table to keep rows compact; model_input/model_output are rendered as name:dtype[shape] entries joined by "; ".

Example

"""Demonstrates the Telekinesis ModelDefinitions datatype."""

import time
from datetime import datetime, timezone

import rerun as rr
from loguru import logger

from telekinesis import datatypes

def model_definitions_example():
    """Demonstrate batch construction (canonical_name input_0/output_N), access, indexing, empty batches, and serialization."""

    # ======================= Create ============================================
    created_at = datetime(2024, 6, 1, tzinfo=timezone.utc)
    updated_at = datetime(2024, 6, 15, tzinfo=timezone.utc)

    model_input = datatypes.ModelTensorDefinition(
        name="images", canonical_name="input_0", dtype="float32", shape=[1, 3, 224, 224]
    )
    model_output = datatypes.ModelTensorDefinition(
        name="logits", canonical_name="output_0", dtype="float32", shape=[1, 1000]
    )
    other_output = datatypes.ModelTensorDefinition(
        name="logits", canonical_name="output_0", dtype="float32", shape=[1, 1000]
    )

    definitions = datatypes.ModelDefinitions(
        model_names=["model-a", "model-b"],
        model_formats=["onnx", "pytorch"],
        visibilities=["private", "public"],
        model_statuses=["uploaded", "deploying"],
        model_descriptions=["first model", None],
        model_inputs=[[model_input], None],
        model_outputs=[[model_output, other_output], None],
        created_ats=[created_at, updated_at],
        updated_ats=[created_at, updated_at],
    )

    # ======================= Visualize =========================================
    rr.init("model_definitions_example", spawn=True)
    datatypes.visualize(definitions, entity_path="/ModelDefinitions")

    # ======================= Inspect ===========================================
    logger.info(f"Number of records: {len(definitions)}")
    logger.info(f"model_names={definitions.model_names}, model_formats={definitions.model_formats}")
    logger.info(f"visibilities={definitions.visibilities}, model_statuses={definitions.model_statuses}")
    logger.info(f"model_inputs={definitions.model_inputs}")
    logger.info(f"created_ats={definitions.created_ats}, updated_ats={definitions.updated_ats}")

    # ======================= Index =============================================
    first = definitions[0]
    subset = definitions[0:1]
    mask = definitions.model_statuses == "uploaded"
    uploaded_only = definitions[mask]

    logger.info(f"definitions[0] = {first}")
    logger.info(f"definitions[0:1] = {len(subset)} record(s), names={subset.model_names}")
    logger.info(f"definitions[uploaded mask] = {len(uploaded_only)} record(s)")

    # ======================= Empty Batch =======================================
    empty = datatypes.ModelDefinitions(
        model_names=[], model_formats=[], visibilities=[], model_statuses=[]
    )
    datatypes.visualize(empty, entity_path="/ModelDefinitions/empty")

    # ======================= Serialize / Deserialize ===========================
    start = time.perf_counter()
    serialized = datatypes.serialize(definitions)
    serialization_ms = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    deserialized = datatypes.deserialize(serialized)["param_0"]
    deserialization_ms = (time.perf_counter() - start) * 1000

    logger.info(f"Deserialized ModelDefinitions: {deserialized}")
    logger.info(f"Round-trip successful: {deserialized == definitions}")
    logger.info(f"Serialization time: {serialization_ms:.3f} ms")
    logger.info(f"Deserialization time: {deserialization_ms:.3f} ms")


if __name__ == "__main__":
    model_definitions_example()