ModelDefinitions
SUMMARY
A collection of machine-learning model definitions.
python
from telekinesis import datatypes
definitions = datatypes.ModelDefinitions(
model_names=["detector"],
model_formats=["onnx"],
visibilities=["public"],
model_statuses=["active"],
)Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
model_names | list[str] | Required | Model names with length N. |
model_formats | list[str] | Required | Model serialization formats with length N, with one format per model. |
visibilities | list[str] | Required | Model visibility values with length N, with one value per model. |
model_statuses | list[str] | Required | Model status values with length N, with one value per model. |
model_descriptions | list[str | None] | None | None | Optional model descriptions with length N. |
model_inputs | list[list[ModelTensorDefinition | dict] | None] | None | None | Optional length-N collection of model input tensor definitions. Tensor-definition dictionaries are converted to ModelTensorDefinition instances. |
model_outputs | list[list[ModelTensorDefinition | dict] | None] | None | None | Optional length-N collection of model output tensor definitions. Tensor-definition dictionaries are converted to ModelTensorDefinition instances. |
created_ats | list[datetime | None] | None | None | Optional model creation datetimes with length N. |
updated_ats | list[datetime | None] | None | None | Optional model update datetimes with 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). |
Required fields and the optional model_descriptions/created_ats/updated_ats lists are stored as the original list objects passed in, without copying. model_inputs/model_outputs are rebuilt into new per-model lists, converting dict entries to ModelTensorDefinition (via ModelTensorDefinition.from_dict) while reusing already-constructed ModelTensorDefinition instances as-is.
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. |
model_names, model_formats, visibilities, model_statuses, and model_descriptions allocate a fresh NumPy object array on every access, so mutating the returned array doesn't affect the instance. model_inputs/model_outputs also allocate a fresh outer ragged array on every access, but each element aliases the stored per-model list, so mutating one of those inner lists in place does affect the instance. created_ats/updated_ats return the stored list directly with no copying at all, so mutating the returned list mutates the instance too. None of these have a separate setter — fields are set only at construction.
Methods
| Method | Type | Description |
|---|---|---|
ModelDefinitions.coerce(value, name="value") | ModelDefinitions | Converts a dict of constructor keyword arguments into a ModelDefinitions, by expanding it as cls(**value). If value is already a ModelDefinitions, it is returned unchanged. The optional name is only used to make validation error messages more descriptive. |
ModelDefinitions.from_pyarrow(col) | ModelDefinitions | Rebuilds a ModelDefinitions from the length-1 ListArray produced by to_pyarrow. col must be a pa.ListArray of length 1 whose struct fields match the expected schema. |
to_pyarrow() | pa.ListArray | Serializes the batch into a length-1 ListArray wrapping a StructArray, with one row per model definition. |
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. |
Visualization
python
import rerun as rr
# Your code block
# ....
rr.init("model_definitions_example", spawn=True)
datatypes.visualize(definitions, entity_path="/definitions", label="ModelDefinitions")Example
python
"""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 creation, inspection, operations, visualization, 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],
)
logger.info(f"Created ModelDefinitions: {definitions}")
# ======================= Inspect ===========================================
logger.info(f"model_names={definitions.model_names}")
logger.info(f"model_formats={definitions.model_formats}")
logger.info(f"visibilities={definitions.visibilities}")
logger.info(f"model_statuses={definitions.model_statuses}")
logger.info(f"model_descriptions={definitions.model_descriptions}")
logger.info(f"model_inputs={definitions.model_inputs}")
logger.info(f"model_outputs={definitions.model_outputs}")
logger.info(f"created_ats={definitions.created_ats}")
logger.info(f"updated_ats={definitions.updated_ats}")
# ======================= Operations =========================================
logger.info(f"length={len(definitions)}")
subset = definitions[0:1]
logger.info(f"definitions[0:1] = {len(subset)} record(s), names={subset.model_names}")
first = definitions[0]
logger.info(f"definitions[0] = {first}")
mask = definitions.model_statuses == "uploaded"
logger.info(f"definitions[uploaded mask] = {len(definitions[mask])} record(s)")
# ModelTensorDefinition exposes its fields as plain attributes plus a dict view.
logger.info(f"model_input.to_dict()={model_input.to_dict()}")
# A ModelDefinitions batch may be empty.
empty = datatypes.ModelDefinitions(
model_names=[], model_formats=[], visibilities=[], model_statuses=[]
)
logger.info(f"length of empty batch={len(empty)}")
# ======================= Visualize =========================================
rr.init("model_definitions_example", spawn=True)
datatypes.visualize(definitions, entity_path="/model_definitions")
datatypes.visualize(empty, entity_path="/model_definitions/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: {definitions == deserialized}")
logger.info(f"Serialization time: {serialization_ms:.3f} ms")
logger.info(f"Deserialization time: {deserialization_ms:.3f} ms")
if __name__ == "__main__":
model_definitions_example()