Source code for ewoksid16a.tasks.fluo.patch_config

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Tue Feb 10 20:40:04 2026

@author: blissadm
"""

import os
import time

from ewokscore import Task
from PyMca5.PyMcaIO import ConfigDict

DEFAULTS = {
    "batchconfig_ext": ".batch.cfg",
    "batch_force": {
        "fit.stripflag": 0,
        "fit.fitweight": 0,
        "fit.linearfitflag": 1,
    },
    "retryPeriod": 3,
    "retryN": 1200,
    "force_overwrite": False,
    "output_root_group": None,
}




class PatchConfigFile(
    Task,
    input_names=[
        "bliss_scan_uri",
        "detector_name",
        "config_file",
        "output_root_uri",
    ],
    optional_input_names=[
        "batchconfig_ext",
        "batch_force",
        "retryPeriod",
        "retryN",
        "force_overwrite",
        "output_root_group",
    ],
    output_names=[
        "bliss_scan_uri",
        "detector_name",
        "output_root_uri",
        "config_file",
        "output_root_group",
    ],
):


    def run(self):

        pars = {**DEFAULTS, **self.get_input_values()}

        self.outputs.bliss_scan_uri = pars["bliss_scan_uri"]
        self.outputs.output_root_uri = pars["output_root_uri"]
        self.outputs.detector_name = pars["detector_name"]
        self.outputs.output_root_group = pars["output_root_group"]
        config_file = pars["config_file"]
        batch_ext = pars["batchconfig_ext"]
        batch_force = pars["batch_force"]
        force_overwrite = pars["force_overwrite"]

        retryPeriod = pars["retryPeriod"]
        retryN = pars["retryN"]

        for i in range(retryN):
            if os.path.isfile(config_file):
                break
            time.sleep(retryPeriod)
        else:
            raise RuntimeError(f"Config file {config_file} not found!")

        root, ext = os.path.splitext(config_file)

        batch_savefile = root + batch_ext

        if force_overwrite or not os.path.isfile(batch_savefile):
            cfg = ConfigDict.ConfigDict(filelist=config_file)

            for k in batch_force:
                _c = cfg
                kk = k.split(".")
                for _k in kk[:-1]:
                    _c = _c[_k]

                _c[kk[-1]] = batch_force[k]

            cfg.write(batch_savefile)

        self.outputs.config_file = batch_savefile