import io
import os
from itertools import chain
from pathlib import Path
from typing import List
import pandas as pd
from immuneML.environment.EnvironmentSettings import EnvironmentSettings
from immuneML.ml_metrics.ClusteringMetric import is_internal, is_external
from immuneML.presentation.TemplateParser import TemplateParser
from immuneML.presentation.html.Util import Util
from immuneML.ml_methods.util.Util import Util as MLUtil
from immuneML.reports.ReportOutput import ReportOutput
from immuneML.reports.ReportResult import ReportResult
from immuneML.util.PathBuilder import PathBuilder
from immuneML.workflows.instructions.clustering.ClusteringInstruction import ClusteringState
[docs]
class ClusteringHTMLBuilder:
CSS_PATH = EnvironmentSettings.html_templates_path / "css/custom.css"
[docs]
@staticmethod
def build(state: ClusteringState) -> Path:
base_path = PathBuilder.build(state.result_path / "../HTML_output/")
html_map = ClusteringHTMLBuilder.make_html_map(state, base_path)
result_file = base_path / f"Clustering_{state.name}.html"
TemplateParser.parse(template_path=EnvironmentSettings.html_templates_path / "Clustering.html",
template_map=html_map, result_path=result_file)
return result_file
[docs]
@staticmethod
def make_html_map(state: ClusteringState, base_path: Path) -> dict:
html_map = {
"css_style": Util.get_css_content(ClusteringHTMLBuilder.CSS_PATH),
"name": state.name,
'immuneML_version': MLUtil.get_immuneML_version(),
"full_specs": Util.get_full_specs_path(base_path),
"logfile": Util.get_logfile_path(base_path),
"discovery_predictions_path": Path(os.path.relpath(path=str(state.predictions_paths['discovery']), start=str(base_path))),
"validation_predictions_path": Path(
os.path.relpath(path=str(state.predictions_paths['validation']), start=str(base_path))),
"performance_table_internal_val": ClusteringHTMLBuilder.make_internal_performance_table(state, 'validation'),
"performance_table_internal_disc": ClusteringHTMLBuilder.make_internal_performance_table(state, 'discovery'),
"label_eval_tables_val": ClusteringHTMLBuilder.make_external_performance_tables(state, 'validation'),
"label_eval_tables_disc": ClusteringHTMLBuilder.make_external_performance_tables(state, 'discovery'),
"cluster_setting_pages_val": ClusteringHTMLBuilder.make_cluster_setting_pages(state, base_path, 'validation'),
"cluster_setting_pages_disc": ClusteringHTMLBuilder.make_cluster_setting_pages(state, base_path, 'discovery')
}
html_map = {**html_map, **{'show_internal_eval': sum([is_internal(m) for m in state.metrics]) > 0,
'show_external_eval': sum([is_external(m) for m in state.metrics]) > 0,
"show_cluster_setting_pages_val": len(html_map['cluster_setting_pages_val']) > 0,
"show_cluster_setting_pages_disc": len(html_map['cluster_setting_pages_disc']) > 0},
**Util.make_dataset_html_map(state.dataset)}
return html_map
[docs]
@staticmethod
def make_cluster_setting_pages(state: ClusteringState, base_path: Path, analysis_desc: str) -> List[dict]:
cluster_setting_pages = []
for cl_setting in state.clustering_settings:
if (state.cl_item_report_results[analysis_desc][cl_setting.get_key()] is not None
and len(state.cl_item_report_results[analysis_desc][cl_setting.get_key()]) > 0):
cluster_setting_pages.append({
'name': cl_setting.get_key(),
'path': ClusteringHTMLBuilder.make_cluster_setting_page(state, cl_setting, base_path, analysis_desc)
})
return cluster_setting_pages
[docs]
@staticmethod
def make_cluster_setting_page(state: ClusteringState, cl_setting, base_path: Path, analysis_desc: str) -> Path:
result_path = base_path / f"{state.name}_{analysis_desc}_clustering_setting_{cl_setting.get_key()}.html"
if state.cl_item_report_results[analysis_desc][cl_setting.get_key()]:
report_results = Util.to_dict_recursive(state.cl_item_report_results[analysis_desc][cl_setting.get_key()], base_path)
report_results = list(chain.from_iterable(report_results.values()))
report_results = ClusteringHTMLBuilder._move_reports_recursive(report_results, base_path)
else:
report_results = None
html_map = {
"css_style": Util.get_css_content(ClusteringHTMLBuilder.CSS_PATH),
"cl_setting_name": cl_setting.get_key(),
"analysis_desc": analysis_desc,
"reports": report_results,
'show_internal_performances': any(is_internal(m) for m in state.metrics)
and state.clustering_items[analysis_desc][cl_setting.get_key()].internal_performance,
'show_external_performances': any(is_external(m) for m in state.metrics)
and state.clustering_items[analysis_desc][cl_setting.get_key()].external_performance
}
if state.clustering_items[analysis_desc][cl_setting.get_key()].internal_performance.path is not None and \
state.clustering_items[analysis_desc][cl_setting.get_key()].internal_performance.path.is_file():
with state.clustering_items[analysis_desc][cl_setting.get_key()].internal_performance.path.open('r') as file:
html_map["internal_performances"] = Util.get_table_string_from_csv_string(file.read(), separator=",")
else:
html_map["show_internal_performances"] = False
if state.clustering_items[analysis_desc][cl_setting.get_key()].external_performance.path is not None and \
state.clustering_items[analysis_desc][cl_setting.get_key()].external_performance.path.is_file():
with state.clustering_items[analysis_desc][cl_setting.get_key()].external_performance.path.open('r') as file:
html_map['external_performances'] = Util.get_table_string_from_csv_string(csv_string=file.read(), separator=",", has_header=True)
else:
html_map["external_performances"] = False
TemplateParser.parse(template_path=EnvironmentSettings.html_templates_path / "ClusteringSettingDetails.html",
template_map=html_map, result_path=result_path)
return result_path.relative_to(base_path)
@staticmethod
def _move_reports_recursive(obj: List[ReportResult], path: Path):
if isinstance(obj, list) and all(isinstance(item, ReportResult) for item in obj):
new_obj = []
for report_result in obj:
rep_result = Util.update_report_paths(report_result, path)
for attribute in vars(rep_result):
attribute_value = getattr(report_result, attribute)
if isinstance(attribute_value, list):
for output in attribute_value:
if isinstance(output, ReportOutput):
output.path = output.path.relative_to(path)
rep_result = Util.to_dict_recursive(rep_result, path)
new_obj.append(rep_result)
return new_obj
else:
return obj