clash_data_serializer_sqlite#

This module provides SQLite-based serialization for clash detection data.

Classes#

ClashDataSerializerSqlite#

class omni.physxclashdetectioncore.clash_data_serializer_sqlite.ClashDataSerializerSqlite#

A class for serializing and deserializing clash data using SQLite database.

This class provides functionality to read, write, update, and delete clash data and clash queries in an SQLite database. It ensures compatibility with specific versions of clash data structures and manages database connections and transactions.

It supports deferred database creation until the first write operation to avoid creating empty database files. The class also includes methods for checking table compatibility, creating necessary database tables, and inserting, updating, and querying clash data and clash queries.

Class Constants:

CLASH_DB_FILE_EXT: str = ".clashdb"#: File extension for clash database files.

SUPPORTED_CLASH_QUERY_VERSION: int = 3#: Supported version of ClashQuery data structure.

SUPPORTED_CLASH_INFO_VERSION: int = 16#: Supported version of ClashInfo data structure.

SUPPORTED_CLASH_FRAME_INFO_VERSION: int = 6#: Supported version of ClashFrameInfo data structure.

Methods:

__init__()#: Initializes the ClashDataSerializerSqlite instance.

static check_serializer_compatibility_with_data_structures() → bool#

Checks if the serializer is compatible with current data structures.

Returns:: True if compatible with current data structures.
Return type:: bool

check_compatibility_of_tables() → bool#

Checks if the current database tables are compatible.

Returns:: True if tables are compatible.
Return type:: bool

check_possibility_of_tables_migration() → bool#

Checks if migration of all tables to the latest version is possible.

Returns:: True if migration of all tables to the latest version is possible.
Return type:: bool

migrate_table( table_name: str, target_version: int, commit: bool = True, ) → bool#

Run all migration steps for the given table from start_version to target_version in a single transaction.

Parameters:

table_name (str) – Name of the table to migrate.
target_version (int) – Target version to migrate to.
commit (bool) – Whether to commit the transaction. Defaults to True.

Returns:

True if the full sequence succeeds; on failure, rolls back and returns False.

Return type:

bool

open(file_path_name: str) → None#

Creates a file or opens an existing one.

Parameters:: file_path_name (str) – Path to the file to be opened.

get_file_path() → str#

Returns the serializer file path.

Returns:: The serializer file path.
Return type:: str

get_file_size() → int#

Returns the serializer file size in bytes.

Returns:: The file size in bytes.
Return type:: int

get_free_list_size() → int#

Returns the size of SQLite’s freelist in bytes.

The freelist contains pages that were previously used but are now free. These pages can be reused for new data.

Returns:: The total size of free pages in bytes.
Return type:: int

data_structures_compatible() → bool#

Returns True if the serializer has no compatibility issues (data structures, tables).

Returns:: True if no compatibility issues.
Return type:: bool

data_structures_migration_to_latest_version_possible() → bool#

Returns True if the serializer can migrate data structures to the latest version.

Returns:: True if migration to the latest version is possible.
Return type:: bool

migrate_data_structures_to_latest_version(file_path_name: str) → bool#

Migrates data structures to the latest version.

The single transaction design ensures atomic schema updates. Either all steps commit successfully or the entire sequence is rolled back, preserving the pre-migration state on error.

Parameters:: file_path_name (str) – Path to the clash data.
Returns:: True if migration was successful, False otherwise.
Return type:: bool

deferred_file_creation_until_first_write_op() → bool#

Returns True if the serializer will postpone file creation until the first write operation is requested.

Returns:: True if file creation is deferred until first write.
Return type:: bool

set_deferred_file_creation_until_first_write_op(value: bool) → None#

Sets if the serializer must postpone file creation until the first write operation is requested.

Parameters:: value (bool) – True to defer file creation, False otherwise.

is_open() → bool#

Returns if the serializer is ready.

Returns:: True if the serializer is ready.
Return type:: bool

save() → bool#

Saves data to the target file.

Returns:: True if save was successful, False otherwise.
Return type:: bool

close() → None#: Closes the opened file.

commit() → None#: Commits any unwritten data to the target file.

vacuum() → None#: Defragments the database and reclaims freed space.

insert_overlap( clash_info: ClashInfo, insert_also_frame_info: bool, update_identifier: bool, commit: bool, ) → int#

Inserts clash data. If already present, insertion is skipped.

Parameters:

clash_info (ClashInfo) – Clash information to insert.
insert_also_frame_info (bool) – Whether to insert frame info as well.
update_identifier (bool) – Update identifier if needed.
commit (bool) – Commit the transaction.

Returns:

ID of the new record.

Return type:

int

update_overlap( clash_info: ClashInfo, update_also_frame_info: bool, commit: bool, ) → int#

Updates clash data if present in the database.

Parameters:

clash_info (ClashInfo) – Clash information to update.
update_also_frame_info (bool) – Whether to update frame info as well.
commit (bool) – Commit the transaction.

Returns:

Number of affected records.

Return type:

int

find_all_overlaps_by_query_id( clash_query_id: int, fetch_also_frame_info: bool, num_frames_to_load: int = -1, first_frame_offset: int = 0, num_overlaps_to_load: int = -1, first_overlap_offset: int = 0, ) → Dict[str, ClashInfo]#

Finds all overlaps associated with a specific query ID.

This method retrieves all ClashInfo objects corresponding to the given clash_query_id from the database and optionally fetches additional frame information for each clash.

Parameters:

clash_query_id (int) – The ID of the query to search for overlaps.
fetch_also_frame_info (bool) – If True, fetches frame information associated with each ClashInfo object.
num_frames_to_load (int) – The maximum number of frames to load when fetching frame information. Defaults to -1 (all frames).
first_frame_offset (int) – The offset for the first frame to load when fetching frame information. Defaults to 0.
num_overlaps_to_load (int) – The maximum number of overlaps to load. Defaults to -1 (all overlaps).
first_overlap_offset (int) – The offset for the first overlap to load. Defaults to 0.

Returns:

A dictionary where the keys are overlap IDs (as strings) and the values are the corresponding ClashInfo objects. Empty dict if no results.

Return type:

Dict[str, ClashInfo]

get_overlaps_count_by_query_id(clash_query_id: int) → int#

Gets the total number of overlaps for a specific query ID.

Parameters:: clash_query_id (int) – The ID of the query to count overlaps for.
Returns:: The total number of overlaps for the query. Returns 0 if no results found.
Return type:: int

get_overlaps_count_by_query_id_grouped_by_state( clash_query_id: int, ) → Dict[ClashState, int]#

Gets the total number of overlaps for a specific query ID grouped by state.

Parameters:: clash_query_id (int) – The ID of the query to count overlaps for.
Returns:: A dictionary where keys are ClashState enum values and values are counts. Empty dict if no results.
Return type:: Dict[ClashState, int]

find_all_overlaps_by_overlap_id( overlap_id: Sequence[int], fetch_also_frame_info: bool, num_frames_to_load: int = -1, first_frame_offset: int = 0, ) → Dict[str, ClashInfo]#

Finds all overlaps by their overlap IDs.

This method retrieves all ClashInfo objects associated with the given overlap IDs and optionally fetches additional frame information for each clash.

Parameters:

overlap_id (Sequence[int]) – A sequence of overlap IDs to search for.
fetch_also_frame_info (bool) – If True, fetches frame information associated with each ClashInfo object.
num_frames_to_load (int) – The maximum number of frames to load. Defaults to -1 (all frames).
first_frame_offset (int) – The offset for the first frame to load. Defaults to 0.

Returns:

A dictionary where keys are overlap IDs and values are ClashInfo objects. Empty dict if no results.

Return type:

Dict[str, ClashInfo]

remove_all_overlaps_by_query_id( clash_query_id: int, commit: bool, ) → int#

Deletes specified clash data related to query_id.

Parameters:

clash_query_id (int) – The ID of the clash query.
commit (bool) – Whether to commit the transaction.

Returns:

Number of deleted rows.

Return type:

int

remove_overlap_by_id(overlap_id: int, commit: bool) → int#

Deletes specified clash data.

Parameters:

overlap_id (int) – The ID of the overlap.
commit (bool) – Whether to commit the transaction.

Returns:

Number of deleted rows.

Return type:

int

fetch_clash_frame_info_by_clash_info_id( clash_info_id: int, num_frames_to_load: int = -1, first_frame_offset: int = 0, ) → Sequence[ClashFrameInfo]#

Fetches frame information associated with a specific clash.

This method retrieves ClashFrameInfo records from the database for a given clash_info_id, ordered by timecode. It supports limiting the number of frames fetched and applying an offset to the starting frame.

Parameters:

clash_info_id (int) – The ID of the clash for which frame information is to be fetched.
num_frames_to_load (int) – The maximum number of frames to load. Defaults to -1 (all frames).
first_frame_offset (int) – The offset for the first frame to load. Defaults to 0.

Returns:

A list of ClashFrameInfo objects. Empty list if no results.

Return type:

Sequence[ClashFrameInfo]

get_clash_frame_info_count_by_clash_info_id(clash_info_id: int) → int#

Gets the total number of frame info records for a specific clash info ID.

Parameters:: clash_info_id (int) – The ID of the clash info to count frame info records for.
Returns:: The total number of frame info records. Returns 0 if no results found.
Return type:: int

insert_clash_frame_info_from_clash_info( clash_info: ClashInfo, commit: bool, ) → int#

Inserts clash_frame_info from ClashInfo.

Parameters:

clash_info (ClashInfo) – The ClashInfo object.
commit (bool) – Whether to commit the transaction.

Returns:

Number of affected records.

Return type:

int

insert_clash_frame_info( clash_frame_info: ClashFrameInfo, clash_info_id: int, commit: bool, ) → int#

Inserts clash_frame_info.

Parameters:

clash_frame_info (ClashFrameInfo) – The ClashFrameInfo object.
clash_info_id (int) – The ID of the clash info.
commit (bool) – Whether to commit the transaction.

Returns:

ID of the new record.

Return type:

int

remove_clash_frame_info_by_clash_info_id( clash_info_id: int, commit: bool, ) → int#

Deletes specified clash_frame_info data.

Parameters:

clash_info_id (int) – The ID of the clash info.
commit (bool) – Whether to commit the transaction.

Returns:

Number of deleted rows.

Return type:

int

fetch_all_queries() → Dict[int, ClashQuery]#

Returns all clash queries.

Returns:: Dictionary of all clash queries. Key is query identifier, value is ClashQuery object.
Return type:: Dict[int, ClashQuery]

insert_query( clash_query: ClashQuery, update_identifier: bool, commit: bool, ) → int#

Inserts clash query.

Parameters:

clash_query (ClashQuery) – The ClashQuery object.
update_identifier (bool) – Whether to update the identifier.
commit (bool) – Whether to commit the transaction.

Returns:

ID of the new record.

Return type:

int

update_query(clash_query: ClashQuery, commit: bool) → int#

Updates clash query if present in the DB.

Parameters:

clash_query (ClashQuery) – The ClashQuery object.
commit (bool) – Whether to commit the transaction.

Returns:

Number of affected records.

Return type:

int

remove_query_by_id(query_id: int, commit: bool) → int#

Deletes specified clash data.

Parameters:

query_id (int) – The ID of the query.
commit (bool) – Whether to commit the transaction.

Returns:

Number of deleted rows.

Return type:

int

find_query(clash_query_id: int) → ClashQuery | None#

Returns specified clash query.

Parameters:: clash_query_id (int) – The ID of the clash query.
Returns:: The ClashQuery object or None if not found.
Return type:: Optional[ClashQuery]

Properties:

db_file_path_name: str#: Gets the database file path name.

deferred_db_creation_until_commit_query: bool#: Gets the deferred database creation until commit query flag.

compatible_with_data_structures: bool#: Gets the compatibility status with data structures.

db_tables_compatible: bool#: Gets the compatibility status of the database tables.

MigrationStep#

class omni.physxclashdetectioncore.clash_data_serializer_sqlite.MigrationStep(sql: str, description: str, target_version: int)#

Represents a migration step to be applied to a database table when upgrading from one version to another.

This is a frozen dataclass that encapsulates information about a single migration step.

Parameters:

sql (str) – The SQL command to execute as part of the migration.
description (str) – A short description of what the migration does.
target_version (int) – The version of the table after applying this migration step.

Attributes:

sql: str#: The SQL command to execute.

description: str#: A description of the migration.

target_version: int#: The target version after migration.