Source code for immuneML.workflows.instructions.MLProcess

import copy
import datetime
from pathlib import Path
from typing import List

from immuneML.data_model.dataset.Dataset import Dataset
from immuneML.environment.Label import Label
from immuneML.environment.LabelConfiguration import LabelConfiguration
from immuneML.hyperparameter_optimization.HPSetting import HPSetting
from immuneML.hyperparameter_optimization.core.HPUtil import HPUtil
from immuneML.hyperparameter_optimization.states.HPItem import HPItem
from immuneML.ml_metrics.Metric import Metric
from immuneML.reports.ReportUtil import ReportUtil
from immuneML.reports.ml_reports.MLReport import MLReport
from immuneML.util.PathBuilder import PathBuilder


[docs]class MLProcess: """ Class that implements the machine learning process: 1. encodes the training dataset 2. encodes the test dataset (using parameters learnt in step 1 if there are any such parameters) 3. trains the ML method on encoded training dataset 4. assesses the method's performance on encoded test dataset It performs the task for a given label configuration, and given list of metrics (used only in the assessment step). """ def __init__(self, train_dataset: Dataset, test_dataset: Dataset, label: Label, metrics: set, optimization_metric: Metric, path: Path, ml_reports: List[MLReport] = None, encoding_reports: list = None, data_reports: list = None, number_of_processes: int = 2, label_config: LabelConfiguration = None, report_context: dict = None, hp_setting: HPSetting = None): self.train_dataset = train_dataset self.test_dataset = test_dataset self.label = label self.label_config = label_config self.method = copy.deepcopy(hp_setting.ml_method) self.path = PathBuilder.build(path) if path is not None else None self.ml_details_path = path / "ml_details.yaml" if path is not None else None self.ml_score_path = path / "ml_score.csv" if path is not None else None self.train_predictions_path = path / "train_predictions.csv" if path is not None else None self.test_predictions_path = path / "test_predictions.csv" if path is not None else None self.report_path = PathBuilder.build(path / "reports") if path is not None else None self.number_of_processes = number_of_processes assert all([isinstance(metric, Metric) for metric in metrics]), \ "MLProcess: metrics are not set to be an instance of Metric." self.metrics = metrics self.metrics.add(Metric.BALANCED_ACCURACY) self.optimization_metric = optimization_metric self.ml_reports = ml_reports if ml_reports is not None else [] self.encoding_reports = encoding_reports if encoding_reports is not None else [] self.data_reports = data_reports if data_reports is not None else [] self.report_context = report_context self.hp_setting = copy.deepcopy(hp_setting) def _set_paths(self): if self.path is None: raise RuntimeError("MLProcess: path is not set, stopping execution...") self.ml_details_path = self.path / "ml_details.yaml" self.ml_score_path = self.path / "ml_score.csv" self.train_predictions_path = self.path / "train_predictions.csv" self.test_predictions_path = self.path / "test_predictions.csv" self.report_path = PathBuilder.build(self.path / "reports")
[docs] def run(self, split_index: int) -> HPItem: print(f"{datetime.datetime.now()}: Evaluating hyperparameter setting: {self.hp_setting}...", flush=True) PathBuilder.build(self.path) self._set_paths() processed_dataset = HPUtil.preprocess_dataset(self.train_dataset, self.hp_setting.preproc_sequence, self.path / "preprocessed_train_dataset", self.report_context) encoded_train_dataset = HPUtil.encode_dataset(processed_dataset, self.hp_setting, self.path / "encoded_datasets", learn_model=True, context=self.report_context, number_of_processes=self.number_of_processes, label_configuration=self.label_config) method = HPUtil.train_method(self.label, encoded_train_dataset, self.hp_setting, self.path, self.train_predictions_path, self.ml_details_path, self.number_of_processes, self.optimization_metric) encoding_train_results = ReportUtil.run_encoding_reports(encoded_train_dataset, self.encoding_reports, self.report_path / "encoding_train", self.number_of_processes) hp_item = self._assess_on_test_dataset(encoded_train_dataset, encoding_train_results, method, split_index) print(f"{datetime.datetime.now()}: Completed hyperparameter setting {self.hp_setting}.\n", flush=True) return hp_item
def _assess_on_test_dataset(self, encoded_train_dataset, encoding_train_results, method, split_index) -> HPItem: if self.test_dataset is not None and self.test_dataset.get_example_count() > 0: processed_test_dataset = HPUtil.preprocess_dataset(self.test_dataset, self.hp_setting.preproc_sequence, self.path / "preprocessed_test_dataset") encoded_test_dataset = HPUtil.encode_dataset(processed_test_dataset, self.hp_setting, self.path / "encoded_datasets", learn_model=False, context=self.report_context, number_of_processes=self.number_of_processes, label_configuration=self.label_config) performance = HPUtil.assess_performance(method, self.metrics, self.optimization_metric, encoded_test_dataset, split_index, self.path, self.test_predictions_path, self.label, self.ml_score_path) encoding_test_results = ReportUtil.run_encoding_reports(encoded_test_dataset, self.encoding_reports, self.report_path / "encoding_test", self.number_of_processes) model_report_results = ReportUtil.run_ML_reports(encoded_train_dataset, encoded_test_dataset, method, self.ml_reports, self.report_path / "ml_method", self.hp_setting, self.label, self.number_of_processes, self.report_context) hp_item = HPItem(method=method, hp_setting=self.hp_setting, train_predictions_path=self.train_predictions_path, test_predictions_path=self.test_predictions_path, ml_details_path=self.ml_details_path, train_dataset=self.train_dataset, test_dataset=self.test_dataset, split_index=split_index, model_report_results=model_report_results, encoding_train_results=encoding_train_results, encoding_test_results=encoding_test_results, performance=performance, encoder=self.hp_setting.encoder) else: hp_item = HPItem(method=method, hp_setting=self.hp_setting, train_predictions_path=self.train_predictions_path, test_predictions_path=None, ml_details_path=self.ml_details_path, train_dataset=self.train_dataset, split_index=split_index, encoding_train_results=encoding_train_results, encoder=self.hp_setting.encoder) return hp_item