edspdf.config

class Config(dict):
    def __init__(self, *args, **kwargs):
        if len(args) == 1 and isinstance(args[0], dict):
            assert len(kwargs) == 0
            kwargs = args[0]
        path = kwargs.pop("__path__", None)
        kwargs = {
            key: Config(value)
            if isinstance(value, dict) and not isinstance(value, Config)
            else value
            for key, value in kwargs.items()
        }
        super().__init__(**kwargs)
        self.__path__ = path

    @classmethod
    def from_str(cls, s: str, resolve: bool = False) -> "Config":
        parser = ConfigParser()
        parser.optionxform = str
        parser.read_string(s)

        config = Config()

        for section in parser.sections():
            parts = split_path(section)
            current = config
            for part in parts:
                if part not in current:
                    current[part] = current = Config()
                else:
                    current = current[part]

            current.clear()
            current.update(
                {k: config_literal_eval(v) for k, v in parser.items(section)}
            )

        if resolve:
            return config.resolve()

        return config

    @classmethod
    def from_disk(cls, path: Union[str, Path], resolve: bool = False) -> "Config":
        s = Path(path).read_text()
        return cls.from_str(s, resolve=resolve)

    def to_disk(self, path: Union[str, Path]):
        s = self.to_str()
        Path(path).write_text(s)

    def serialize(self):
        """
        Try to convert non-serializable objects using the RESOLVED object
        back to their original catalogue + params form

        Returns
        -------
        Config
        """
        refs = {}

        def rec(o, path=()):
            if o is None or isinstance(
                o, (str, int, float, bool, tuple, list, Reference)
            ):
                return o
            if isinstance(o, collections.Mapping):
                serialized = {k: rec(v, (*path, k)) for k, v in o.items()}
                if isinstance(o, Config):
                    serialized = Config(serialized)
                    serialized.__path__ = o.__path__
                return serialized
            cfg = None
            try:
                cfg = o.cfg
            except AttributeError:
                try:
                    cfg = RESOLVED[o]
                except KeyError:
                    pass
            if cfg is not None:
                if id(o) in refs:
                    return refs[id(o)]
                else:
                    refs[id(o)] = Reference(join_path(path))
                return rec(cfg, path)
            raise TypeError(f"Cannot dump {o!r}")

        result = rec(self)
        return result

    def to_str(self):
        additional_sections = {}

        def rec(o, path=()):
            if isinstance(o, collections.Mapping):
                if isinstance(o, Config) and o.__path__ is not None:
                    res = {k: rec(v, (*o.__path__, k)) for k, v in o.items()}
                    current = additional_sections
                    for part in o.__path__[:-1]:
                        current = current.setdefault(part, Config())
                    current[o.__path__[-1]] = res
                    return Reference(join_path(o.__path__))
                else:
                    return {k: rec(v, (*path, k)) for k, v in o.items()}
            return o

        prepared = flatten_sections(rec(self.serialize()))
        prepared.update(flatten_sections(additional_sections))

        parser = ConfigParser()
        parser.optionxform = str
        for section_name, section in prepared.items():
            parser.add_section(section_name)
            parser[section_name].update(
                {k: config_literal_dump(v) for k, v in section.items()}
            )
        s = StringIO()
        parser.write(s)
        return s.getvalue()

    def resolve(self, _path=(), leaves=None, deep=True):
        from .registry import registry  # local import because circular deps

        copy = Config(**self)
        if leaves is None:
            leaves = {}
        missing = []
        items = [(k, v) for k, v in copy.items()] if deep else []
        last_count = len(leaves)
        while len(items):
            traced_missing_values = []
            for key, value in items:
                try:
                    if isinstance(value, Config):
                        if (*_path, key) not in leaves:
                            leaves[(*_path, key)] = value.resolve((*_path, key), leaves)
                        copy[key] = leaves[(*_path, key)]
                    elif isinstance(value, Reference):
                        try:
                            leaves[(*_path, key)] = leaves[tuple(split_path(value))]
                        except KeyError:
                            raise MissingReference([value])
                        else:
                            copy[key] = leaves[(*_path, key)]

                except MissingReference as e:
                    traced_missing_values.extend(e.references)
                    missing.append((key, value))
            if len(missing) > 0 and len(leaves) <= last_count:
                raise MissingReference(dedup(traced_missing_values))
            items = list(missing)
            last_count = len(leaves)
            missing = []

        registries = [
            (key, value, registry._catalogue[key[1:]])
            for key, value in copy.items()
            if key.startswith("@")
        ]
        assert len(registries) <= 1, (
            f"Cannot resolve using multiple " f"registries at {'.'.join(_path)}"
        )

        def patch_errors(errors: Union[Sequence[ErrorWrapper], ErrorWrapper]):
            if isinstance(errors, list):
                res = []
                for error in errors:
                    res.append(patch_errors(error))
                return res
            return ErrorWrapper(errors.exc, (*_path, *errors.loc_tuple()))

        if len(registries) == 1:
            params = dict(copy)
            params.pop(registries[0][0])
            fn = registries[0][2].get(registries[0][1])
            try:
                resolved = fn(**params)
                try:
                    resolved.cfg
                except Exception:
                    try:
                        RESOLVED[resolved] = self
                    except Exception:
                        print(f"Could not store original config for {resolved}")
                        pass

                return resolved
            except ValidationError as e:
                raise ValidationError(patch_errors(e.raw_errors), e.model)

        return copy

    def merge(
        self,
        *updates: Union[Dict[str, Any], "Config"],
        remove_extra: bool = False,
    ) -> "Config":
        """
        Deep merge two configs. Largely inspired from spaCy config merge function.

        Parameters
        ----------
        updates: Union[Config, Dict]
            Configs to update the original config
        remove_extra:
            If true, restricts update to keys that existed in the original config

        Returns
        -------
        The new config
        """

        def deep_set(current, path, val):
            try:
                path = split_path(path)
                for part in path[:-1]:
                    current = (
                        current[part] if remove_extra else current.setdefault(part, {})
                    )
            except KeyError:
                return
            if path[-1] not in current and remove_extra:
                return
            current[path[-1]] = val

        def rec(old, new):
            for key, new_val in list(new.items()):
                if "." in key:
                    deep_set(old, key, new_val)
                    continue

                if key not in old:
                    if remove_extra:
                        continue
                    else:
                        old[key] = new_val
                        continue

                old_val = old[key]
                if isinstance(old_val, dict) and isinstance(new_val, dict):
                    old_resolver = next((k for k in old_val if k.startswith("@")), None)
                    new_resolver = next((k for k in new_val if k.startswith("@")), None)
                    if (
                        new_resolver is not None
                        and old_resolver is not None
                        and (
                            old_resolver != new_resolver
                            or old_val.get(old_resolver) != new_val.get(new_resolver)
                        )
                    ):
                        old[key] = new_val
                    else:
                        rec(old[key], new_val)
                else:
                    old[key] = new_val
            return old

        config = deepcopy(self)
        for u in updates:
            u = deepcopy(u)
            rec(config, u)
        return Config(**config)