File size: 2,127 Bytes
951a78d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import json
import torch
from typing import Any

# DEVICES

if torch.cuda.is_available():
    device = torch.device("cuda")
elif torch.backends.mps.is_available():
    device = torch.device("mps")
else:
    device = torch.device("cpu")

print(f"[MultimodalReasoning] PyTorch is using primary device: {device}.")

# CONFIG PATHS
class ConfigNode(dict):
    """
    Dict that supports both:
        cfg["Path"]
        cfg.Path

    Nested dictionaries are automatically wrapped too.
    """

    def __init__(self, data: dict):
        super().__init__()

        for key, value in data.items():
            self[key] = self._wrap(value)

    @staticmethod
    def _wrap(value: Any) -> Any:
        if isinstance(value, dict):
            return ConfigNode(value)

        if isinstance(value, list):
            return [ConfigNode._wrap(x) for x in value]

        return value

    def __getattr__(self, key: str) -> Any:
        try:
            return self[key]
        except KeyError:
            raise AttributeError(f"Config key not found: {key}") from None

    def __setattr__(self, key: str, value: Any):
        self[key] = self._wrap(value)

    def __delattr__(self, key: str):
        try:
            del self[key]
        except KeyError:
            raise AttributeError(f"Config key not found: {key}") from None

with open("configs.json", "r", encoding="utf-8") as f:
    configs = ConfigNode(json.load(f))

def get(path: str | tuple[str, ...] | list[str], default: Any = None, *, strict: bool = True) -> Any:
    """
    Examples:
        get("LLM/MODEL_NAME")
        get(("DATASET", "REASONING_SFT"))
        get(["DATASET", "REASONING_SFT", "Path"])
    """

    if isinstance(path, str):
        parts = [p for p in path.split("/") if p]
    else:
        parts = list(path)

    current = configs

    for key in parts:
        try:
            current = current[key]
        except (KeyError, TypeError):
            if strict:
                raise KeyError(f"Config path not found: {'/'.join(parts)} at key '{key}'")
            return default

    return current

def get_device():
    return device