pathforge.policy¶

Use-case layer. Policies coordinate domain objects via registries and the config without importing concrete framework packages directly.

Base¶

class pathforge.policy.base.PolicyBase[source]¶

Bases: ABC

Base class for PathForge policies.

Policies may be constructed either from an experiment-like object exposing a cfg attribute or directly from a validated config object in legacy code paths. In both cases self.cfg is guaranteed to point at the active configuration object.

__init__(experiment: ExperimentLike | Any) → None[source]¶

Parameters:: experiment (ExperimentLike | Any)
Return type:: None

abstractmethod execute(*args: Any, **kwargs: Any) → Any[source]¶

Execute the policy mode in the given experiment context.

Parameters:

args (Any)
kwargs (Any)

Return type:

Any

Feature Extraction¶

class pathforge.policy.feature_extraction.FeatureExtractionPolicy[source]¶

Bases: PolicyBase

Extract tile features into per-slide H5 artifacts.

combo_keys = ['feature_extraction', 'tile_px', 'tile_mpp', 'color_norm']¶

__init__(experiment: Experiment)[source]¶

Parameters:: experiment (Experiment)

build_combos() → list[ComboConfig][source]¶

Return type:: list[ComboConfig]

execute(*, datasets: list[WSIDataset] | None = None, combos: list[ComboConfig] | None = None) → dict[str, Any][source]¶

Execute the policy mode in the given experiment context.

Parameters:

datasets (list[WSIDataset] | None)
combos (list[ComboConfig] | None)

Return type:

dict[str, Any]

execute_combo(*, combo_cfg: ComboConfig, datasets: list[WSIDataset]) → None[source]¶

Parameters:

combo_cfg (ComboConfig)
datasets (list[WSIDataset])

Return type:

None

execute_dataset(dataset: WSIDataset, combo_cfg: ComboConfig) → None[source]¶

Parameters:

dataset (WSIDataset)
combo_cfg (ComboConfig)

Return type:

None

process_slide(dataset: WSIDataset, wsi: WSI, combo_cfg: ComboConfig) → None[source]¶

Parameters:

dataset (WSIDataset)
wsi (WSI)
combo_cfg (ComboConfig)

Return type:

None

Benchmarking¶

class pathforge.policy.benchmarking.BenchmarkingPolicy[source]¶

Bases: PolicyBase

Run PathForge benchmarking in experiment or legacy config mode.

__init__(experiment: Experiment | Config | Any)[source]¶

Initialize one benchmarking policy over an experiment-like context.

Parameters:: experiment (Experiment | Config | Any) – Either a fully prepared Experiment or a legacy config-like object accepted by PolicyBase.

property feature_policy: FeatureExtractionPolicy | None¶

Lazily build the feature-extraction policy on first access.

The policy is only available in experiment-context mode; in legacy config mode this returns None.

execute() → dict[str, Any][source]¶

Run the configured benchmark workflow.

Returns:: Summary containing at least status and num_runs. In experiment mode, each run delegates to the task-specific benchmark executor. In legacy mode, the method evaluates expanded search-space configs directly.
Return type:: dict[str, Any]

execute_combination(combo_cfg: ComboConfig) → dict[str, Any][source]¶

Run one fully specified combo through feature resolution and task execution.

Parameters:: combo_cfg (ComboConfig) – Benchmark combination describing one bag/model/task run.
Returns:: One-run benchmark summary including the delegated task output.
Return type:: dict[str, Any]

ensure_bag_features_exist(*, combo_cfg: ComboConfig, annotations_df: DataFrame | None = None) → None[source]¶

Ensure all required slide-level bag features exist for one combo.

Parameters:

combo_cfg (ComboConfig) – Benchmark combination whose bag source must exist.
annotations_df (DataFrame | None) – Optional annotation table. When omitted, annotations are loaded from the experiment.

Return type:

None

build_bag_datasets_for_combo(*, combo_cfg: ComboConfig, annotations_df: DataFrame | None = None) → list[Any][source]¶

Build task-ready bag datasets for one benchmark combo.

Parameters:

combo_cfg (ComboConfig)
annotations_df (DataFrame | None)

Return type:

list[Any]

group_bag_datasets_by_use(bag_datasets: list[Any]) → dict[str, list[Any]][source]¶

Group bag datasets by their configured used_for role.

Parameters:: bag_datasets (list[Any])
Return type:: dict[str, list[Any]]

Optimization¶

class pathforge.policy.optimization.OptimizationPolicy[source]¶

Bases: PolicyBase

Optimization Policy that dynamically configures Optuna based on Config.

__init__(config: Config) → None[source]¶

Parameters:: config (Config)
Return type:: None

objective(trial: optuna.Trial) → float[source]¶

Parameters:: trial (optuna.Trial)
Return type:: float

execute(*, n_trials: int | None = None, finalize: bool = True) → optuna.Study[source]¶

Run local or shared-storage trials and optionally write study outputs.

Parameters:

n_trials (int | None)
finalize (bool)

Return type:

optuna.Study

Inference¶

class pathforge.policy.inference.InferencePolicy[source]¶

Bases: PolicyBase

Run task-specific inference over an explicit CSV-selected input batch.

Inputs:

experiment (Experiment):: Experiment context with validated config, project metadata, and annotations.

Outputs:

dict[str, Any]:: Inference execution summary with the timestamped invocation folder and per-combination task outputs.

__init__(experiment: Experiment)[source]¶

Parameters:: experiment (Experiment)

execute(*, input_csv: Path) → dict[str, Any][source]¶

Execute inference for every configured inference combination.

Inputs:

input_csv (Path):: CSV file selecting the slides to use as inference/query inputs. Required columns are dataset and slide.

Outputs:

dict[str, Any]:: Summary with status, output_dir, num_runs, and task_outputs.

Parameters:: input_csv (Path)
Return type:: dict[str, Any]

Shared Utilities¶

class pathforge.policy.utils.ComboConfig[source]¶

Bases: object

Generic, dynamically-populated benchmark combination configuration.

Inputs:

keyword arguments (object):: Benchmark parameter values keyed by parameter name. Each key becomes an attribute on the created object.

Outputs:

ComboConfig:: Configuration object exposing combo values and optional <key>_params hyperparameter dictionaries.

Semantic goal:

Represent one fully-materialized benchmark parameter combination in a lightweight attribute-based object.

Example

combo_cfg = ComboConfig.from_keys_values(
    keys=["feature_extraction", "tile_px"],
    values=[
        BenchmarkParamEntry(value="uni", hyperparams={"family": "foundation"}),
        256,
    ],
)
assert combo_cfg.feature_extraction == "uni"
assert combo_cfg.get_hyperparams("feature_extraction") == {
    "family": "foundation",
}

__init__(**kwargs: object) → None[source]¶

Parameters:: kwargs (object)
Return type:: None

classmethod from_keys_values(keys: list[str], values: list[object]) → ComboConfig[source]¶

Build one combo config from aligned key/value lists.

Inputs:

keys (list[str]):: Parameter names. Shape: (n_keys,).
values (list[object]):: Parameter values aligned with keys. Shape: (n_keys,).

Outputs:

ComboConfig:: Combo object containing one attribute per key and one <key>_params attribute per key.

Parameters:

keys (list[str])
values (list[object])

Return type:

ComboConfig

to_dict() → dict[str, object][source]¶

Return a shallow dictionary representation of the combo.

Return type:: dict[str, object]

get(key: str, default: object = None) → object[source]¶

Return one combo value by key with a dict-like fallback default.

Parameters:

key (str)
default (object)

Return type:

object

get_hyperparams(key: str, default: dict[str, Any] | None = None) → dict[str, Any][source]¶

Return the hyperparameters attached to one combo value.

Parameters:

key (str)
default (dict[str, Any] | None)

Return type:

dict[str, Any]

pathforge.policy.utils.apply_search_params(config: Config, params: dict[str, Any]) → Config[source]¶

Apply one benchmark/optimization parameter set to a config copy.

Parameters:

config (Config)
params (dict[str, Any])

Return type:

Config

pathforge.policy.utils.benchmark_search_space(config: Config) → dict[str, list[Any]][source]¶

Return the non-empty benchmark grid declared in the config.

Parameters:: config (Config)
Return type:: dict[str, list[Any]]

pathforge.policy.utils.build_mil_model_for_config(config: Config, *, model_name: str, input_dim: int, output_dim: int, extra_kwargs: dict[str, Any] | None = None) → Any[source]¶

Build one MIL model while respecting backend-specific user config.

Parameters:

config (Config) – Active PathForge configuration.
model_name (str) – Registry key selected for this run.
input_dim (int) – Bag feature dimensionality inferred from BagDataset.
output_dim (int) – Task output dimensionality inferred from annotations.
extra_kwargs (dict[str, Any] | None) – Optional extra constructor kwargs supplied by callers.

Returns:

Instantiated model object from the MODELS registry.

Return type:

Any

pathforge.policy.utils.build_bag_dataset_for_task(config: Config, *, feature_dir: str | Path, name: str, dataset_entry: Any | None = None) → BagDataset[source]¶

Construct a task-aware bag dataset from config and a feature directory.

Parameters:

config (Config)
feature_dir (str | Path)
name (str)
dataset_entry (Any | None)

Return type:

BagDataset

pathforge.policy.utils.calculate_combinations(params: dict[str, list[Any]]) → list[dict[str, Any]][source]¶

Calculate parameter combinations from a mapping of names to candidate values.

This wrapper remains for compatibility, but ComboConfig is canonically implemented in pathforge.core.experiments.base to avoid duplicate implementations across PathForge.

Parameters:: params (dict[str, list[Any]])
Return type:: list[dict[str, Any]]

pathforge.policy.utils.infer_model_dimensions(dataset: BagDataset) → tuple[int, int][source]¶

Infer (input_dim, output_dim) from one prepared bag dataset.

Parameters:: dataset (BagDataset)
Return type:: tuple[int, int]

pathforge.policy.utils.collect_run_summary_row(config: Config, *, run_index: int, status: str, objective_metric: str, objective_value: float | None = None, checkpoint_path: str | None = None, error: str | None = None, extra_fields: dict[str, Any] | None = None) → dict[str, Any][source]¶

Build one flat experiment-summary row for a benchmark or optimization run.

Parameters:

config (Config) – Active run configuration.
run_index (int) – Zero-based run index within the experiment.
status (str) – Run status such as "success" or "failed".
objective_metric (str) – Metric used for run ranking.
objective_value (float | None) – Metric value for the completed run.
checkpoint_path (str | None) – Optional best-checkpoint path.
error (str | None) – Optional error message for failed runs.
extra_fields (dict[str, Any] | None) – Optional extra scalar fields to append.

Returns:

Flat summary row safe to store in CSV.

Return type:

dict[str, Any]

pathforge.policy.utils.experiment_output_root(config: Config) → Path[source]¶

Return the experiment-level output directory for policy artifacts.

Parameters:: config (Config) – Active PathForge configuration.
Returns:: Absolute output directory used for experiment-wide reports.
Return type:: Path

pathforge.policy.utils.metric_should_minimize(metric_name: str) → bool[source]¶

Return whether lower values should rank higher for one metric name.

Parameters:: metric_name (str)
Return type:: bool

pathforge.policy.utils.optimization_search_space(config: Config) → dict[str, SearchSpaceParameter][source]¶

Return the merged optimization search space for one config.

The optimization policy samples hyperparameter ranges from optimization.search_space and automatically exposes every non-empty benchmark grid list as a categorical Optuna search dimension. This keeps pipeline-component selection and training hyperparameter search aligned between benchmarking and optimization modes.

Parameters:: config (Config)
Return type:: dict[str, SearchSpaceParameter]

pathforge.policy.utils.resolve_dataset_feature_dir(dataset_entry: Any) → Path[source]¶

Resolve the directory containing bag feature files for one dataset.

Parameters:: dataset_entry (Any)
Return type:: Path

pathforge.policy.utils.save_benchmark_visualizations(summary_csv_path: Path, *, output_dir: Path, objective_metric: str, minimize: bool, logger: Logger | None = None) → list[Path][source]¶

Build ranked benchmark-wide visualizations from one summary CSV.

Parameters:

summary_csv_path (Path)
output_dir (Path)
objective_metric (str)
minimize (bool)
logger (Logger | None)

Return type:

list[Path]

pathforge.policy.utils.save_global_summary_visualizations(summary_csv_path: Path, *, output_dir: Path, title_prefix: str = 'Pipeline Results') → list[Path][source]¶

Render reusable ranked HTML views from a saved global results CSV.

Parameters:

summary_csv_path (Path) – Benchmark or optimization summary written by write_experiment_summary_csv().
output_dir (Path) – Directory receiving the standalone HTML files.
title_prefix (str) – Human-readable label used in chart titles.

Returns:

Created HTML paths. An empty list is returned when the CSV: has no successful finite objective values.

Return type:

list[Path]

Example

files = save_global_summary_visualizations(
    Path("benchmark_results.csv"),
    output_dir=Path("summary_visualizations"),
)

pathforge.policy.utils.save_optuna_visualizations(study: Any, *, output_dir: Path, logger: Logger | None = None) → list[Path][source]¶

Export supported Optuna visualization figures as HTML and PNG files.

The current Optuna documentation exposes Plotly-backed visualization helpers including plot_param_importances, plot_optimization_history, plot_rank, plot_timeline, and plot_hypervolume_history. Hypervolume history is exported only for multi-objective studies because Optuna requires at least two objectives and a caller-provided reference point.

Parameters:

study (Any)
output_dir (Path)
logger (Logger | None)

Return type:

list[Path]

pathforge.policy.utils.suggest_parameter(trial: Any, *, name: str, spec: SearchSpaceParameter) → Any[source]¶

Suggest one optimization parameter from a validated search-space spec.

Parameters:

trial (Any)
name (str)
spec (SearchSpaceParameter)

Return type:

Any

pathforge.policy.utils.write_experiment_summary_csv(rows: list[dict[str, Any]], *, output_path: Path, objective_metric: str, minimize: bool) → DataFrame[source]¶

Write one sorted experiment-wide summary CSV and return its dataframe.

Parameters:

rows (list[dict[str, Any]])
output_path (Path)
objective_metric (str)
minimize (bool)

Return type:

DataFrame