Spaces:
Sleeping
Sleeping
File size: 3,763 Bytes
57528c5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | 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,
)
|