@module: sce.io
@depends: pandas, tomllib
@exports: DatasetInfo, ensure_dataset, get_dataset_info, list_datasets, load_dataset, save_dataset, verify_all_datasets, verify_dataset
@paper_ref: N/A
@data_flow: dataset name/config -> parquet path or download -> pandas.DataFrame
dataclass
Metadata for a dataset known to the repository.
Source code in sce/io/__init__.py
| @dataclass(frozen=True)
class DatasetInfo:
"""Metadata for a dataset known to the repository."""
name: str
path: Path
source: str
description: str
remote_source: str | None
checksum: str | None
size_bytes: int | None
exists_locally: bool
|
list_datasets() -> list[DatasetInfo]
List datasets defined under the repo config directory.
TOML files without a [dataset].path entry (e.g. shared report
defaults) are not dataset configs and are skipped.
Source code in sce/io/__init__.py
| def list_datasets() -> list[DatasetInfo]:
"""List datasets defined under the repo config directory.
TOML files without a ``[dataset].path`` entry (e.g. shared report
defaults) are not dataset configs and are skipped.
"""
infos = []
for path in CONFIG_DIR.glob("*.toml"):
payload = _load_toml(path)
if "path" not in payload.get("dataset", {}):
continue
infos.append(_dataset_info_from_config(path))
return sorted(infos, key=lambda item: item.name)
|
get_dataset_info(dataset: str | Path) -> DatasetInfo
Return dataset metadata for a dataset name or config path.
Source code in sce/io/__init__.py
| def get_dataset_info(dataset: str | Path) -> DatasetInfo:
"""Return dataset metadata for a dataset name or config path."""
_, config_path, _ = _resolve_dataset_reference(dataset)
return _dataset_info_from_config(config_path)
|
ensure_dataset(dataset: str | Path, force_download: bool = False) -> Path
Ensure a dataset parquet exists locally, downloading it if configured as remote.
Source code in sce/io/__init__.py
| def ensure_dataset(dataset: str | Path, force_download: bool = False) -> Path:
"""Ensure a dataset parquet exists locally, downloading it if configured as remote."""
info = get_dataset_info(dataset)
if info.exists_locally and not force_download:
return info.path
if info.remote_source is None:
raise FileNotFoundError(f"Dataset not found locally: {info.path}")
if not DOWNLOAD_SCRIPT.exists():
raise FileNotFoundError(f"Dataset downloader not found: {DOWNLOAD_SCRIPT}")
command = [sys.executable, str(DOWNLOAD_SCRIPT), "--dataset", info.path.name]
if force_download:
command.append("--force")
result = subprocess.run(command, capture_output=True, text=True, check=False)
if result.returncode != 0:
error_text = result.stderr.strip() or result.stdout.strip() or "dataset download failed"
raise RuntimeError(error_text)
return info.path
|
load_dataset(dataset: str | Path, force_download: bool = False, **read_kwargs: Any) -> pd.DataFrame
Load a configured dataset by name or config path.
Source code in sce/io/__init__.py
| def load_dataset(
dataset: str | Path, force_download: bool = False, **read_kwargs: Any
) -> pd.DataFrame:
"""Load a configured dataset by name or config path."""
dataset_path = ensure_dataset(dataset, force_download=force_download)
return pd.read_parquet(dataset_path, **read_kwargs)
|
save_dataset(df: DataFrame, path: str | Path, **write_kwargs: Any) -> Path
Save a DataFrame to parquet, creating parent directories if needed.
Source code in sce/io/__init__.py
| def save_dataset(df: pd.DataFrame, path: str | Path, **write_kwargs: Any) -> Path:
"""Save a DataFrame to parquet, creating parent directories if needed."""
output_path = Path(path)
if not output_path.is_absolute():
output_path = PROJECT_ROOT / output_path
output_path.parent.mkdir(parents=True, exist_ok=True)
df.to_parquet(output_path, index=False, **write_kwargs)
return output_path
|
verify_dataset(dataset: str | Path) -> bool
Verify local size and checksum against the manifest when available.
Source code in sce/io/__init__.py
| def verify_dataset(dataset: str | Path) -> bool:
"""Verify local size and checksum against the manifest when available."""
info = get_dataset_info(dataset)
if not info.path.exists() or info.checksum is None or info.size_bytes is None:
return info.path.exists()
return info.path.stat().st_size == info.size_bytes and _sha256(info.path) == info.checksum
|
verify_all_datasets() -> dict[str, bool]
Verify all configured datasets and return a name->status mapping.
Source code in sce/io/__init__.py
| def verify_all_datasets() -> dict[str, bool]:
"""Verify all configured datasets and return a name->status mapping."""
return {info.name: verify_dataset(info.name) for info in list_datasets()}
|