Main pipeline runner that orchestrates data loading, processing, and saving. Initializes all components from the global config: - Logger - Model (optional, for MLLM operators) - DataLoaders (with checkpoint support) - DataSaver - Pipeline (with statistics collection) Su
| 27 | |
| 28 | |
| 29 | class Runner: |
| 30 | """Main pipeline runner that orchestrates data loading, processing, and saving. |
| 31 | |
| 32 | Initializes all components from the global config: |
| 33 | - Logger |
| 34 | - Model (optional, for MLLM operators) |
| 35 | - DataLoaders (with checkpoint support) |
| 36 | - DataSaver |
| 37 | - Pipeline (with statistics collection) |
| 38 | |
| 39 | Supports two execution modes: |
| 40 | - Normal pipeline mode: load -> process -> save |
| 41 | - Cache-images-only mode: iterate data to populate LMDB image cache |
| 42 | """ |
| 43 | |
| 44 | def __init__(self, cache_images_only: bool = False): |
| 45 | cfg = get_cfg() |
| 46 | |
| 47 | print_cfg() |
| 48 | |
| 49 | self.cfg = cfg |
| 50 | self.work_dir = cfg.work_dir |
| 51 | self.cache_images_only = cache_images_only |
| 52 | |
| 53 | self.logger = build_from_cfg( |
| 54 | self.cfg.logger, |
| 55 | HOOKS, |
| 56 | work_dir=self.work_dir, |
| 57 | wandb_project="datastudio", |
| 58 | wandb_config=self.cfg, |
| 59 | ) |
| 60 | |
| 61 | if cfg.get("model", None): |
| 62 | self.model = build_from_cfg(cfg.model, MODELS, logger=self.logger) |
| 63 | else: |
| 64 | self.model = None |
| 65 | self.logger.warning("No model is used in this pipeline") |
| 66 | |
| 67 | # Checkpoint manager is now created inside build_dataloaders |
| 68 | self.dataloaders = build_dataloaders( |
| 69 | cfg.dataloader, |
| 70 | logger=self.logger, |
| 71 | work_dir=self.work_dir, |
| 72 | ) |
| 73 | self.datasaver = build_from_cfg(cfg.datasaver, DATASAVER, logger=self.logger) |
| 74 | |
| 75 | # Pass stats_collector from datasaver to pipeline for unified statistics |
| 76 | stats_collector = getattr(self.datasaver, 'stats_collector', None) |
| 77 | self.pipeline = build_from_cfg( |
| 78 | cfg.pipeline, PIPELINES, logger=self.logger, model=self.model, |
| 79 | stats_collector=stats_collector |
| 80 | ) |
| 81 | |
| 82 | save_file = build_file(self.work_dir, prefix="config.yaml") |
| 83 | save_cfg(save_file) |
| 84 | |
| 85 | def run(self): |
| 86 | """Execute the runner in the configured mode.""" |