utils.py 19.4 KB
Newer Older
1
import argparse
2
3
4
import collections
import fnmatch
import functools
5
import hashlib
6
import importlib.util
7
import inspect
8
import json
9
10
11
import logging
import os
import re
12
from dataclasses import asdict, is_dataclass
13
from itertools import islice
14
from pathlib import Path
15
from typing import Any, Callable, Dict, Generator, List, Optional, Tuple
16

Lintang Sutawika's avatar
Lintang Sutawika committed
17
import numpy as np
18
import yaml
19
from jinja2 import BaseLoader, Environment, StrictUndefined
sdtblck's avatar
sdtblck committed
20

lintangsutawika's avatar
lintangsutawika committed
21

22
SPACING = " " * 47
sdtblck's avatar
sdtblck committed
23

24
25
26
27
28
HIGHER_IS_BETTER_SYMBOLS = {
    True: "↑",
    False: "↓",
}

sdtblck's avatar
sdtblck committed
29

Lintang Sutawika's avatar
Lintang Sutawika committed
30
31
def setup_logging(verbosity=logging.INFO):
    # Configure the root logger
Baber Abbasi's avatar
Baber Abbasi committed
32
33
34
35
36
37
38
39
40
41
42
    class CustomFormatter(logging.Formatter):
        def format(self, record):
            if record.name.startswith("lm_eval."):
                record.name = record.name[len("lm_eval.") :]
            return super().format(record)

    formatter = CustomFormatter(
        "%(asctime)s %(levelname)-8s [%(name)s:%(lineno)d] %(message)s",
        datefmt="%Y-%m-%d:%H:%M:%S",
    )

Lintang Sutawika's avatar
Lintang Sutawika committed
43
44
45
46
47
48
49
50
51
52
53
    log_level = os.environ.get("LOGLEVEL", verbosity) or verbosity

    level_map = {
        "DEBUG": logging.DEBUG,
        "INFO": logging.INFO,
        "WARNING": logging.WARNING,
        "ERROR": logging.ERROR,
        "CRITICAL": logging.CRITICAL,
    }

    log_level = level_map.get(str(log_level).upper(), logging.INFO)
Baber Abbasi's avatar
Baber Abbasi committed
54

Lintang Sutawika's avatar
Lintang Sutawika committed
55
    if not logging.root.handlers:
Baber Abbasi's avatar
Baber Abbasi committed
56
57
58
59
60
61
62
        handler = logging.StreamHandler()
        handler.setFormatter(formatter)

        root_logger = logging.getLogger()
        root_logger.addHandler(handler)
        root_logger.setLevel(log_level)

Lintang Sutawika's avatar
Lintang Sutawika committed
63
64
65
66
67
68
69
70
        if log_level == logging.DEBUG:
            third_party_loggers = ["urllib3", "filelock", "fsspec"]
            for logger_name in third_party_loggers:
                logging.getLogger(logger_name).setLevel(logging.INFO)
    else:
        logging.getLogger().setLevel(log_level)


71
72
73
74
def hash_string(string: str) -> str:
    return hashlib.sha256(string.encode("utf-8")).hexdigest()


75
76
77
78
79
80
81
82
83
84
85
86
def escaped_split(text, sep_char, maxsplit=-1):
    """Split text into a list on occurrences of the given separation
    character `sep_char`. The separation character may be escaped by a
    backslash to avoid splitting at that location.

    The separation character must be a string of size 1.

    If `maxsplit` is given, at most `maxsplit` splits are done (thus,
    the list will have at most `maxsplit + 1` elements). If `maxsplit`
    is not specified or less than 0, then there is no limit on the
    number of splits (all possible splits are made).
    """
Baber Abbasi's avatar
Baber Abbasi committed
87
88
89
    assert len(sep_char) == 1, (
        "separation string must be a single character for escaped splitting"
    )
90
91
92
93
94
95
96
97

    if maxsplit == 0:
        return text
    maxsplit = max(0, maxsplit)

    return re.split(r"(?<!\\)" + sep_char, text, maxsplit)


haileyschoelkopf's avatar
haileyschoelkopf committed
98
99
100
101
102
def handle_arg_string(arg):
    if arg.lower() == "true":
        return True
    elif arg.lower() == "false":
        return False
103
104
105
106
107
108
    elif arg.isnumeric():
        return int(arg)
    try:
        return float(arg)
    except ValueError:
        return arg
haileyschoelkopf's avatar
haileyschoelkopf committed
109
110


111
112
113
114
115
116
117
118
119
def handle_non_serializable(o):
    if isinstance(o, np.int64) or isinstance(o, np.int32):
        return int(o)
    elif isinstance(o, set):
        return list(o)
    else:
        return str(o)


120
121
122
123
124
125
126
127
128
129
130
131
def sanitize_list(sub):
    """
    Takes possible nested list and recursively converts all inner component to strings
    """
    if isinstance(sub, list):
        return [sanitize_list(item) for item in sub]
    if isinstance(sub, tuple):
        return tuple(sanitize_list(item) for item in sub)
    else:
        return str(sub)


Baber Abbasi's avatar
Baber Abbasi committed
132
def simple_parse_args_string(args_string: Optional[str]) -> dict:
Jason Phang's avatar
gpt3  
Jason Phang committed
133
134
135
136
137
    """
    Parses something like
        args1=val1,arg2=val2
    Into a dictionary
    """
Baber Abbasi's avatar
Baber Abbasi committed
138
139
    if args_string is None:
        return {}
Jason Phang's avatar
Jason Phang committed
140
    args_string = args_string.strip()
Jason Phang's avatar
gpt3  
Jason Phang committed
141
142
    if not args_string:
        return {}
143
    arg_list = [arg for arg in args_string.split(",") if arg]
haileyschoelkopf's avatar
haileyschoelkopf committed
144
    args_dict = {
145
146
        kv[0]: handle_arg_string("=".join(kv[1:]))
        for kv in [arg.split("=") for arg in arg_list]
haileyschoelkopf's avatar
haileyschoelkopf committed
147
    }
Jason Phang's avatar
gpt3  
Jason Phang committed
148
    return args_dict
Leo Gao's avatar
Leo Gao committed
149

Fabrizio Milo's avatar
Fabrizio Milo committed
150

Leo Gao's avatar
Leo Gao committed
151
152
def join_iters(iters):
    for iter in iters:
Leo Gao's avatar
Leo Gao committed
153
        yield from iter
Leo Gao's avatar
Leo Gao committed
154
155


156
157
158
159
160
def group(arr, fn):
    res = collections.defaultdict(list)

    for ob in arr:
        res[fn(ob)].append(ob)
Fabrizio Milo's avatar
Fabrizio Milo committed
161

162
163
    return list(res.values())

Fabrizio Milo's avatar
Fabrizio Milo committed
164

gakada's avatar
gakada committed
165
166
167
# Returns a list containing all values of the source_list that
# match at least one of the patterns
def pattern_match(patterns, source_list):
168
    if isinstance(patterns, str):
169
170
        patterns = [patterns]

gakada's avatar
gakada committed
171
172
173
174
175
176
177
    task_names = set()
    for pattern in patterns:
        for matching in fnmatch.filter(source_list, pattern):
            task_names.add(matching)
    return sorted(list(task_names))


Baber Abbasi's avatar
Baber Abbasi committed
178
def softmax(x) -> np.ndarray:
Lintang Sutawika's avatar
Lintang Sutawika committed
179
180
181
182
183
    """Compute softmax values for each sets of scores in x."""
    e_x = np.exp(x - np.max(x))
    return e_x / e_x.sum()


Baber Abbasi's avatar
Baber Abbasi committed
184
def general_detokenize(string) -> str:
Leo Gao's avatar
Leo Gao committed
185
186
187
    string = string.replace(" n't", "n't")
    string = string.replace(" )", ")")
    string = string.replace("( ", "(")
Fabrizio Milo's avatar
Fabrizio Milo committed
188
189
    string = string.replace('" ', '"')
    string = string.replace(' "', '"')
Leo Gao's avatar
Fix  
Leo Gao committed
190
    string = re.sub(r" (['.,])", r"\1", string)
191
192
193
    return string


194
195
196
197
198
199
200
201
202
203
204
def get_file_task_name(filename: str) -> str:
    """
    Given the sample results filenames, extracts and returns the task name.
    """
    return filename[filename.find("_") + 1 : filename.rfind("_")]


def get_file_datetime(filename: str) -> str:
    """
    Given the results and sample results filenames, extracts and returns the datetime.
    """
205
    return filename[filename.rfind("_") + 1 :].replace(".jsonl", "")
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


def sanitize_model_name(model_name: str) -> str:
    """
    Given the model name, returns a sanitized version of it.
    """
    return re.sub(r"[\"<>:/\|\\?\*\[\]]+", "__", model_name)


def sanitize_task_name(task_name: str) -> str:
    """
    Given the task name, returns a sanitized version of it.
    """
    return re.sub(r"\W", "_", task_name)


def get_latest_filename(filenames: List[str]) -> str:
    """
    Given a list of filenames, returns the filename with the latest datetime.
    """
    return max(filenames, key=lambda f: get_file_datetime(f))


def get_results_filenames(filenames: List[str]) -> List[str]:
    """
    Extracts filenames that correspond to aggregated results.
    """
    return [f for f in filenames if "/results_" in f and ".json" in f]


def get_sample_results_filenames(filenames: List[str]) -> List[str]:
    """
    Extracts filenames that correspond to sample results.
    """
    return [f for f in filenames if "/samples_" in f and ".json" in f]


243
244
245
def get_rolling_token_windows(
    token_list: List[int], prefix_token: int, max_seq_len: int, context_len: int
) -> Generator[Tuple[List[int], List[int]], None, None]:
Jason Phang's avatar
Jason Phang committed
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
    """
    - context_len allows for a rolling window context, allowing each prediction window to potentially
      condition on some context

    :param token_list: list
        List of tokens to be PREDICTED
    :param max_seq_len: int
        max_seq_len of model (or max_seq_len we want to use)
    :param context_len: int
        Amount of desired token context for prediction. Needs to be at least 1.
    :param prefix_token: token
        Dummy token like <eos> so the first token has something to condition on
    :return: generator
        Generator of tuples
            (input_tokens, pred_tokens)
        Note: Score only the last len(pred_tokens) logits of the LM
    """
    assert 1 <= context_len <= max_seq_len
    if not token_list:
        return
    # +1 offset, going from input->preds
    pred_len = max_seq_len - context_len + 1
    predicted = 0

    # Special handling for first window: predict all tokens
    first_seq_len = min(max_seq_len, len(token_list))
272
    yield [prefix_token] + token_list[: first_seq_len - 1], token_list[:first_seq_len]
Jason Phang's avatar
Jason Phang committed
273
274
275
276
277
    predicted += first_seq_len

    while predicted < len(token_list):
        window_pred_len = min(len(token_list) - predicted, pred_len)
        window_end = predicted + window_pred_len
Leo Gao's avatar
Leo Gao committed
278

Jason Phang's avatar
Jason Phang committed
279
        yield (
lintangsutawika's avatar
lintangsutawika committed
280
281
            token_list[window_end - max_seq_len - 1 : window_end - 1],
            token_list[window_end - window_pred_len : window_end],
Jason Phang's avatar
Jason Phang committed
282
283
284
        )
        predicted += window_pred_len

Fabrizio Milo's avatar
Fabrizio Milo committed
285

286
287
288
def make_disjoint_window(
    pair: Tuple[List[int], List[int]],
) -> Tuple[List[int], List[int]]:
Fabrizio Milo's avatar
Fabrizio Milo committed
289
    """Takes output from get_rolling_token_windows and makes the context not overlap with the continuation"""
Leo Gao's avatar
Leo Gao committed
290
    a, b = pair
291
    return a[: len(a) - (len(b) - 1)], b
Fabrizio Milo's avatar
Fabrizio Milo committed
292

Jason Phang's avatar
Jason Phang committed
293

294
295
296
297
298
299
300
301
302
303
304
305
class EnhancedJSONEncoder(json.JSONEncoder):
    """
    Provides a proper json encoding for the loggers and trackers json dumps.
    Notably manages the json encoding of dataclasses.
    """

    def default(self, o):
        if is_dataclass(o):
            return asdict(o)
        return super().default(o)


306
class Reorderer:
baberabb's avatar
baberabb committed
307
308
309
310
311
312
313
    def __init__(self, arr: List[Any], fn: Callable) -> None:
        """Reorder an array according to some function

        Args:
            arr (List[Any]): The initial array
            fn (Callable[[Any], Any]): A function to determine the priority of elements
        """
314
315
316
        self.size = len(arr)
        arr = list(enumerate(arr))
        arr = group(arr, lambda x: fn(x[1]))
317
318
319
        # arr = [([y[0] for y in x], x[0][1]) for x in arr]
        # TODO: overhaul reorderer. It currently grouped requests by content but we don't want this
        arr = [([y[0]], x[0][1]) for x in arr for y in x]
320
321
322
        arr.sort(key=lambda x: fn(x[1]))

        self.arr = arr
Fabrizio Milo's avatar
Fabrizio Milo committed
323

324
    def get_reordered(self):
baberabb's avatar
baberabb committed
325
326
327
328
329
        """Gets the reordered array

        Returns:
            List[Any]: The reordered array
        """
330
        return [x[1] for x in self.arr]
Fabrizio Milo's avatar
Fabrizio Milo committed
331

332
    def get_original(self, newarr):
baberabb's avatar
baberabb committed
333
334
335
336
337
338
339
340
        """Restores the original order of a new array based on the old array's order

        Args:
            newarr (List[Any]): The array to be restored

        Returns:
            List[Any]: The array restored to the original order
        """
341
342
343
344
        res = [None] * self.size
        cov = [False] * self.size

        for (inds, _), v in zip(self.arr, newarr):
Fabrizio Milo's avatar
Fabrizio Milo committed
345
            for ind in inds:
346
347
                res[ind] = v
                cov[ind] = True
Fabrizio Milo's avatar
Fabrizio Milo committed
348

349
        assert all(cov)
Fabrizio Milo's avatar
Fabrizio Milo committed
350

351
352
        return res

Fabrizio Milo's avatar
Fabrizio Milo committed
353

Lintang Sutawika's avatar
Lintang Sutawika committed
354
def make_table(result_dict, column: str = "results", sort_results: bool = False):
355
    """Generate table of results."""
356
    from pytablewriter import LatexTableWriter, MarkdownTableWriter
357

lintangsutawika's avatar
lintangsutawika committed
358
    if column == "results":
lintangsutawika's avatar
lintangsutawika committed
359
360
361
        column_name = "Tasks"
    elif column == "groups":
        column_name = "Groups"
lintangsutawika's avatar
lintangsutawika committed
362

lintangsutawika's avatar
lintangsutawika committed
363
    all_headers = [
lintangsutawika's avatar
lintangsutawika committed
364
        column_name,
lintangsutawika's avatar
lintangsutawika committed
365
366
        "Version",
        "Filter",
367
        "n-shot",
lintangsutawika's avatar
lintangsutawika committed
368
        "Metric",
369
        "",
lintangsutawika's avatar
lintangsutawika committed
370
371
372
373
        "Value",
        "",
        "Stderr",
    ]
374

lintangsutawika's avatar
lintangsutawika committed
375
376
377
378
379
    md_writer = MarkdownTableWriter()
    latex_writer = LatexTableWriter()
    md_writer.headers = all_headers
    latex_writer.headers = all_headers

380
381
    values = []

382
383
    keys = result_dict[column].keys()
    if sort_results:
Lintang Sutawika's avatar
Lintang Sutawika committed
384
385
386
        # sort entries alphabetically by task or group name.
        # NOTE: we default here to false, because order matters for multi-level table printing a la mmlu.
        # sorting here would mess that up
387
388
389
        keys = sorted(keys)
    for k in keys:
        dic = result_dict[column][k]
Lintang Sutawika's avatar
Lintang Sutawika committed
390
391
        version = result_dict["versions"].get(k, "    N/A")
        n = str(result_dict.get("n-shot", " ").get(k, " "))
392
        higher_is_better = result_dict.get("higher_is_better", {}).get(k, {})
393
394
395
396

        if "alias" in dic:
            k = dic.pop("alias")

397
        metric_items = dic.items()
Lintang Sutawika's avatar
Lintang Sutawika committed
398
        metric_items = sorted(metric_items)
399
400

        for (mf), v in metric_items:
401
            m, _, f = mf.partition(",")
402
403
404
            if m.endswith("_stderr"):
                continue

405
406
            hib = HIGHER_IS_BETTER_SYMBOLS.get(higher_is_better.get(m), "")

Lintang Sutawika's avatar
Lintang Sutawika committed
407
408
            v = "%.4f" % v if isinstance(v, float) else v

409
410
            if m + "_stderr" + "," + f in dic:
                se = dic[m + "_stderr" + "," + f]
Lintang Sutawika's avatar
Lintang Sutawika committed
411
                se = "   N/A" if se == "N/A" else "%.4f" % se
Lintang Sutawika's avatar
Lintang Sutawika committed
412
                values.append([k, version, f, n, m, hib, v, "±", se])
413
            else:
Lintang Sutawika's avatar
Lintang Sutawika committed
414
                values.append([k, version, f, n, m, hib, v, "", ""])
415
416
417
418
419
420
421
422
423
424
425
            k = ""
            version = ""
    md_writer.value_matrix = values
    latex_writer.value_matrix = values

    # todo: make latex table look good
    # print(latex_writer.dumps())

    return md_writer.dumps()


426
427
def positional_deprecated(fn):
    """
Fabrizio Milo's avatar
Fabrizio Milo committed
428
    A decorator to nudge users into passing only keyword args (`kwargs`) to the
429
430
    wrapped function, `fn`.
    """
Fabrizio Milo's avatar
Fabrizio Milo committed
431

432
433
    @functools.wraps(fn)
    def _wrapper(*args, **kwargs):
Fabrizio Milo's avatar
Fabrizio Milo committed
434
435
436
        if len(args) != 1 if inspect.ismethod(fn) else 0:
            print(
                f"WARNING: using {fn.__name__} with positional arguments is "
437
                "deprecated and will be disallowed in a future version of "
Fabrizio Milo's avatar
Fabrizio Milo committed
438
439
                "lm-evaluation-harness!"
            )
440
        return fn(*args, **kwargs)
Fabrizio Milo's avatar
Fabrizio Milo committed
441

442
    return _wrapper
Stephen Hogg's avatar
Stephen Hogg committed
443

Fabrizio Milo's avatar
Fabrizio Milo committed
444

445
446
447
448
def ignore_constructor(loader, node):
    return node


449
def import_function(loader: yaml.Loader, node, yaml_path: Path):
lintangsutawika's avatar
lintangsutawika committed
450
451
    function_name = loader.construct_scalar(node)

lintangsutawika's avatar
lintangsutawika committed
452
    *module_name, function_name = function_name.split(".")
453
    if isinstance(module_name, list):
lintangsutawika's avatar
lintangsutawika committed
454
        module_name = ".".join(module_name)
455
    module_path = yaml_path.parent / f"{module_name}.py"
lintangsutawika's avatar
lintangsutawika committed
456

457
458
459
460
    spec = importlib.util.spec_from_file_location(module_name, module_path.as_posix())

    if spec is None:
        raise ImportError(f"Could not import module {module_name} from {module_path}.")
lintangsutawika's avatar
lintangsutawika committed
461
    module = importlib.util.module_from_spec(spec)
462
463
464

    if spec.loader is None:
        raise ImportError(f"Module loader is None, {module_name} from {module_path}.")
lintangsutawika's avatar
lintangsutawika committed
465
466
467
468
469
    spec.loader.exec_module(module)

    function = getattr(module, function_name)
    return function

lintangsutawika's avatar
lintangsutawika committed
470

471
472
473
474
def load_yaml_config(yaml_path=None, yaml_config=None, yaml_dir=None, mode="full"):
    if mode == "simple":
        constructor_fn = ignore_constructor
    elif mode == "full":
475
476
477
478
        if yaml_path is None:
            raise ValueError("yaml_path must be provided if mode is 'full'.")
        # Attach yaml_path to the import function so that it can be used later
        constructor_fn = functools.partial(import_function, yaml_path=Path(yaml_path))
lintangsutawika's avatar
lintangsutawika committed
479

480
    loader = yaml.CLoader if yaml.__with_libyaml__ else yaml.FullLoader
481
    # Add the import_function constructor to the YAML loader
482
    yaml.add_constructor("!function", constructor_fn, Loader=loader)
483
484
    if yaml_config is None:
        with open(yaml_path, "rb") as file:
485
            yaml_config = yaml.load(file, Loader=loader)
lintangsutawika's avatar
lintangsutawika committed
486

lintangsutawika's avatar
lintangsutawika committed
487
488
    if yaml_dir is None:
        yaml_dir = os.path.dirname(yaml_path)
489
490
491
492
493
494
495

    assert yaml_dir is not None

    if "include" in yaml_config:
        include_path = yaml_config["include"]
        del yaml_config["include"]

496
        if isinstance(include_path, str):
497
498
499
500
501
502
503
504
505
506
507
508
509
            include_path = [include_path]

        # Load from the last one first
        include_path.reverse()
        final_yaml_config = {}
        for path in include_path:
            # Assumes that path is a full path.
            # If not found, assume the included yaml
            # is in the same dir as the original yaml
            if not os.path.isfile(path):
                path = os.path.join(yaml_dir, path)

            try:
510
                included_yaml_config = load_yaml_config(yaml_path=path, mode=mode)
511
512
513
514
515
516
517
518
                final_yaml_config.update(included_yaml_config)
            except Exception as ex:
                # If failed to load, ignore
                raise ex

        final_yaml_config.update(yaml_config)
        return final_yaml_config
    return yaml_config
lintangsutawika's avatar
lintangsutawika committed
519
520


Ethan Smith's avatar
Ethan Smith committed
521
def regex_replace(string, pattern, repl, count: int = 0):
522
523
    """Implements the `re.sub` function as a custom Jinja filter."""
    return re.sub(pattern, repl, string, count=count)
lintangsutawika's avatar
lintangsutawika committed
524

lintangsutawika's avatar
lintangsutawika committed
525

526
527
528
env = Environment(
    loader=BaseLoader, undefined=StrictUndefined, keep_trailing_newline=True
)
529
env.filters["regex_replace"] = regex_replace
530
531


baberabb's avatar
baberabb committed
532
def apply_template(template: str, doc: dict) -> str:
533
534
    rtemplate = env.from_string(template)
    return rtemplate.render(**doc)
535
536


537
def create_iterator(raw_iterator, *, rank=0, world_size=1, limit=None):
538
539
540
    """
    Method for creating a (potentially) sliced and limited
    iterator from a raw document iterator. Used for splitting data
541
542
543
    among ranks in multigpu setting or only pulling a sample of documents
    """
    return islice(raw_iterator, rank, limit, world_size)
544
545
546
547
548
549
550
551
552
553


def weighted_f1_score(items):
    from sklearn.metrics import f1_score

    unzipped_list = list(zip(*items))
    golds = unzipped_list[0]
    preds = unzipped_list[1]
    fscore = f1_score(golds, preds, average="weighted")
    return fscore
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646


def parse_namespace(namespace: argparse.Namespace) -> Dict[str, Any]:
    """
    Convert an argparse Namespace object to a dictionary.

    Handles all argument types including boolean flags, lists, and None values.
    Only includes arguments that were explicitly set (ignores defaults unless used).

    Args:
        namespace: The argparse.Namespace object to convert

    Returns:
        Dictionary containing all the namespace arguments and their values
    """

    config = {key: value for key, value in vars(namespace).items()}
    for key in config:
        # TODO: pass this list as param
        if key in [
            "wandb_args",
            "wandb_config_args",
            "hf_hub_log_args",
            "metadata",
            "model_args",
        ]:
            if not isinstance(config[key], dict):
                config[key] = simple_parse_args_string(config[key])
    if "model_args" not in config:
        config["model_args"] = {}
    if "metadata" not in config:
        config["metadata"] = {}

    non_default_args = []
    if hasattr(namespace, "_explicit_args"):
        non_default_args = namespace._explicit_args

    return config, non_default_args


class TrackExplicitAction(argparse.Action):
    def __call__(self, parser, namespace, values, option_string=None):
        # Create a set on the namespace to track explicitly set arguments if it doesn't exist
        if not hasattr(namespace, "_explicit_args"):
            setattr(namespace, "_explicit_args", set())
        # Record that this argument was explicitly provided
        namespace._explicit_args.add(self.dest)
        setattr(namespace, self.dest, values)


class TrackExplicitStoreTrue(argparse.Action):
    def __init__(
        self, option_strings, dest, nargs=0, const=True, default=False, **kwargs
    ):
        # Ensure that nargs is 0, as required by store_true actions.
        if nargs != 0:
            raise ValueError("nargs for store_true actions must be 0")
        super().__init__(
            option_strings, dest, nargs=0, const=const, default=default, **kwargs
        )

    def __call__(self, parser, namespace, values, option_string=None):
        # Initialize or update the set of explicitly provided arguments.
        if not hasattr(namespace, "_explicit_args"):
            setattr(namespace, "_explicit_args", set())
        namespace._explicit_args.add(self.dest)
        setattr(namespace, self.dest, self.const)


def non_default_update(console_dict, local_dict, non_default_args):
    """
    Update local_dict by taking non-default values from console_dict.

    Args:
        console_dict (dict): The dictionary that potentially provides updates.
        local_dict (dict): The dictionary to be updated.
        non_default_args (list): The list of args passed by user in console.

    Returns:
        dict: The updated local_dict.
    """
    result_config = {}

    for key in set(console_dict.keys()).union(local_dict.keys()):
        if key in non_default_args:
            result_config[key] = console_dict[key]
        else:
            if key in local_dict:
                result_config[key] = local_dict[key]
            else:
                result_config[key] = console_dict[key]

    return result_config