pyvrs.base

class pyvrs.base.BaseVRSReader[source]

Bases: ABC

A Pythonic reader for VRS files. Behaves as a filterable list - has a length (number of records), can be indexed to retrieve VRSRecords, and can be iterated over and sliced just like a regular Python list. Significant file reads are only done when record state is queried, so it remains performant even for larger VRS files.

This has several child classes and following are the relationship between them.

                                |-- SyncVRSReader
                |-- VRSReader --
                |               |-- AsyncVRSReader
BaseVRSReader --|
                |                       |-- SyncFilteredVRSReader
                |-- FilteredVRSReader --
                                        |-- AsyncFilteredVRSReader

Note

  • BaseVRSReader: Base abstract class that defines the common functions across all child classes.

  • VRSReader: Abstract class that represents entire file (e.g. file without any filters). The methods in child classes of this operats against all records. When user call filtered_by_fields method, that call will create FilteredVRSReader that represents slice of the file.

  • FilteredVRSReader: Abstract class that represents the slice of the original file (After applying filter). This class essentially has the exact same methods as VRSReader but operate against subset of the file. Note that you can’t ‘re-filter’ an already filtered VRSReader.

  • SyncVRSReader: Synchronous version of VRSReader.

  • AsyncVRSReader: Asynchronous version of VRSReader, only difference between SyncVRSReader is AsyncVRSReader supports __aiter__, __anext__ for async iteration as well as __getitem__ operates asynchronously.

  • SyncFilteredVRSReader: Synchronous version of FilteredVRSReader.

  • AsyncFilteredVRSReader: Asynchronous version of FilteredVRSReader, same difference as SyncVRSReader vs AsyncVRSReader.

abstract property file_tags: Mapping[str, str]

Return a dict of all file tags present in this VRS file.

Returns:

{<tag>: <value>}

Return type:

Dictionary of all file tags

abstract find_stream(recordable_type_id: int, tag_name: str, tag_value: str) → str[source]

Find stream matching recordable type and tag, and return its stream id.

Parameters:

  • recordable_type_id – stream_id is <recordable_type_id>-<instance_id>

  • tag_name – tag name that you are interested in

  • tag_value – tag value that you are interested in

Returns:

Stream ID that starts with recordable_type_id and has a given tag pair.

abstract find_streams(recordable_type_id: int, flavor: str = '') → List[str][source]

Find streams matching recordable type and flavor, and return sets of stream ids.

Parameters:

  • recordable_type_id – stream_id is <recordable_type_id>-<instance_id>

  • tag_name – tag name that you are interested in

  • tag_value – tag value that you are interested in

Returns:

A set of stream IDs that start with recordable_type_id and has a given flavor.

abstract get_estimated_frame_rate(stream_id: str) → float[source]

Get the estimated frame rate for the given stream_id.

Parameters:

stream_id – stream_id that you are interested in.

Returns:

The estimated frame rate.

abstract get_record_index_by_time(stream_id: str, timestamp: float, epsilon: float | None = None, record_type: RecordType | None = None) → int[source]

Get index in filtered records by timestamp.

Parameters:

  • stream_id – stream_id that you are interested in.

  • timestamp – timestamp that you are interested in.

  • epsilon – Optional argument. If specified we search for record in range of (timestamp-epsilon)-(timestamp+epsilon) and returns the nearest record.

  • record_type – Optional argument. If specified we search for record with the record_type.

Returns:

The absolute index of the record corresponds to the stream_id & timestamp.

Raises:

  • TimestampNotFoundError – If epsilon is not None and the record doesn’t exist within the time range.

  • ValueError – If epsilon is None and the record isn’t found using lower_bound.

abstract get_records_count(stream_id: str, record_type: RecordType) → int[source]

Get the number of records for the stream_id & record_type.

Parameters:

  • stream_id – stream_id you are interested in.

  • record_type – record type you are interested in.

Returns:

The number of records for stream_id & record type

abstract get_stream_info(stream_id: str) → Dict[str, str][source]

Get details about a stream.

Parameters:

stream_id – stream_id you are interested in.

Returns:

An information about the stream in a dictionary.

abstract get_timestamp_for_index(index: int) → float[source]

Get the timestamp corresponding to the given index.

Parameters:

index – the index for the record

Returns:

A timestamp corresponds to the index

abstract get_timestamp_list(indices: List[int] | None = None) → List[float][source]

Get the list of timestamps corresponding to the given indices.

Parameters:

indices – the list of indices we want to get the timestamp.

Returns:

A list of timestamps correspond to the indices, if indices are None, we get the full timestamp list.

abstract property max_timestamp: float

Return a maximum timestamp of this VRS file.

abstract might_contain_audio(stream_id: str) → bool[source]

Check if the given stream_id contains an audio data.

Parameters:

stream_id – stream_id that you are interested in.

Returns:

Based on the config record, return if the stream contains an audio data.

abstract might_contain_images(stream_id: str) → bool[source]

Check if the given stream_id contains an image data.

Parameters:

stream_id – stream_id that you are interested in.

Returns:

Based on the config record, return if the stream contains an image data.

abstract property min_timestamp: float

Return a minimum timestamp of this VRS file.

abstract property n_records: int

Return a number of records in this VRS file.

abstract read_next_record(stream_id: str, record_type: str, index: int) → VRSRecord | None[source]

Read the first record that matches stream_id and record_type and its index is greater or equal than given index.

Parameters:

  • stream_id – stream_id that you are interested in.

  • record_type – record_type that you are interested in.

  • index – the absolute index in the file. Based on this index, try to find the previous record that matches stream_id & record_type

Returns:

VRSRecord if there is a record, otherwise None

abstract read_prev_record(stream_id: str, record_type: str, index: int) → VRSRecord | None[source]

Read the last record that matches stream_id and record_type and its index is smaller or equal than given index.

Parameters:

  • stream_id – stream_id that you are interested in.

  • record_type – record_type that you are interested in.

  • index – the absolute index in the file. Based on this index, try to find the previous record that matches stream_id & record_type

Returns:

VRSRecord if there is a record, otherwise None

abstract read_record_by_time(stream_id: str, timestamp: float, epsilon: float | None = None, record_type: RecordType | None = None) → VRSRecord[source]

Read record by timestamp.

Parameters:

  • stream_id – stream_id that you are interested in.

  • timestamp – timestamp that you are interested in.

  • epsilon – Optional argument. If specified we search for record in range of (timestamp-epsilon)-(timestamp+epsilon) and returns the nearest record.

  • record_type – Optional argument. If specified we search for record with the record_type.

Returns:

VRSRecord corresponds to the stream_id & timestamp.

Raises:

  • TimestampNotFoundError – If epsilon is not None and the record doesn’t exist within the time range.

  • ValueError – If epsilon is None and the record isn’t found using lower_bound.

abstract property record_types: Set[str]

Return a set of record types in this VRS file.

abstract set_image_conversion(conversion: ImageConversion) → None[source]

Set default image conversion policy, and clears any stream specific setting.

Parameters:

conversion – The image conversion you want to apply for all streams.

abstract set_stream_image_conversion(stream_id: str, conversion: ImageConversion) → None[source]

Set image conversion policy for a specific stream.

Parameters:

  • stream_id – The stream_id you want to apply image conversion to.

  • conversion – The image conversion you want to apply for a specific stream.

abstract set_stream_type_image_conversion(recordable_type_id: str, conversion: ImageConversion) → int[source]

Set image conversion policy for streams of a specific device type.

Parameters:

  • recordable_type_id – The recordable_type_id you want to apply image conversion to. If you specify 1000, streams with id 1000-* are the targets.

  • conversion – The image conversion you want to apply for a specific stream.

Returns:

The number of streams affected.

abstract property stream_ids: Set[str]

Return a set of stream ids in this VRS file.

abstract property stream_tags: Mapping[str, Mapping[str, Any]]

Return a dict of all per-stream tags present in this VRS file.

Returns:

{<stream_id>: {<tag>: <value>}}

Return type:

Dictionary of all per-stream tags

property time_range: float

Return a timestamp range of this VRS file.