savant_rs.utils

class savant_rs.utils.AtomicCounter(initial)

get

next

set(value)

class savant_rs.utils.BBoxMetricType

Allows configuring what kind of Intersection over Something to use.

IoU - Intersection over Union IoSelf - Intersection over Self (Intersection / Area of Self) IoOther - Intersection over Other (Intersection / Area of Other)

IoOther = BBoxMetricType.IoOther

IoSelf = BBoxMetricType.IoSelf

IoU = BBoxMetricType.IoU

class savant_rs.utils.ByteBuffer(v, checksum=None)

A fast copy-free byte buffer intended to efficiently exchange data between Rust & Python

Parameters:

v (Vec<u8>) – The byte buffer.
checksum (Optional[int]) – The checksum of the byte buffer.

Returns:

The byte buffer.

Return type:

ByteBuffer

bytes

Returns the bytes as Python bytes.

Returns:: The bytes as Python bytes.
Return type:: bytes

checksum

Returns the checksum of the byte buffer.

Returns:: The checksum of the byte buffer. If not set when creating the byte buffer, None is returned.
Return type:: None or int

is_empty()

Returns whether the byte buffer is empty.

Returns:: Whether the byte buffer is empty.
Return type:: bool

len()

Returns the length of the byte buffer.

Returns:: The length of the byte buffer.
Return type:: int

class savant_rs.utils.MaybeTelemetrySpan(span=None)

static current()

is_span

is_valid

nested_span(name)

nested_span_when(name, predicate)

trace_id

class savant_rs.utils.PropagatedContext

Represents a context that can be propagated to remote system like Python code

as_dict(): Returns the context in the form of a dictionary

nested_span(name)

Create a new child span

GIL Management: GIL is released during the call

Parameters:: name (str) – Name of the span
Returns:: A new span
Return type:: OTLPSpan

nested_span_when(name, predicate)

Creates a nested span only when log level is enabled

GIL Management: This method releases the GIL.

Parameters:: name (str) – The name of the span.
Returns:: a span that maybe a nested if the log level is enabled
Return type:: MaybeTelemetrySpan

class savant_rs.utils.TelemetrySpan(name)

A Span to be used locally. Works as a guard (use with with statement).

add_event(name, attributes=Ellipsis)

Adds an event to the span.

GIL Management: This method releases the GIL.

Parameters:

name (str) – The name of the event.
attributes (dict[str, str]) – The attributes of the event.

static constructor(name)

Create a root span with the given name for a new trace. Can be used as __init__ method.

Parameters:: name (str) – The name of the span.
Returns:: The created span.
Return type:: OTLPSpan

static context_depth()

static current()

static default()

enter()

is_valid

Checks if the span is a valid span (i.e. has a valid trace ID).

Returns:: True if the span is valid, False otherwise.
Return type:: bool

nested_span(name)

Create a child span with the given name.

GIL Management: This method releases the GIL.

Parameters:: name (str) – The name of the span.
Returns:: new span
Return type:: TelemetrySpan

nested_span_when(name, predicate)

Creates a nested span only when log level is enabled

GIL Management: This method releases the GIL.

Parameters:: name (str) – The name of the span.
Returns:: a span that maybe a nested if the log level is enabled
Return type:: MaybeTelemetrySpan

propagate()

Creates a propagation context from the current span.

GIL Management: This method releases the GIL.

Returns:: The created context.
Return type:: PropagatedContext

set_bool_attribute(key, value): Install the attribute with bool value.

set_bool_vec_attribute(key, value): Install the attribute with bool array value (list[bool]).

set_float_attribute(key, value): Install the attribute with float value.

set_float_vec_attribute(key, value): Install the attribute with float array value (list[float]).

set_int_attribute(key, value): Install the attribute with int value. Must be a valid int64, large numbers are not supported (use string instead).

set_int_vec_attribute(key, value): Install the attribute with int array value (list[int]). Must be a valid int64, large numbers are not supported (use string instead).

set_status_error(message): Configures the span status as Error with the given message.

set_status_ok(): Configures the span status as OK.

set_status_unset(): Configures the span status as Unset.

set_string_attribute(key, value): Install the attribute with str value.

set_string_vec_attribute(key, value): Install the attribute with string array value (list[str]).

span_id()

Returns the span ID of the span.

Returns:: The span ID.
Return type:: str

trace_id()

Returns the trace ID of the span.

Returns:: The trace ID.
Return type:: str

class savant_rs.utils.VideoObjectBBoxTransformation

static scale(x, y)

static shift(x, y)

class savant_rs.utils.VideoObjectBBoxType

Determines which object bbox is a subject of the operation

Detection = VideoObjectBBoxType.Detection

TrackingInfo = VideoObjectBBoxType.TrackingInfo

savant_rs.utils.enable_dl_detection(): Enables deadlock detection

savant_rs.utils.estimate_gil_contention(): When loglevel is set to Trace reports the number of nanoseconds spent waiting for the GIL The report is sent to the current telemetry span

savant_rs.utils.eval_expr(query, ttl=100, no_gil=True)

savant_rs.utils.forget_video_id(source_id): Drop process-global state for source_id. A subsequent mint_video_id() for the same source restarts the keyframe-anchor state machine.

savant_rs.utils.gen_empty_frame()

savant_rs.utils.gen_frame()

savant_rs.utils.incremental_uuid_v7(): Returns a new UUID v7

savant_rs.utils.mint_video_id(source_id, pts, is_keyframe, wall_clock_ns=None)

Mint a new VideoId from the process-global generator and return it as a UUID-formatted string.

Parameters:

source_id (str) – Stream identifier the frame belongs to. Used purely as the per-source accounting key inside the global generator; not encoded in the returned id.
pts (int) – Frame PTS in stream timebase.
is_keyframe (bool) – True updates the per-source keyframe anchor and may bump the epoch field if a PTS reset is detected.
wall_clock_ns (int, optional) – Demuxer wall clock in nanoseconds. Defaults to the current system time. Pass an explicit value when replaying or when coordinating ids across processes.

Returns:

The minted id, encoded in UUID hyphenated form.

Return type:

str

savant_rs.utils.relative_time_uuid_v7(uuid, offset_millis)

Returns a new UUID v7 that is offset from the given UUID by the given number of milliseconds

The offset can be positive or negative

Parameters:

uuid (str) – The UUID to offset
offset_millis (int) – The number of milliseconds to offset the UUID by

Returns:

The new UUID

Return type:

str

savant_rs.utils.relative_time_video_id(id, offset_millis)

Return a copy of id with its ts_ns component shifted by offset_millis. Migration target for relative_time_uuid_v7(). Deterministic — identical inputs always produce identical output.

Parameters:

id (str) – VideoId in UUID hyphenated form.
offset_millis (int) – Signed shift applied to ts_ns (in milliseconds for ergonomic parity with the legacy relative_time_uuid_v7).

savant_rs.utils.round_2_digits(v)

savant_rs.utils.video_id_lower_bound(ts_ns): Inclusive lower bound for a wall-clock-time range scan at ts_ns. Pair with video_id_upper_bound(). Source identity is supplied by the caller’s surrounding lookup, not by the boundary.

savant_rs.utils.video_id_upper_bound(ts_ns): Inclusive upper bound for a wall-clock-time range scan at ts_ns.