| """Prepare explicit, hashed data for a new Sudoku training workflow.""" |
|
|
| from pathlib import Path |
|
|
| from omegaconf import DictConfig, OmegaConf |
| from softasp_core.config.schemas import DatasetConfig |
| from softasp_domain.sudoku.downloader import SudokuDownloader |
| from softasp_domain.sudoku.preprocessor import SudokuPreprocessor |
|
|
| from softasp.sudoku_data import select_hdf5 |
| from softasp.sudoku_reporting import file_sha256 |
|
|
|
|
| def prepare_scratch_data(cfg: DictConfig) -> Path: |
| """Generate in a fresh directory, or validate an explicitly pinned existing file.""" |
| if cfg.pipeline.data_mode == "existing": |
| return select_hdf5(cfg) |
| |
| import clingo |
| from torchvision.datasets import MNIST |
|
|
| raw_dir = ( |
| ( |
| Path(cfg.paths.raw_data_root) |
| if cfg.paths.get("raw_data_root") |
| else Path(cfg.paths.data_root) / "raw" / "sudoku" |
| ) |
| .expanduser() |
| .resolve() |
| ) |
| if cfg.pipeline.download: |
| SudokuDownloader(raw_dir).run() |
| |
| for train in (True, False): |
| MNIST(root=str(raw_dir / "mnist"), train=train, download=False) |
| destination = ( |
| Path(cfg.paths.data_root).expanduser().resolve() |
| / "processed" |
| / "sudoku" |
| / "from_scratch" |
| / cfg.pipeline.name |
| ) |
| destination.mkdir(parents=True, exist_ok=False) |
| dataset = DatasetConfig(**cfg.dataset) |
| path = SudokuPreprocessor(raw_dir, destination, cfg.preprocessing, dataset).run() |
| OmegaConf.update( |
| cfg, |
| "data_artifact", |
| { |
| "path": str(path.resolve()), |
| "sha256": file_sha256(path), |
| |
| "protocol_id": None, |
| }, |
| ) |
| return select_hdf5(cfg) |
|
|
