| 432 | ) |
| 433 | |
| 434 | def _run_pipeline(self) -> None: |
| 435 | # Validate if given stage sequence is valid |
| 436 | self._validate_pipeline_sequence( |
| 437 | stages=self._pipeline_stages, |
| 438 | ) |
| 439 | |
| 440 | current_artifact = PipelineArtifact(data=self._model, context=self._run_context) |
| 441 | |
| 442 | # Execute stages from registry in the order specified by pipeline_stages |
| 443 | for stage_type in self._pipeline_stages: |
| 444 | stage = self._stage_registry.get(stage_type) |
| 445 | if stage is None: |
| 446 | raise ValueError(f"Stage {stage_type} not found in registry") |
| 447 | |
| 448 | logging.info(f"Executing stage: {stage_type}") |
| 449 | |
| 450 | start = time.perf_counter() |
| 451 | stage.run(current_artifact) |
| 452 | elapsed = (time.perf_counter() - start) * 1000 |
| 453 | current_artifact = stage.get_artifacts() |
| 454 | current_artifact.add_context("duration_ms", int(elapsed)) |
| 455 | |
| 456 | logging.info(f"Stage {stage_type} execution done") |
| 457 | |
| 458 | self._stage_to_artifacts[stage_type] = current_artifact |
| 459 | |
| 460 | def export(self) -> None: |
| 461 | """ |