Skip to content

config

Interfaces for Configuration values for programs - run_process.py - test_process.py

ProcessConfig dataclass

Configuration parameters for PROCESS runs

Source code in process/core/io/vary_run/config.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@dataclass
class ProcessConfig:
    """Configuration parameters for PROCESS runs"""

    wdir: Path = field(default_factory=Path.cwd)
    """Working directory"""
    or_in_dat: Path = Path("IN.DAT")
    """Original IN.DAT file"""
    niter: int = 10
    """(Maximum) number of iterations"""
    u_seed: int | None = None
    """User specified seed value for the random number generator"""
    factor: float = 1.5
    """Multiplication factor adjusting the range in which the original
iteration variables should get varied"""
    solver: str = "vmcon"
    """Solver to use"""
    comment: str = " "
    """additional comment to be written into README.txt"""
    _filename: Path | None = None
    """filename if supplied"""

    @classmethod
    def from_file(cls, filename: str | Path, solver: str = "vmcon"):
        if isinstance(filename, str):
            filename = Path(filename)

        if not filename.is_file():
            raise FileNotFoundError(f"Config file '{filename}' not found")

        wdir = (
            None if (buf := cls.get_attribute(filename, "wdir")) is None else Path(buf)
        )

        if wdir in {None, Path()}:
            wdir = filename.parent

        or_in_dat = (
            None
            if (buf := cls.get_attribute(filename, "ORIGINAL_IN_DAT")) is None
            else Path(buf)
        )
        or_in_dat = (
            cls.get_attribute(filename, "original_in_dat")
            if or_in_dat is None
            else or_in_dat
        )
        or_in_dat = Path("ref_IN.DAT") if or_in_dat is None else Path(or_in_dat)
        if not or_in_dat.is_file():
            or_in_dat = wdir / or_in_dat

        niter = 10 if (buf := cls.get_attribute(filename, "niter")) is None else int(buf)
        u_seed = (
            None
            if (buf := cls.get_attribute(filename, "seed")) in {None, "None"}
            else int(buf)
        )
        factor = (
            1.5 if (buf := cls.get_attribute(filename, "factor")) is None else float(buf)
        )

        comment = cls.get_comment(filename)
        if not comment:
            print(f"No comment in config file {filename}")

        return cls(
            wdir, or_in_dat, niter, u_seed, factor, solver, comment or " ", filename
        )

    @staticmethod
    def get_attribute(filename: Path, attributename: str):
        """Gets class attribute from configuration file"""
        with open(filename) as configfile:
            for line in configfile:
                condense = line.replace(" ", "")
                if (condense[0] != "*") and len(condense) > 1 and "=" in line:
                    varname = line[: line.find("=")]
                    varname = varname.replace(" ", "")
                    varname = varname.upper()
                    auxvar = line[line.find("=") + 1 :]
                    auxvar = auxvar.replace(" ", "")
                    auxvar = auxvar.rstrip()
                    if varname == attributename.upper() and auxvar == "":
                        return None
                    if varname == attributename.upper() and auxvar != "":
                        return auxvar

        return None

    @staticmethod
    def get_comment(filename):
        """Gets the comment line from the configuration file"""
        with open(filename) as configfile:
            for line in configfile:
                condense = line.replace(" ", "")
                condense = condense.rstrip()
                lcase = condense.lower()
                if len(condense) > 0 and (condense[0] != "*") and lcase[:7] == "comment":
                    return line[line.find("=") + 1 :]
        return ""

    def __post_init__(self):
        self._current_iteration = 0
        self._base_input = "{}_IN.DAT"
        self._base_output = "{}_MFILE.DAT"

        if not isinstance(self.or_in_dat, Path) or not self.or_in_dat.is_file():
            raise FileNotFoundError(
                f"Error: {self.or_in_dat} does not exist! Create file or modify config file!",
            )

    def __iter__(self):
        return self

    def __next__(self):
        _neqns, itervars = get_neqns_itervars(wdir=self.wdir)
        lbs, ubs = get_variable_range(itervars, self.factor, self.wdir)
        indat, mfile = self.infile, self.outfile
        self._current_iteration += 1
        if self._current_iteration >= self.niter:
            self.error_status2readme(mfile=mfile)
            raise StopIteration
        self.run_process(self.wdir / indat, self.solver)
        check_input_error(wdir=self.wdir, mfile=mfile)

        return indat, mfile, itervars, lbs, ubs

    @property
    def infile(self):
        return self._base_input.format(self._current_iteration)

    @property
    def outfile(self):
        return self._base_output.format(self._current_iteration)

    def echo(self):
        """Echos the attributes of the class"""
        if self.wdir != Path.cwd():
            print(f"Working directory:   {self.wdir}")
        print(f"Original IN.DAT:     {self.or_in_dat}")
        print(f"Number of iterations {self.niter}")

        if self.u_seed is not None:
            print(f"random seed          {self.u_seed}")
        print(f"variable range factor {self.factor}")
        if self._filename is not None:
            print(f"Config file          {self._filename}")
        if self.comment != "":
            print(f"Comment  {self.comment}")

    def prepare_wdir(self):
        """Prepares the working directory"""
        if not self.wdir.is_dir():
            self.wdir.mkdir(exist_ok=True, parents=True)

        copy(self.or_in_dat, self.wdir / "0_IN.DAT")

        if self._filename is not None:
            with suppress(SameFileError):
                copy(self._filename, self.wdir)

        for file in (
            "OUT.DAT",
            "MFILE.DAT",
            "README.txt",
            "SolverTest.out",
            "process.log",
            "uncertainties.nc",
            "time.info",
        ):
            Path(self.wdir, file).unlink(missing_ok=True)
        for f in self.wdir.glob("*.pdf"):
            f.unlink()

        os.chdir(self.wdir)

    def create_readme(self):
        """Creates README.txt containing comment"""
        if self.comment != "":
            Path(self.wdir, "README.txt").write_text(self.comment)

    def error_status2readme(self, mfile="MFILE.DAT"):
        """Appends PROCESS outcome to README.txt"""
        m_file = MFile(filename=self.wdir / mfile)

        error_status = (
            f"Error status: {m_file.data['error_status'].get_scan(-1)}  "
            f"Error ID: {m_file.data['error_id'].get_scan(-1)}\n"
        )

        if self.comment != "":
            with open(Path(self.wdir, "README.txt"), "a") as readme:
                readme.write(error_status)
        else:
            Path(self.wdir, "README.txt").write_text(error_status)

    def modify_in_dat(self):
        """Modifies the original IN.DAT file"""

    def setup(self):
        """Sets up the program for running"""
        self.echo()

        self.prepare_wdir()

        self.create_readme()

        self.modify_in_dat()

        check_in_dat("0_IN.DAT")

        self.generator = default_rng(seed=self.u_seed)

    def run_process(self, input_path: Path, solver: str = "vmcon"):
        """Perform a single run of PROCESS, catching any errors.

        Parameters
        ----------
        input_path :
            the input file to run on
        solver :
            which solver to use, as specified in solver.py, defaults to "vmcon"
        """
        # TODO should call SingleRun directly...at least this is not a subprocess!
        from process.main import process_cli  # noqa:PLC0415

        print("PROCESS run started ...", end="")

        try:
            # Run process on an IN.DAT file
            process_cli.main(
                args=["-i", str(input_path), "--solver", solver], standalone_mode=False
            )
        except SystemExit as err:
            if err.code != 0:
                # Process has exited with a non-zero exit code.
                # Catch this exception to allow execution to continue without exiting
                logger.exception(
                    "There was a problem with the PROCESS execution!",
                )
        except (KeyboardInterrupt, click.exceptions.Abort):
            raise KeyboardInterrupt from None
        except Exception:
            logger.exception(
                "There was a problem with the PROCESS execution!",
            )

        print("finished.")

wdir = field(default_factory=(Path.cwd)) class-attribute instance-attribute

Working directory

or_in_dat = Path('IN.DAT') class-attribute instance-attribute

Original IN.DAT file

niter = 10 class-attribute instance-attribute

(Maximum) number of iterations

u_seed = None class-attribute instance-attribute

User specified seed value for the random number generator

factor = 1.5 class-attribute instance-attribute

Multiplication factor adjusting the range in which the original iteration variables should get varied

solver = 'vmcon' class-attribute instance-attribute

Solver to use

comment = ' ' class-attribute instance-attribute

additional comment to be written into README.txt

infile property

outfile property

from_file(filename, solver='vmcon') classmethod

Source code in process/core/io/vary_run/config.py 
 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
 96
 97
 98
 99
100
101
@classmethod
def from_file(cls, filename: str | Path, solver: str = "vmcon"):
    if isinstance(filename, str):
        filename = Path(filename)

    if not filename.is_file():
        raise FileNotFoundError(f"Config file '{filename}' not found")

    wdir = (
        None if (buf := cls.get_attribute(filename, "wdir")) is None else Path(buf)
    )

    if wdir in {None, Path()}:
        wdir = filename.parent

    or_in_dat = (
        None
        if (buf := cls.get_attribute(filename, "ORIGINAL_IN_DAT")) is None
        else Path(buf)
    )
    or_in_dat = (
        cls.get_attribute(filename, "original_in_dat")
        if or_in_dat is None
        else or_in_dat
    )
    or_in_dat = Path("ref_IN.DAT") if or_in_dat is None else Path(or_in_dat)
    if not or_in_dat.is_file():
        or_in_dat = wdir / or_in_dat

    niter = 10 if (buf := cls.get_attribute(filename, "niter")) is None else int(buf)
    u_seed = (
        None
        if (buf := cls.get_attribute(filename, "seed")) in {None, "None"}
        else int(buf)
    )
    factor = (
        1.5 if (buf := cls.get_attribute(filename, "factor")) is None else float(buf)
    )

    comment = cls.get_comment(filename)
    if not comment:
        print(f"No comment in config file {filename}")

    return cls(
        wdir, or_in_dat, niter, u_seed, factor, solver, comment or " ", filename
    )

get_attribute(filename, attributename) staticmethod

Gets class attribute from configuration file

Source code in process/core/io/vary_run/config.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@staticmethod
def get_attribute(filename: Path, attributename: str):
    """Gets class attribute from configuration file"""
    with open(filename) as configfile:
        for line in configfile:
            condense = line.replace(" ", "")
            if (condense[0] != "*") and len(condense) > 1 and "=" in line:
                varname = line[: line.find("=")]
                varname = varname.replace(" ", "")
                varname = varname.upper()
                auxvar = line[line.find("=") + 1 :]
                auxvar = auxvar.replace(" ", "")
                auxvar = auxvar.rstrip()
                if varname == attributename.upper() and auxvar == "":
                    return None
                if varname == attributename.upper() and auxvar != "":
                    return auxvar

    return None

get_comment(filename) staticmethod

Gets the comment line from the configuration file

Source code in process/core/io/vary_run/config.py 
123
124
125
126
127
128
129
130
131
132
133
@staticmethod
def get_comment(filename):
    """Gets the comment line from the configuration file"""
    with open(filename) as configfile:
        for line in configfile:
            condense = line.replace(" ", "")
            condense = condense.rstrip()
            lcase = condense.lower()
            if len(condense) > 0 and (condense[0] != "*") and lcase[:7] == "comment":
                return line[line.find("=") + 1 :]
    return ""

echo()

Echos the attributes of the class

Source code in process/core/io/vary_run/config.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def echo(self):
    """Echos the attributes of the class"""
    if self.wdir != Path.cwd():
        print(f"Working directory:   {self.wdir}")
    print(f"Original IN.DAT:     {self.or_in_dat}")
    print(f"Number of iterations {self.niter}")

    if self.u_seed is not None:
        print(f"random seed          {self.u_seed}")
    print(f"variable range factor {self.factor}")
    if self._filename is not None:
        print(f"Config file          {self._filename}")
    if self.comment != "":
        print(f"Comment  {self.comment}")

prepare_wdir()

Prepares the working directory

Source code in process/core/io/vary_run/config.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def prepare_wdir(self):
    """Prepares the working directory"""
    if not self.wdir.is_dir():
        self.wdir.mkdir(exist_ok=True, parents=True)

    copy(self.or_in_dat, self.wdir / "0_IN.DAT")

    if self._filename is not None:
        with suppress(SameFileError):
            copy(self._filename, self.wdir)

    for file in (
        "OUT.DAT",
        "MFILE.DAT",
        "README.txt",
        "SolverTest.out",
        "process.log",
        "uncertainties.nc",
        "time.info",
    ):
        Path(self.wdir, file).unlink(missing_ok=True)
    for f in self.wdir.glob("*.pdf"):
        f.unlink()

    os.chdir(self.wdir)

create_readme()

Creates README.txt containing comment

Source code in process/core/io/vary_run/config.py 
210
211
212
213
def create_readme(self):
    """Creates README.txt containing comment"""
    if self.comment != "":
        Path(self.wdir, "README.txt").write_text(self.comment)

error_status2readme(mfile='MFILE.DAT')

Appends PROCESS outcome to README.txt

Source code in process/core/io/vary_run/config.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def error_status2readme(self, mfile="MFILE.DAT"):
    """Appends PROCESS outcome to README.txt"""
    m_file = MFile(filename=self.wdir / mfile)

    error_status = (
        f"Error status: {m_file.data['error_status'].get_scan(-1)}  "
        f"Error ID: {m_file.data['error_id'].get_scan(-1)}\n"
    )

    if self.comment != "":
        with open(Path(self.wdir, "README.txt"), "a") as readme:
            readme.write(error_status)
    else:
        Path(self.wdir, "README.txt").write_text(error_status)

modify_in_dat()

Modifies the original IN.DAT file

Source code in process/core/io/vary_run/config.py 
230
231
def modify_in_dat(self):
    """Modifies the original IN.DAT file"""

setup()

Sets up the program for running

Source code in process/core/io/vary_run/config.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
def setup(self):
    """Sets up the program for running"""
    self.echo()

    self.prepare_wdir()

    self.create_readme()

    self.modify_in_dat()

    check_in_dat("0_IN.DAT")

    self.generator = default_rng(seed=self.u_seed)

run_process(input_path, solver='vmcon')

Perform a single run of PROCESS, catching any errors.

Parameters:

Name Type Description Default
input_path Path

the input file to run on

 required
solver str

which solver to use, as specified in solver.py, defaults to "vmcon"

 'vmcon'
Source code in process/core/io/vary_run/config.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def run_process(self, input_path: Path, solver: str = "vmcon"):
    """Perform a single run of PROCESS, catching any errors.

    Parameters
    ----------
    input_path :
        the input file to run on
    solver :
        which solver to use, as specified in solver.py, defaults to "vmcon"
    """
    # TODO should call SingleRun directly...at least this is not a subprocess!
    from process.main import process_cli  # noqa:PLC0415

    print("PROCESS run started ...", end="")

    try:
        # Run process on an IN.DAT file
        process_cli.main(
            args=["-i", str(input_path), "--solver", solver], standalone_mode=False
        )
    except SystemExit as err:
        if err.code != 0:
            # Process has exited with a non-zero exit code.
            # Catch this exception to allow execution to continue without exiting
            logger.exception(
                "There was a problem with the PROCESS execution!",
            )
    except (KeyboardInterrupt, click.exceptions.Abort):
        raise KeyboardInterrupt from None
    except Exception:
        logger.exception(
            "There was a problem with the PROCESS execution!",
        )

    print("finished.")

RunProcessConfig dataclass

Bases: ProcessConfig

Configuration parameters of the run_process.py program

Source code in process/core/io/vary_run/config.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
@dataclass
class RunProcessConfig(ProcessConfig):
    """Configuration parameters of the run_process.py program"""

    no_allowed_unfeasible: int = 0
    """the number of allowed unfeasible points in a sweep"""
    create_itervar_diff: bool = False
    """boolean to indicate the creation of a summary file of the iteration variable values at each stage"""
    dictvar: dict[str, str] = field(default_factory=dict)
    """Dictionary mapping variable name to new value"""
    del_var: list[str] = field(default_factory=list)
    """List of variables to be deleted from IN.DAT"""
    add_ixc: list[str] = field(default_factory=list)
    """ List of iteration variables to be added to IN.DAT"""
    del_ixc: list[str] = field(default_factory=list)
    """List of iteration variables to be deleted from IN.DAT"""
    add_icc: list[str] = field(default_factory=list)
    """List of constrained equations to be added to IN.DAT"""
    del_icc: list[str] = field(default_factory=list)
    """List of constrained equations to be deleted from IN.DAT"""

    @classmethod
    def from_file(cls, filename: str | Path = "run_process.conf", solver: str = "vmcon"):
        self = super().from_file(filename, solver)

        no_allowed_unfeasible = (
            0
            if (buf := cls.get_attribute(self._filename, "no_allowed_unfeasible"))
            is None
            else int(buf)
        )

        buf = cls.get_attribute(self._filename, "create_itervar_diff")

        create_itervar_diff = False
        if buf is not None:
            if buf.lower() in {"true", "y", "yes"}:
                create_itervar_diff = True
            elif buf.lower() in {"false", "n", "no"}:
                create_itervar_diff = False
            else:
                logger.warning("Value for create_itervar_diff is not defined!")

        add_ixc = cls.get_attribute_csv_list(self._filename, "add_ixc")
        del_ixc = cls.get_attribute_csv_list(self._filename, "del_ixc")
        add_icc = cls.get_attribute_csv_list(self._filename, "add_icc")
        del_icc = cls.get_attribute_csv_list(self._filename, "del_icc")

        return cls(
            self.wdir,
            self.or_in_dat,
            self.niter,
            self.u_seed,
            self.factor,
            self.solver,
            self.comment,
            self._filename,
            no_allowed_unfeasible,
            create_itervar_diff,
            cls.get_dictvar(self._filename),
            cls.get_del_var(self._filename),
            add_ixc,
            del_ixc,
            add_icc,
            del_icc,
        )

    @staticmethod
    def get_attribute_csv_list(filename, attributename):
        """Get class attribute list from configuration file
        expects comma separated values
        """
        attribute_list = []

        with open(filename) as configfile:
            for line in configfile:
                condense = line.replace(" ", "").rstrip()
                lcase = condense.lower()
                if (
                    (len(condense) > 0)
                    and (condense[0] != "*")
                    and (attributename == lcase[: len(attributename)])
                ):
                    buf = condense[condense.find("=") + 1 :].split(",")
                    if buf[-1] == "":  # if last value has ended on comma
                        buf = buf[:-1]
                    attribute_list += buf
        return attribute_list

    @staticmethod
    def get_del_var(filename):
        """Sets the del_var attribute from the config file"""
        del_var = []
        with open(filename) as configfile:
            for line in configfile:
                condense = line.replace(" ", "")
                condense = condense.rstrip()
                lcase = condense.lower()
                if (
                    (len(condense) > 0)
                    and (condense[0] != "*")
                    and (lcase[:8] == "del_var_")
                    and (len(condense) > 8)
                ):
                    del_var += [condense[8:]]
        return del_var

    @staticmethod
    def get_dictvar(filename):
        """Sets the dictvar attribute from config file"""
        dictvar = {}
        with open(filename) as configfile:
            for line in configfile:
                condense = line.replace(" ", "")
                condense = condense.rstrip()
                lcase = condense.lower()
                if len(condense) > 0 and (condense[0] != "*") and "=" in lcase:
                    varname = lcase[: lcase.find("=")]
                    auxvar = condense[condense.find("=") + 1 :]
                    if varname[:4] == "var_" and auxvar != "":
                        dictvar[varname[4:]] = auxvar
        return dictvar

    def __next__(self):
        indat, mfile, itervars, lbs, ubs = super().__next__()

        if not process_stopped(wdir=self.wdir, mfile=mfile):
            no_unfeasible = no_unfeasible_mfile(self.wdir, mfile)
            if no_unfeasible <= self.no_allowed_unfeasible:
                if no_unfeasible > 0:
                    logger.warning(
                        "Non feasible point(s) in sweep, But finished anyway! %s ",
                        no_unfeasible,
                    )
                if process_warnings(self.wdir, mfile):
                    print(
                        "\nThere were warnings in the final PROCESS run. "
                        "Please check the log file!\n"
                    )
                    # This means success: feasible solution found
                    raise StopIteration
                logger.warning(
                    "%s non-feasible point(s) in sweep! Rerunning!", no_unfeasible
                )
        else:
            print("PROCESS has stopped without finishing!")

        return indat, mfile, itervars, lbs, ubs

    def echo(self):
        """Echos the values of the current class"""
        super().echo()

        print(f"no. allowed UNFEASIBLE points {self.no_allowed_unfeasible:d}")
        if self.create_itervar_diff:
            print("Set to create a summary file of the iteration variable values!")

        for n, v in (
            ("add_ixc", self.add_ixc),
            ("del_ixc", self.del_ixc),
            ("add_icc", self.add_icc),
            ("del_icc", self.del_icc),
        ):
            if v != []:
                print(n, v)

        for key, value in self.dictvar.items():
            print(f"set {key}  to {value}")
        if self.del_var != []:
            print("del_var", self.del_var)

    def modify_in_dat(self):
        """Modifies IN.DAT using the configuration parameters"""
        # Need to keep this order!
        # If bounds are modified in vars, but ixc is newly added,
        # bounds do not get put into IN.DAT. Hence, vars needs to be modified
        # first.
        self.modify_ixc()
        self.modify_icc()
        self.modify_vars()

    def modify_vars(self):
        """Modifies IN.DAT by adding, deleting and modifiying variables"""
        in_dat = InDat(filename="0_IN.DAT")

        # add and modify variables
        for key in self.dictvar:
            set_variable_in_indat(in_dat, key, self.dictvar[key])

        # delete variables
        for key_ in self.del_var:
            key = key_.lower()
            if "bound" in key:
                number = (key.split("("))[1].split(")")[0]
                if "boundu" in key:
                    in_dat.remove_bound(number, "u")
                else:
                    in_dat.remove_bound(number, "l")
            else:
                in_dat.remove_parameter(key)

        in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")

    def modify_ixc(self):
        """Modifies the array of iteration variables in IN.DAT"""
        # check that there is no variable in both lists
        if set(self.add_ixc).intersection(self.del_ixc) != set():
            logger.error(
                "You are trying to add and delete the same variable from ixc!",
                file=sys.stderr,
            )
            sys.exit()

        in_dat = InDat(filename="0_IN.DAT")

        for iter_var in self.add_ixc:
            in_dat.add_iteration_variable(int(iter_var))

        for iter_var in self.del_ixc:
            in_dat.remove_iteration_variable(int(iter_var))

        in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")

    def modify_icc(self):
        """Modifies the array of constraint equations in IN.DAT"""
        # check that there is no variable in both lists
        if set(self.add_icc).intersection(self.del_icc) != set():
            logger.error(
                "You are trying to add and delete the same variable from icc!",
                file=sys.stderr,
            )
            sys.exit()

        in_dat = InDat(filename="0_IN.DAT")

        for constr in self.add_icc:
            in_dat.add_constraint_equation(int(constr))

        for constr in self.del_icc:
            in_dat.remove_constraint_equation(int(constr))

        in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")

no_allowed_unfeasible = 0 class-attribute instance-attribute

the number of allowed unfeasible points in a sweep

create_itervar_diff = False class-attribute instance-attribute

boolean to indicate the creation of a summary file of the iteration variable values at each stage

dictvar = field(default_factory=dict) class-attribute instance-attribute

Dictionary mapping variable name to new value

del_var = field(default_factory=list) class-attribute instance-attribute

List of variables to be deleted from IN.DAT

add_ixc = field(default_factory=list) class-attribute instance-attribute

List of iteration variables to be added to IN.DAT

del_ixc = field(default_factory=list) class-attribute instance-attribute

List of iteration variables to be deleted from IN.DAT

add_icc = field(default_factory=list) class-attribute instance-attribute

List of constrained equations to be added to IN.DAT

del_icc = field(default_factory=list) class-attribute instance-attribute

List of constrained equations to be deleted from IN.DAT

from_file(filename='run_process.conf', solver='vmcon') classmethod

Source code in process/core/io/vary_run/config.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
@classmethod
def from_file(cls, filename: str | Path = "run_process.conf", solver: str = "vmcon"):
    self = super().from_file(filename, solver)

    no_allowed_unfeasible = (
        0
        if (buf := cls.get_attribute(self._filename, "no_allowed_unfeasible"))
        is None
        else int(buf)
    )

    buf = cls.get_attribute(self._filename, "create_itervar_diff")

    create_itervar_diff = False
    if buf is not None:
        if buf.lower() in {"true", "y", "yes"}:
            create_itervar_diff = True
        elif buf.lower() in {"false", "n", "no"}:
            create_itervar_diff = False
        else:
            logger.warning("Value for create_itervar_diff is not defined!")

    add_ixc = cls.get_attribute_csv_list(self._filename, "add_ixc")
    del_ixc = cls.get_attribute_csv_list(self._filename, "del_ixc")
    add_icc = cls.get_attribute_csv_list(self._filename, "add_icc")
    del_icc = cls.get_attribute_csv_list(self._filename, "del_icc")

    return cls(
        self.wdir,
        self.or_in_dat,
        self.niter,
        self.u_seed,
        self.factor,
        self.solver,
        self.comment,
        self._filename,
        no_allowed_unfeasible,
        create_itervar_diff,
        cls.get_dictvar(self._filename),
        cls.get_del_var(self._filename),
        add_ixc,
        del_ixc,
        add_icc,
        del_icc,
    )

get_attribute_csv_list(filename, attributename) staticmethod

Get class attribute list from configuration file expects comma separated values

Source code in process/core/io/vary_run/config.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
@staticmethod
def get_attribute_csv_list(filename, attributename):
    """Get class attribute list from configuration file
    expects comma separated values
    """
    attribute_list = []

    with open(filename) as configfile:
        for line in configfile:
            condense = line.replace(" ", "").rstrip()
            lcase = condense.lower()
            if (
                (len(condense) > 0)
                and (condense[0] != "*")
                and (attributename == lcase[: len(attributename)])
            ):
                buf = condense[condense.find("=") + 1 :].split(",")
                if buf[-1] == "":  # if last value has ended on comma
                    buf = buf[:-1]
                attribute_list += buf
    return attribute_list

get_del_var(filename) staticmethod

Sets the del_var attribute from the config file

Source code in process/core/io/vary_run/config.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@staticmethod
def get_del_var(filename):
    """Sets the del_var attribute from the config file"""
    del_var = []
    with open(filename) as configfile:
        for line in configfile:
            condense = line.replace(" ", "")
            condense = condense.rstrip()
            lcase = condense.lower()
            if (
                (len(condense) > 0)
                and (condense[0] != "*")
                and (lcase[:8] == "del_var_")
                and (len(condense) > 8)
            ):
                del_var += [condense[8:]]
    return del_var

get_dictvar(filename) staticmethod

Sets the dictvar attribute from config file

Source code in process/core/io/vary_run/config.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
@staticmethod
def get_dictvar(filename):
    """Sets the dictvar attribute from config file"""
    dictvar = {}
    with open(filename) as configfile:
        for line in configfile:
            condense = line.replace(" ", "")
            condense = condense.rstrip()
            lcase = condense.lower()
            if len(condense) > 0 and (condense[0] != "*") and "=" in lcase:
                varname = lcase[: lcase.find("=")]
                auxvar = condense[condense.find("=") + 1 :]
                if varname[:4] == "var_" and auxvar != "":
                    dictvar[varname[4:]] = auxvar
    return dictvar

echo()

Echos the values of the current class

Source code in process/core/io/vary_run/config.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def echo(self):
    """Echos the values of the current class"""
    super().echo()

    print(f"no. allowed UNFEASIBLE points {self.no_allowed_unfeasible:d}")
    if self.create_itervar_diff:
        print("Set to create a summary file of the iteration variable values!")

    for n, v in (
        ("add_ixc", self.add_ixc),
        ("del_ixc", self.del_ixc),
        ("add_icc", self.add_icc),
        ("del_icc", self.del_icc),
    ):
        if v != []:
            print(n, v)

    for key, value in self.dictvar.items():
        print(f"set {key}  to {value}")
    if self.del_var != []:
        print("del_var", self.del_var)

modify_in_dat()

Modifies IN.DAT using the configuration parameters

Source code in process/core/io/vary_run/config.py 
455
456
457
458
459
460
461
462
463
def modify_in_dat(self):
    """Modifies IN.DAT using the configuration parameters"""
    # Need to keep this order!
    # If bounds are modified in vars, but ixc is newly added,
    # bounds do not get put into IN.DAT. Hence, vars needs to be modified
    # first.
    self.modify_ixc()
    self.modify_icc()
    self.modify_vars()

modify_vars()

Modifies IN.DAT by adding, deleting and modifiying variables

Source code in process/core/io/vary_run/config.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def modify_vars(self):
    """Modifies IN.DAT by adding, deleting and modifiying variables"""
    in_dat = InDat(filename="0_IN.DAT")

    # add and modify variables
    for key in self.dictvar:
        set_variable_in_indat(in_dat, key, self.dictvar[key])

    # delete variables
    for key_ in self.del_var:
        key = key_.lower()
        if "bound" in key:
            number = (key.split("("))[1].split(")")[0]
            if "boundu" in key:
                in_dat.remove_bound(number, "u")
            else:
                in_dat.remove_bound(number, "l")
        else:
            in_dat.remove_parameter(key)

    in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")

modify_ixc()

Modifies the array of iteration variables in IN.DAT

Source code in process/core/io/vary_run/config.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
def modify_ixc(self):
    """Modifies the array of iteration variables in IN.DAT"""
    # check that there is no variable in both lists
    if set(self.add_ixc).intersection(self.del_ixc) != set():
        logger.error(
            "You are trying to add and delete the same variable from ixc!",
            file=sys.stderr,
        )
        sys.exit()

    in_dat = InDat(filename="0_IN.DAT")

    for iter_var in self.add_ixc:
        in_dat.add_iteration_variable(int(iter_var))

    for iter_var in self.del_ixc:
        in_dat.remove_iteration_variable(int(iter_var))

    in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")

modify_icc()

Modifies the array of constraint equations in IN.DAT

Source code in process/core/io/vary_run/config.py 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
def modify_icc(self):
    """Modifies the array of constraint equations in IN.DAT"""
    # check that there is no variable in both lists
    if set(self.add_icc).intersection(self.del_icc) != set():
        logger.error(
            "You are trying to add and delete the same variable from icc!",
            file=sys.stderr,
        )
        sys.exit()

    in_dat = InDat(filename="0_IN.DAT")

    for constr in self.add_icc:
        in_dat.add_constraint_equation(int(constr))

    for constr in self.del_icc:
        in_dat.remove_constraint_equation(int(constr))

    in_dat.write_in_dat(output_filename=self.wdir / "0_IN.DAT")