"""Fixed Origin adapter for origin.analysis@v1. The worker accepts one Node-created job directory. It performs only versioned, declarative analysis operations, writes the raw data and results into an Origin project, and never evaluates user code or accepts user-controlled paths. """ from __future__ import annotations import csv import hashlib import json import math import os import sys from datetime import datetime, timezone from importlib.metadata import PackageNotFoundError, version from itertools import pairwise from pathlib import Path from typing import Any import numpy as np ADAPTER_VERSION = "0.1.0" ALGORITHM_VERSION = "origin-analysis-mvp-1" OPERATIONS = { "data_check", "normalize", "smooth", "differentiate", "integrate", "linear_fit", } OUTPUT_MEDIA = { "analysis.opju": ("project", "application/x-origin-project"), "result-table.csv": ("result_table", "text/csv"), "result-table.xlsx": ( "result_workbook", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ), "diagnostics.json": ("diagnostics", "application/json"), "analysis-spec.json": ("analysis_spec", "application/json"), "provenance.json": ("provenance", "application/json"), } def _probe() -> int: health = "ready" detail = "Origin COM 与科研分析运行时可用" software_version = None try: if sys.platform != "win32": raise RuntimeError("Origin analysis adapter requires Windows") import winreg import originpro with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, r"Origin.ApplicationSI\CLSID"): pass originpro_version = version("originpro") numpy_version = version("numpy") detail = ( "Origin COM 与科研分析运行时可用" f"（originpro {originpro_version}，numpy {numpy_version}）" ) del originpro except ( FileNotFoundError, ImportError, OSError, PackageNotFoundError, RuntimeError, ) as exc: health = "unavailable" detail = str(exc) print( json.dumps( { "adapter_version": ADAPTER_VERSION, "software": "OriginPro", "software_version": software_version, "health": health, "detail": detail, }, ensure_ascii=False, ) ) return 0 def _atomic_json(path: Path, value: Any) -> None: temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex()) try: with temporary.open("w", encoding="utf-8", newline="\n") as handle: json.dump(value, handle, ensure_ascii=False, indent=2, allow_nan=False) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _atomic_csv(path: Path, headers: list[str], rows: list[list[Any]]) -> None: temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex()) try: with temporary.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.writer(handle, lineterminator="\n") writer.writerow(headers) writer.writerows(rows) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _atomic_xlsx(path: Path, headers: list[str], rows: list[list[Any]]) -> None: from openpyxl import Workbook temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex() + ".xlsx") try: workbook = Workbook(write_only=True) worksheet = workbook.create_sheet("Result") worksheet.append(headers) for row in rows: worksheet.append(row) workbook.save(temporary) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _read_rows(path: Path, sheet: str | None) -> tuple[list[str], list[list[Any]]]: suffix = path.suffix.lower() if suffix == ".csv": with path.open("r", encoding="utf-8-sig", newline="") as handle: rows = list(csv.reader(handle)) if len(rows) < 2: raise ValueError("CSV_INPUT_EMPTY") return [str(item) for item in rows[0]], rows[1:] if suffix == ".json": value = json.loads(path.read_text(encoding="utf-8")) if isinstance(value, list) and value and all(isinstance(item, dict) for item in value): headers = list(value[0]) if any(set(item) != set(headers) for item in value): raise ValueError("JSON_OBJECT_COLUMNS_MISMATCH") return headers, [[item.get(name) for name in headers] for item in value] if ( isinstance(value, dict) and value and all(isinstance(item, list) for item in value.values()) ): headers = list(value) length = max(len(value[name]) for name in headers) return headers, [ [value[name][index] if index < len(value[name]) else None for name in headers] for index in range(length) ] raise ValueError("JSON_INPUT_SHAPE_UNSUPPORTED") if suffix == ".xlsx": from openpyxl import load_workbook workbook = load_workbook(path, read_only=True, data_only=True) try: worksheet = workbook[sheet] if sheet else workbook.active rows = list(worksheet.iter_rows(values_only=True)) finally: workbook.close() if len(rows) < 2: raise ValueError("XLSX_INPUT_EMPTY") return [str(item or "") for item in rows[0]], [list(row) for row in rows[1:]] raise ValueError("INPUT_TYPE_UNSUPPORTED") def _input_file(job_dir: Path, key: str) -> Path: directory = job_dir / "input" / key files = [ path for path in directory.iterdir() if path.is_file() and not path.name.startswith(".") ] if len(files) != 1: raise ValueError(f"INPUT_FILE_COUNT_INVALID:{key}") return files[0] def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _manifest(path: Path) -> dict[str, Any]: artifact_id, media_type = OUTPUT_MEDIA[path.name] return { "artifact_id": artifact_id, "filename": path.name, "media_type": media_type, "size_bytes": path.stat().st_size, "sha256": _file_sha256(path), } def _missing(value: Any) -> bool: return value is None or (isinstance(value, str) and not value.strip()) def _finite_number(value: Any, role: str, row_number: int) -> float: if isinstance(value, bool) or _missing(value): raise ValueError(f"{role.upper()}_VALUE_NOT_NUMERIC:row={row_number}") try: result = float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{role.upper()}_VALUE_NOT_NUMERIC:row={row_number}") from exc if not math.isfinite(result): raise ValueError(f"{role.upper()}_VALUE_NOT_FINITE:row={row_number}") return result def _column_index(headers: list[str], name: Any, role: str) -> int: if not isinstance(name, str) or name not in headers: raise ValueError(f"{role.upper()}_COLUMN_NOT_FOUND") return headers.index(name) def _xy_values( headers: list[str], rows: list[list[Any]], binding: dict[str, Any] ) -> tuple[np.ndarray, np.ndarray]: x_index = _column_index(headers, binding.get("x"), "x") y_index = _column_index(headers, binding.get("y"), "y") x = np.array( [ _finite_number(row[x_index] if x_index < len(row) else None, "x", index) for index, row in enumerate(rows, start=2) ], dtype=float, ) y = np.array( [ _finite_number(row[y_index] if y_index < len(row) else None, "y", index) for index, row in enumerate(rows, start=2) ], dtype=float, ) if len(x) < 2: raise ValueError("ANALYSIS_REQUIRES_AT_LEAST_TWO_ROWS") differences = np.diff(x) if np.any(differences == 0): raise ValueError("X_VALUES_DUPLICATED") if np.any(differences < 0): raise ValueError("X_VALUES_NOT_STRICTLY_INCREASING") return x, y def _json_number(value: float) -> float | None: return float(value) if math.isfinite(float(value)) else None def _column_profile(name: str, values: list[Any]) -> dict[str, Any]: missing = sum(_missing(item) for item in values) numeric: list[float] = [] non_numeric = 0 non_finite = 0 for item in values: if _missing(item): continue if isinstance(item, bool): non_numeric += 1 continue try: number = float(item) except (TypeError, ValueError): non_numeric += 1 continue if not math.isfinite(number): non_finite += 1 continue numeric.append(number) return { "column": name, "row_count": len(values), "non_missing": len(values) - missing, "missing": missing, "numeric": len(numeric), "non_numeric": non_numeric, "non_finite": non_finite, "minimum": min(numeric) if numeric else None, "maximum": max(numeric) if numeric else None, "mean": float(np.mean(numeric)) if numeric else None, } def _data_check( headers: list[str], rows: list[list[Any]], spec: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: profiles = [ _column_profile( header, [row[index] if index < len(row) else None for row in rows], ) for index, header in enumerate(headers) ] issues: list[dict[str, Any]] = [] if any(item["missing"] for item in profiles): issues.append({"code": "missing", "detail": "输入包含缺失值。"}) if any(item["non_finite"] for item in profiles): issues.append({"code": "non_finite", "detail": "输入包含非有限数值。"}) binding = spec["data"] if "x" in binding: x_index = _column_index(headers, binding["x"], "x") x_values: list[float] = [] x_complete = True for row_number, row in enumerate(rows, start=2): try: x_values.append( _finite_number( row[x_index] if x_index < len(row) else None, "x", row_number, ) ) except ValueError: x_complete = False break if x_complete and len(x_values) >= 2: differences = np.diff(np.array(x_values, dtype=float)) if np.any(differences == 0): issues.append({"code": "duplicate_x", "detail": "X 列包含重复值。"}) if np.any(differences < 0): issues.append( {"code": "non_monotonic_x", "detail": "X 列不是严格递增。"} ) positive = differences[differences > 0] if len(positive) >= 2: relative_span = float(np.ptp(positive) / np.mean(positive)) if relative_span > 1e-6: issues.append( { "code": "uneven_spacing", "detail": "X 列采样间隔不均匀。", "relative_spacing_span": relative_span, } ) fail_on = set(spec["parameters"]["fail_on"]) failed_codes = sorted({item["code"] for item in issues} & fail_on) result_headers = [ "column", "row_count", "non_missing", "missing", "numeric", "non_numeric", "non_finite", "minimum", "maximum", "mean", ] result_rows = [[item.get(header) for header in result_headers] for item in profiles] diagnostics = { "operation": "data_check", "passed": not failed_codes, "failed_checks": failed_codes, "issues": issues, "row_count": len(rows), "column_count": len(headers), } return result_headers, result_rows, diagnostics def _normalize( x: np.ndarray, y: np.ndarray, parameters: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: method = parameters["method"] selected_x = None if method == "max_abs": divisor = float(np.max(np.abs(y))) elif method == "area": divisor = float(np.trapezoid(y, x)) elif method == "reference": reference_x = float(parameters["reference_x"]) if reference_x < x[0] or reference_x > x[-1]: raise ValueError("REFERENCE_X_OUTSIDE_DATA_RANGE") divisor = float(np.interp(reference_x, x, y)) selected_x = reference_x else: raise ValueError("NORMALIZE_METHOD_UNSUPPORTED") scale = max(1.0, float(np.max(np.abs(y)))) if abs(divisor) <= np.finfo(float).eps * scale: raise ValueError("NORMALIZE_DIVISOR_IS_ZERO") normalized = y / divisor rows = [[float(a), float(b), float(c)] for a, b, c in zip(x, y, normalized, strict=True)] return ["x", "y", "normalized_y"], rows, { "operation": "normalize", "method": method, "divisor": divisor, "reference_x": selected_x, "row_count": len(rows), } def _smooth( x: np.ndarray, y: np.ndarray, parameters: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: window = int(parameters["window"]) polynomial_order = int(parameters["polynomial_order"]) if window % 2 == 0: raise ValueError("SMOOTH_WINDOW_MUST_BE_ODD") if polynomial_order >= window: raise ValueError("SMOOTH_POLYNOMIAL_ORDER_INVALID") if window > len(x): raise ValueError("SMOOTH_WINDOW_EXCEEDS_ROW_COUNT") half = window // 2 smoothed = np.empty_like(y) for index in range(len(x)): start = min(max(index - half, 0), len(x) - window) stop = start + window local_x = x[start:stop] - x[index] coefficients = np.polynomial.polynomial.polyfit( local_x, y[start:stop], polynomial_order ) smoothed[index] = coefficients[0] residual = y - smoothed rows = [ [float(a), float(b), float(c), float(d)] for a, b, c, d in zip(x, y, smoothed, residual, strict=True) ] return ["x", "y", "smoothed_y", "residual"], rows, { "operation": "smooth", "method": "savitzky_golay", "window": window, "polynomial_order": polynomial_order, "rmse": float(np.sqrt(np.mean(residual**2))), "boundary_handling": "shifted_full_window_local_polynomial", "row_count": len(rows), } def _differentiate( x: np.ndarray, y: np.ndarray, parameters: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: if len(x) < 3: raise ValueError("DIFFERENTIATE_REQUIRES_AT_LEAST_THREE_ROWS") order = int(parameters["order"]) derivative = y.copy() for _ in range(order): derivative = np.gradient(derivative, x, edge_order=2) rows = [ [float(a), float(b), float(c)] for a, b, c in zip(x, y, derivative, strict=True) ] return ["x", "y", f"derivative_order_{order}"], rows, { "operation": "differentiate", "order": order, "method": "numpy_gradient_nonuniform", "edge_order": 2, "row_count": len(rows), } def _cumulative_trapezoid(x: np.ndarray, y: np.ndarray) -> np.ndarray: cumulative = np.zeros_like(y) cumulative[1:] = np.cumsum(np.diff(x) * (y[:-1] + y[1:]) / 2) return cumulative def _integrate( x: np.ndarray, y: np.ndarray, parameters: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: cumulative = _cumulative_trapezoid(x, y) lower = float(parameters.get("from", x[0])) upper = float(parameters.get("to", x[-1])) if lower >= upper: raise ValueError("INTEGRATE_RANGE_INVALID") if lower < x[0] or upper > x[-1]: raise ValueError("INTEGRATE_RANGE_OUTSIDE_DATA") integral = float(np.interp(upper, x, cumulative) - np.interp(lower, x, cumulative)) rows = [ [float(a), float(b), float(c)] for a, b, c in zip(x, y, cumulative, strict=True) ] return ["x", "y", "cumulative_integral"], rows, { "operation": "integrate", "method": "trapezoid", "from": lower, "to": upper, "integral": integral, "full_integral": float(cumulative[-1]), "row_count": len(rows), } _T_CRITICAL_95 = ( 12.706, 4.303, 3.182, 2.776, 2.571, 2.447, 2.365, 2.306, 2.262, 2.228, 2.201, 2.179, 2.160, 2.145, 2.131, 2.120, 2.110, 2.101, 2.093, 2.086, 2.080, 2.074, 2.069, 2.064, 2.060, 2.056, 2.052, 2.048, 2.045, 2.042, ) def _t_critical_95(degrees_of_freedom: int) -> float: if degrees_of_freedom <= 0: raise ValueError("LINEAR_FIT_DEGREES_OF_FREEDOM_INVALID") if degrees_of_freedom <= len(_T_CRITICAL_95): return _T_CRITICAL_95[degrees_of_freedom - 1] anchors = ((30, 2.042), (40, 2.021), (60, 2.000), (120, 1.980), (10**9, 1.960)) for (left_df, left_value), (right_df, right_value) in pairwise(anchors): if degrees_of_freedom <= right_df: ratio = (degrees_of_freedom - left_df) / (right_df - left_df) return left_value + ratio * (right_value - left_value) return 1.960 def _linear_fit( x: np.ndarray, y: np.ndarray, parameters: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: include_intercept = bool(parameters["include_intercept"]) design = np.column_stack((x, np.ones_like(x))) if include_intercept else x[:, None] coefficients, _, rank, _ = np.linalg.lstsq(design, y, rcond=None) if rank != design.shape[1]: raise ValueError("LINEAR_FIT_DESIGN_RANK_DEFICIENT") fitted = design @ coefficients residual = y - fitted parameter_count = design.shape[1] degrees_of_freedom = len(x) - parameter_count if degrees_of_freedom <= 0: raise ValueError("LINEAR_FIT_INSUFFICIENT_ROWS") sum_squared_error = float(residual @ residual) mean_squared_error = sum_squared_error / degrees_of_freedom covariance = mean_squared_error * np.linalg.inv(design.T @ design) standard_errors = np.sqrt(np.diag(covariance)) slope = float(coefficients[0]) intercept = float(coefficients[1]) if include_intercept else 0.0 slope_se = float(standard_errors[0]) intercept_se = float(standard_errors[1]) if include_intercept else 0.0 t_critical = _t_critical_95(degrees_of_freedom) slope_ci = [slope - t_critical * slope_se, slope + t_critical * slope_se] intercept_ci = ( [intercept - t_critical * intercept_se, intercept + t_critical * intercept_se] if include_intercept else [0.0, 0.0] ) total_sum_squares = ( float(np.sum((y - np.mean(y)) ** 2)) if include_intercept else float(y @ y) ) r_squared = 1 - sum_squared_error / total_sum_squares if total_sum_squares else 1.0 rows = [ [float(a), float(b), float(c), float(d)] for a, b, c, d in zip(x, y, fitted, residual, strict=True) ] return ["x", "y", "fitted_y", "residual"], rows, { "operation": "linear_fit", "include_intercept": include_intercept, "confidence_level": 0.95, "slope": slope, "intercept": intercept, "slope_standard_error": slope_se, "intercept_standard_error": intercept_se, "slope_confidence_interval": slope_ci, "intercept_confidence_interval": intercept_ci, "t_critical": t_critical, "degrees_of_freedom": degrees_of_freedom, "r_squared": r_squared, "rmse": math.sqrt(mean_squared_error), "sum_squared_error": sum_squared_error, "row_count": len(rows), } def analyze( headers: list[str], rows: list[list[Any]], analysis: dict[str, Any] ) -> tuple[list[str], list[list[Any]], dict[str, Any]]: operation = analysis["type"] if operation not in OPERATIONS: raise ValueError("ANALYSIS_TYPE_NOT_IMPLEMENTED") if operation == "data_check": return _data_check(headers, rows, analysis) x, y = _xy_values(headers, rows, analysis["data"]) parameters = analysis["parameters"] functions = { "normalize": _normalize, "smooth": _smooth, "differentiate": _differentiate, "integrate": _integrate, "linear_fit": _linear_fit, } return functions[operation](x, y, parameters) def _origin_value(value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value return json.dumps(value, ensure_ascii=False, sort_keys=True) def _write_origin_project( project: Path, headers: list[str], rows: list[list[Any]], result_headers: list[str], result_rows: list[list[Any]], diagnostics: dict[str, Any], analysis: dict[str, Any], ) -> None: import originpro as op op.set_show(False) try: op.new() raw_sheet = op.new_sheet("w", lname="RawData") for index, header in enumerate(headers): raw_sheet.from_list( index, [row[index] if index < len(row) else None for row in rows], lname=header, ) result_sheet = op.new_sheet("w", lname="Result") for index, header in enumerate(result_headers): result_sheet.from_list( index, [row[index] if index < len(row) else None for row in result_rows], lname=header, ) diagnostic_sheet = op.new_sheet("w", lname="Diagnostics") diagnostic_items = list(diagnostics.items()) diagnostic_sheet.from_list(0, [item[0] for item in diagnostic_items], lname="metric") diagnostic_sheet.from_list( 1, [_origin_value(item[1]) for item in diagnostic_items], lname="value", ) if analysis["type"] == "linear_fit": binding = analysis["data"] x_index = _column_index(headers, binding["x"], "x") y_index = _column_index(headers, binding["y"], "y") fit = op.LinearFit() fit.set_data(raw_sheet, x_index, y_index) if not analysis["parameters"]["include_intercept"]: fit.fix_intercept(0) fit.report(3) op.save(str(project)) finally: if op.oext: op.exit() def _validate_project(path: Path) -> None: if not path.is_file() or path.stat().st_size < 128: raise RuntimeError("ORIGIN_PROJECT_INVALID") def run(job_dir: Path) -> list[dict[str, Any]]: job_dir = job_dir.resolve(strict=True) request_record = json.loads( (job_dir / "request" / "request.json").read_text(encoding="utf-8") ) request = request_record["request"] input_spec = request["inputs"][0] analysis = request["operation"]["analysis"] if analysis["data"]["input"] != input_spec["key"]: raise ValueError("ANALYSIS_INPUT_KEY_NOT_BOUND") input_path = _input_file(job_dir, input_spec["key"]) headers, rows = _read_rows(input_path, (input_spec.get("selector") or {}).get("sheet")) if not headers or len(set(headers)) != len(headers) or any(not item for item in headers): raise ValueError("INPUT_HEADERS_INVALID") result_headers, result_rows, diagnostics = analyze(headers, rows, analysis) output = job_dir / "output" output.mkdir(exist_ok=True) result_table = output / "result-table.csv" diagnostics_path = output / "diagnostics.json" analysis_spec_path = output / "analysis-spec.json" provenance_path = output / "provenance.json" project_path = output / "analysis.opju" _atomic_csv(result_table, result_headers, result_rows) _atomic_json(diagnostics_path, diagnostics) _atomic_json(analysis_spec_path, request) _write_origin_project( project_path, headers, rows, result_headers, result_rows, diagnostics, analysis, ) _validate_project(project_path) try: originpro_version = version("originpro") except PackageNotFoundError: originpro_version = "embedded" provenance = { "adapter_version": ADAPTER_VERSION, "algorithm_version": ALGORITHM_VERSION, "originpro_version": originpro_version, "numpy_version": np.__version__, "request_digest": request_record["request_digest"], "input": { "key": input_spec["key"], "filename": input_path.name, "sha256": _file_sha256(input_path), "sheet": (input_spec.get("selector") or {}).get("sheet"), }, "operation": analysis["type"], "origin_linear_fit_report": analysis["type"] == "linear_fit", } _atomic_json(provenance_path, provenance) artifacts = [ _manifest(project_path), _manifest(result_table), _manifest(diagnostics_path), _manifest(analysis_spec_path), _manifest(provenance_path), ] if any(item["key"] == "result_workbook" for item in request["outputs"]): workbook_path = output / "result-table.xlsx" _atomic_xlsx(workbook_path, result_headers, result_rows) artifacts.append(_manifest(workbook_path)) return artifacts def main() -> int: if sys.argv[1:] == ["--probe"]: return _probe() if len(sys.argv) != 2: print("[ERR] Usage: worker.py ", file=sys.stderr) return 2 job_dir = Path(sys.argv[1]) request_record: dict[str, Any] = {} try: request_record = json.loads( (job_dir / "request" / "request.json").read_text(encoding="utf-8") ) artifacts = run(job_dir) terminal = { "job_id": request_record["job_id"], "lease_id": request_record["lease_id"], "request_digest": request_record["request_digest"], "status": "succeeded", "error": {}, "artifact_manifest": artifacts, "terminal_at": datetime.now(timezone.utc).isoformat(), } _atomic_json(job_dir / "artifacts.json", artifacts) _atomic_json(job_dir / "terminal.json", terminal) print("[OK] Origin analysis job completed.") return 0 except Exception as exception: # noqa: BLE001 - terminalize every worker failure terminal = { "job_id": request_record.get("job_id", ""), "lease_id": request_record.get("lease_id", ""), "request_digest": request_record.get("request_digest", ""), "status": "failed", "error": {"code": type(exception).__name__, "detail": str(exception)[:500]}, "artifact_manifest": [], "terminal_at": datetime.now(timezone.utc).isoformat(), } _atomic_json(job_dir / "terminal.json", terminal) print(f"[ERR] {type(exception).__name__}: {exception}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())