qbiocode.apps.quvine.reproducibility.dataset_registry module

Dataset Registry for QuVINE

Central registry mapping dataset names to their locations, metadata, and task splits.

Summary

Classes:

DatasetEntry	Registry entry for a single dataset instance.
DatasetMetadata	Metadata for a dataset.
DatasetRegistry	Central registry for all datasets used in QuVINE experiments.
DatasetType	Type of dataset.

Reference

class DatasetType(*values)

Bases: Enum

Type of dataset.

SYNTHETIC = 'synthetic'

REAL_WORLD = 'real_world'

PPI = 'ppi'

class DatasetMetadata(name, dataset_type, num_nodes, num_edges, is_directed, has_node_features, has_node_labels, generation_params=None, description=None, ppi_source=None, disease=None, requested_size=None, actual_size=None, largest_connected_component_size=None)

Bases: object

Metadata for a dataset.

name: str

dataset_type: DatasetType

num_nodes: int

num_edges: int

is_directed: bool

has_node_features: bool

has_node_labels: bool

generation_params: Optional[Dict[str, Any]] = None

description: Optional[str] = None

ppi_source: Optional[str] = None

disease: Optional[str] = None

requested_size: Optional[int] = None

actual_size: Optional[int] = None

largest_connected_component_size: Optional[int] = None

class DatasetEntry(dataset_name, dataset_type, repetition_id, seed, graph_path, metadata_path, available_tasks, split_paths, disease_node_path=None, ppi_source=None, disease=None, requested_size=None, actual_size=None, registry_group=None)

Bases: object

Registry entry for a single dataset instance.

Each entry represents one graph instance for a fixed dataset/source/disease/size/repetition combination.

dataset_name: str

dataset_type: DatasetType

repetition_id: int

seed: int

graph_path: Path

metadata_path: Path

available_tasks: List[str]

split_paths: Dict[str, Path]

disease_node_path: Optional[Path] = None

ppi_source: Optional[str] = None

disease: Optional[str] = None

requested_size: Optional[int] = None

actual_size: Optional[int] = None

registry_group: Optional[str] = None

to_dict()

Convert to dictionary for JSON serialization.

Return type:

Dict

classmethod from_dict(data)

Create from dictionary.

Return type:

DatasetEntry

class DatasetRegistry(registry_path=None)

Bases: object

Central registry for all datasets used in QuVINE experiments.

Maintains a mapping from (dataset_name, repetition_id) to dataset locations and metadata. Ensures all methods access the same data files.

__init__(registry_path=None)

Initialize dataset registry.

Parameters:

registry_path (Path, optional) – Path to registry JSON file. If provided, loads existing registry.

register(dataset_name, dataset_type, repetition_id, seed, graph_path, metadata_path, available_tasks, split_paths=None, disease_node_path=None, ppi_source=None, disease=None, requested_size=None, actual_size=None, registry_group=None)

Register a dataset instance.

Parameters:

• dataset_name (str) – Name of the dataset

• dataset_type (DatasetType) – Type of dataset

• repetition_id (int) – Repetition index

• seed (int) – Seed used for generation/sampling

• graph_path (Path) – Path to saved graph file

• metadata_path (Path) – Path to metadata JSON

• available_tasks (List[str]) – Tasks available for this dataset

• split_paths (Dict[str, Path], optional) – Mapping from task name to split file path

Return type:

None

get(dataset_name, repetition_id)

Get dataset entry.

Parameters:

• dataset_name (str) – Dataset name

• repetition_id (int) – Repetition ID

Returns:

Dataset entry if found, None otherwise

Return type:

DatasetEntry or None

add_split(dataset_name, repetition_id, task, split_path)

Add a task split to an existing dataset entry.

Parameters:

• dataset_name (str) – Dataset name

• repetition_id (int) – Repetition ID

• task (str) – Task name

• split_path (Path) – Path to split file

Return type:

None

get_all_datasets()

Get list of all unique dataset names.

Return type:

List[str]

get_repetitions(dataset_name)

Get all repetition IDs for a dataset.

Return type:

List[int]

get_entries_for_dataset(dataset_name)

Get all entries for a dataset across all repetitions.

Return type:

List[DatasetEntry]

validate_consistency(dataset_name, repetition_id, task)

Validate that all required files exist for a dataset-task combination.

Parameters:

• dataset_name (str) – Dataset name

• repetition_id (int) – Repetition ID

• task (str) – Task name

Returns:

True if all files exist, False otherwise

Return type:

bool

save(output_path=None)

Save registry to JSON file.

Parameters:

output_path (Path, optional) – Output path. If None, uses self.registry_path.

Return type:

None

load(input_path)

Load registry from JSON file.

Parameters:

input_path (Path) – Input path

Return type:

None

get_summary()

Get summary statistics of the registry.

Returns:

Summary statistics

Return type:

dict

export_registry_json(output_path)

Explicit helper for writing a registry JSON artifact.

Return type:

None

export_registry_csv(output_path)

Export flattened registry records to CSV.

Return type:

None