Spaces:
Sleeping
Sleeping
| from typing import Any | |
| from app.utils.r_core.contract import ALL_ANALYSIS_KEYS, META_BASED_KEYS, META_FREE_KEYS, PREPROCESSING_KEYS | |
| from app.utils.r_core.job import AnalysisSpec, InputData | |
| def _has_methods(section: Any) -> bool: | |
| return bool(isinstance(section, dict) and section.get("methods")) | |
| def _section_enabled(section: Any) -> bool: | |
| return bool(isinstance(section, dict) and section.get("enabled")) | |
| def normalize_params(params: Any) -> dict[str, Any]: | |
| if not isinstance(params, dict): | |
| return {} | |
| normalized = {key: value for key, value in params.items() if key in ALL_ANALYSIS_KEYS} | |
| normalization = normalized.get("normalization") | |
| if isinstance(normalization, dict) and normalization.get("enabled") and not normalization.get("method"): | |
| normalized["normalization"] = {**normalization, "method": "vector"} | |
| smoothing = normalized.get("smoothing") | |
| if isinstance(smoothing, dict) and smoothing.get("enabled") and not smoothing.get("methods"): | |
| normalized["smoothing"] = {**smoothing, "methods": ["savitzky-golay"]} | |
| baseline_correction = normalized.pop("baseline_correction", None) | |
| baseline = normalized.get("baseline") | |
| if isinstance(baseline_correction, dict) and baseline_correction.get("enabled"): | |
| if not isinstance(baseline, dict): | |
| baseline = {} | |
| if not baseline.get("methods"): | |
| baseline = {**baseline, "methods": ["polyfit"]} | |
| if baseline_correction.get("parameters") and not baseline.get("parameters"): | |
| baseline = {**baseline, "parameters": baseline_correction["parameters"]} | |
| normalized["baseline"] = baseline | |
| elif baseline_correction is not None and "baseline" not in normalized: | |
| normalized["baseline_correction"] = baseline_correction | |
| return normalized | |
| def selected_methods(params: dict[str, Any]) -> dict[str, list[str]]: | |
| selected: dict[str, list[str]] = {} | |
| for key, section in params.items(): | |
| if isinstance(section, dict) and isinstance(section.get("methods"), list): | |
| selected[key] = [str(method) for method in section["methods"]] | |
| elif key == "normalization" and isinstance(section, dict) and section.get("method"): | |
| selected[key] = [str(section["method"])] | |
| elif _section_enabled(section): | |
| selected[key] = ["enabled"] | |
| return selected | |
| def has_selected_work(params: dict[str, Any]) -> bool: | |
| selected = selected_methods(params) | |
| if selected: | |
| return True | |
| normalization = params.get("normalization") | |
| return bool(isinstance(normalization, dict) and normalization.get("method")) | |
| def build_execution_plan(params: dict[str, Any]) -> list[str]: | |
| plan: list[str] = [] | |
| if any(key in params for key in PREPROCESSING_KEYS): | |
| plan.append("preprocessing") | |
| if any(_has_methods(params.get(key)) for key in META_FREE_KEYS): | |
| plan.append("algorithm_analysis") | |
| if any(_has_methods(params.get(key)) for key in META_BASED_KEYS): | |
| plan.append("metadata_algorithm_analysis") | |
| return plan | |
| def infer_analysis_type(params: dict[str, Any]) -> str: | |
| if any(_has_methods(params.get(key)) for key in META_BASED_KEYS): | |
| return "meta_based" | |
| return "meta_free" | |
| def parse_analysis_spec( | |
| project_id: str, | |
| params: dict[str, Any], | |
| input_data: InputData, | |
| bands_csv_path=None, | |
| ) -> AnalysisSpec: | |
| normalized = normalize_params(params) | |
| return AnalysisSpec( | |
| project_id=project_id, | |
| params=normalized, | |
| input_data=input_data, | |
| selected_methods=selected_methods(normalized), | |
| execution_plan=build_execution_plan(normalized), | |
| analysis_type=infer_analysis_type(normalized), | |
| bands_csv_path=bands_csv_path, | |
| ) | |