from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
from anndata import AnnData
from revise.svc import SVC
[docs]
@dataclass
class PipelineContext:
merged_config: Dict[str, Any]
raw_config: Dict[str, Any]
config_path: str
profile: Optional[str]
runtime: Dict[str, Any]
route_key: str
run_dir: Path
logger: logging.Logger
dry_run: bool = False
legacy_config: Any = None
legacy_runner: Any = None
st_adata: Optional[AnnData] = None
sc_ref_adata: Optional[AnnData] = None
real_st_adata: Optional[AnnData] = None
svc: Optional[SVC] = None
stage_trace: List[str] = field(default_factory=list)
quality_metrics: Dict[str, Any] = field(default_factory=dict)
artifacts: Dict[str, Any] = field(default_factory=dict)
provenance: Dict[str, Any] = field(default_factory=dict)
@property
def io(self) -> Dict[str, Any]:
return self.merged_config.get("io", {})
@property
def columns(self) -> Dict[str, Any]:
return self.merged_config.get("columns", {})
@property
def legacy_mode(self) -> bool:
return bool(self.runtime.get("legacy_mode", False))
@property
def route(self) -> Dict[str, Any]:
return {
"platform": self.runtime.get("platform"),
"confounding": self.runtime.get("confounding"),
"mode": self.runtime.get("mode"),
"task": self.runtime.get("task"),
"svc_kind": self.runtime.get("svc_kind"),
"strategy": self.runtime.get("strategy"),
"platform_adapter": self.runtime.get("platform_adapter"),
"cf_strategy": self.runtime.get("cf_strategy"),
"ot_solver": self.runtime.get("ot_solver"),
}