class ConfigDict:
    def __init__(self, d):
        self._conf = normalize_or_die(d)

    def __call__(self, fixed=None, preset=None, fallback=None):
        result = dogmatize(fixed or {})
        recursive_fill_in(result, self._conf)
        recursive_fill_in(result, preset or {})
        added = result.revelation()
        config_summary = ConfigSummary(added, result.modified, result.typechanges)
        config_summary.update(undogmatize(result))
        return config_summary def load_config_file(filename):
    handler = get_handler(filename)
    with open(filename, "r" + handler.mode) as f:
        return handler.load(f) def chain_evaluate_config_scopes(config_scopes, fixed=None, preset=None, fallback=None):
    fixed = fixed or {}
    fallback = fallback or {}
    final_config = dict(preset or {})
    config_summaries = []
    for config in config_scopes:
        cfg = config(fixed=fixed, preset=final_config, fallback=fallback)
        config_summaries.append(cfg)
        final_config.update(cfg)

    if not config_scopes:
        final_config.update(fixed)

    return undogmatize(final_config), config_summaries def dogmatize(obj):
    if isinstance(obj, dict):
        return DogmaticDict({key: dogmatize(val) for key, val in obj.items()})
    elif isinstance(obj, list):
        return DogmaticList([dogmatize(value) for value in obj])
    elif isinstance(obj, tuple):
        return tuple(dogmatize(value) for value in obj)
    else:
        return obj def undogmatize(obj):
    if isinstance(obj, DogmaticDict):
        return dict({key: undogmatize(value) for key, value in obj.items()})
    elif isinstance(obj, DogmaticList):
        return list([undogmatize(value) for value in obj])
    elif isinstance(obj, tuple):
        return tuple(undogmatize(value) for value in obj)
    else:
        return obj class ConfigSummary(dict):
    def __init__(
        self, added=(), modified=(), typechanged=(), ignored_fallbacks=(), docs=()
    ):
        super().__init__()
        self.added = set(added)
        self.modified = set(modified)  # TODO: test for this member
        self.typechanged = dict(typechanged)
        self.ignored_fallbacks = set(ignored_fallbacks)  # TODO: test
        self.docs = dict(docs)
        self.ensure_coherence()

    def update_from(self, config_mod, path=""):
        added = config_mod.added
        updated = config_mod.modified
        typechanged = config_mod.typechanged
        self.added &= {join_paths(path, a) for a in added}
        self.modified |= {join_paths(path, u) for u in updated}
        self.typechanged.update(
            {join_paths(path, k): v for k, v in typechanged.items()}
        )
        self.ensure_coherence()
        for k, v in config_mod.docs.items():
            if not self.docs.get(k, ""):
                self.docs[k] = v

    def update_add(self, config_mod, path=""):
        added = config_mod.added
        updated = config_mod.modified
        typechanged = config_mod.typechanged
        self.added |= {join_paths(path, a) for a in added}
        self.modified |= {join_paths(path, u) for u in updated}
        self.typechanged.update(
            {join_paths(path, k): v for k, v in typechanged.items()}
        )
        self.docs.update(
            {
                join_paths(path, k): v
                for k, v in config_mod.docs.items()
                if path == "" or k != "seed"
            }
        )
        self.ensure_coherence()

    def ensure_coherence(self):
        # make sure parent paths show up as updated appropriately
        self.modified |= {p for a in self.added for p in iter_prefixes(a)}
        self.modified |= {p for u in self.modified for p in iter_prefixes(u)}
        self.modified |= {p for t in self.typechanged for p in iter_prefixes(t)}

        # make sure there is no overlap
        self.added -= set(self.typechanged.keys())
        self.modified -= set(self.typechanged.keys())
        self.modified -= self.added def make_read_only(o):
    """Makes objects read-only.

    Converts every `list` and `dict` into `ReadOnlyList` and `ReadOnlyDict` in
    a nested structure of `list`s, `dict`s and `tuple`s. Does not modify `o`
    but returns the converted structure.
    """
    if type(o) == dict:
        return ReadOnlyDict({k: make_read_only(v) for k, v in o.items()})
    elif type(o) == list:
        return ReadOnlyList([make_read_only(v) for v in o])
    elif type(o) == tuple:
        return tuple(map(make_read_only, o))
    else:
        return o def get_host_info(additional_host_info: List[HostInfoGetter] = None):
    """Collect some information about the machine this experiment runs on.

    Returns
    -------
    dict
        A dictionary with information about the CPU, the OS and the
        Python version of this machine.

    """
    additional_host_info = additional_host_info or []
    # can't use += because we don't want to modify the mutable argument.
    additional_host_info = additional_host_info + _host_info_gatherers_list
    all_host_info_gatherers = host_info_gatherers.copy()
    for getter in additional_host_info:
        all_host_info_gatherers[getter.name] = getter
    host_info = {}
    for k, v in all_host_info_gatherers.items():
        try:
            host_info[k] = v()
        except IgnoreHostInfo:
            pass
    return host_info def create_rnd(seed):
    assert isinstance(seed, int), "Seed has to be integer but was {} {}".format(
        repr(seed), type(seed)
    )
    if opt.has_numpy:
        if SETTINGS.CONFIG.NUMPY_RANDOM_LEGACY_API:
            return opt.np.random.RandomState(seed)
        else:
            return opt.np.random.default_rng(seed)
    else:
        return random.Random(seed) def get_seed(rnd=None):
    if rnd is None:
        return random.randint(*SEEDRANGE)
    else:
        try:
            return rnd.integers(*SEEDRANGE, dtype=int)
        except Exception:
            return rnd.randint(*SEEDRANGE) class Run:
    """Represent and manage a single run of an experiment."""

    def __init__(
        self,
        config,
        config_modifications,
        main_function,
        observers,
        root_logger,
        run_logger,
        experiment_info,
        host_info,
        pre_run_hooks,
        post_run_hooks,
        captured_out_filter=None,
    ):

        self._id = None
        """The ID of this run as assigned by the first observer"""

        self.captured_out = ""
        """Captured stdout and stderr"""

        self.config = config
        """The final configuration used for this run"""

        self.config_modifications = config_modifications
        """A ConfigSummary object with information about config changes"""

        self.experiment_info = experiment_info
        """A dictionary with information about the experiment"""

        self.host_info = host_info
        """A dictionary with information about the host"""

        self.info = {}
        """Custom info dict that will be sent to the observers"""

        self.root_logger = root_logger
        """The root logger that was used to create all the others"""

        self.run_logger = run_logger
        """The logger that is used for this run"""

        self.main_function = main_function
        """The main function that is executed with this run"""

        self.observers = observers
        """A list of all observers that observe this run"""

        self.pre_run_hooks = pre_run_hooks
        """List of pre-run hooks (captured functions called before this run)"""

        self.post_run_hooks = post_run_hooks
        """List of post-run hooks (captured functions called after this run)"""

        self.result = None
        """The return value of the main function"""

        self.status = None
        """The current status of the run, from QUEUED to COMPLETED"""

        self.start_time = None
        """The datetime when this run was started"""

        self.stop_time = None
        """The datetime when this run stopped"""

        self.debug = False
        """Determines whether this run is executed in debug mode"""

        self.pdb = False
        """If true the pdb debugger is automatically started after a failure"""

        self.meta_info = {}
        """A custom comment for this run"""

        self.beat_interval = 10.0  # sec
        """The time between two heartbeat events measured in seconds"""

        self.unobserved = False
        """Indicates whether this run should be unobserved"""

        self.force = False
        """Disable warnings about suspicious changes"""

        self.queue_only = False
        """If true then this run will only fire the queued_event and quit"""

        self.captured_out_filter = captured_out_filter
        """Filter function to be applied to captured output"""

        self.fail_trace = None
        """A stacktrace, in case the run failed"""

        self.capture_mode = None
        """Determines the way the stdout/stderr are captured"""

        self._heartbeat = None
        self._failed_observers = []
        self._output_file = None

        self._metrics = metrics_logger.MetricsLogger()

    def open_resource(self, filename, mode="r"):
        """Open a file and also save it as a resource.

        Opens a file, reports it to the observers as a resource, and returns
        the opened file.

        In Sacred terminology a resource is a file that the experiment needed
        to access during a run. In case of a MongoObserver that means making
        sure the file is stored in the database (but avoiding duplicates) along
        its path and md5 sum.

        See also :py:meth:`sacred.Experiment.open_resource`.

        Parameters
        ----------
        filename : str
            name of the file that should be opened
        mode : str
            mode that file will be open

        Returns
        -------
        file
            the opened file-object

        """
        filename = os.path.abspath(filename)
        self._emit_resource_added(filename)  # TODO: maybe non-blocking?
        return open(filename, mode)

    def add_resource(self, filename):
        """Add a file as a resource.

        In Sacred terminology a resource is a file that the experiment needed
        to access during a run. In case of a MongoObserver that means making
        sure the file is stored in the database (but avoiding duplicates) along
        its path and md5 sum.

        See also :py:meth:`sacred.Experiment.add_resource`.

        Parameters
        ----------
        filename : str
            name of the file to be stored as a resource
        """
        filename = os.path.abspath(filename)
        self._emit_resource_added(filename)

    def add_artifact(self, filename, name=None, metadata=None, content_type=None):
        """Add a file as an artifact.

        In Sacred terminology an artifact is a file produced by the experiment
        run. In case of a MongoObserver that means storing the file in the
        database.

        See also :py:meth:`sacred.Experiment.add_artifact`.

        Parameters
        ----------
        filename : str
            name of the file to be stored as artifact
        name : str, optional
            optionally set the name of the artifact.
            Defaults to the filename.
        metadata: dict
            optionally attach metadata to the artifact.
            This only has an effect when using the MongoObserver.
        content_type: str, optional
            optionally attach a content-type to the artifact.
            This only has an effect when using the MongoObserver.
        """
        filename = os.path.abspath(filename)
        name = os.path.basename(filename) if name is None else name
        self._emit_artifact_added(name, filename, metadata, content_type)

    def __call__(self, *args):
        r"""Start this run.

        Parameters
        ----------
        \*args
            parameters passed to the main function

        Returns
        -------
            the return value of the main function

        """
        if self.start_time is not None:
            raise RuntimeError(
                "A run can only be started once. "
                "(Last start was {})".format(self.start_time)
            )

        if self.unobserved:
            self.observers = []
        else:
            self.observers = sorted(self.observers, key=lambda x: -x.priority)

        self.warn_if_unobserved()
        set_global_seed(self.config["seed"])

        if self.capture_mode is None and not self.observers:
            capture_mode = "no"
        else:
            capture_mode = self.capture_mode
        capture_mode, capture_stdout = get_stdcapturer(capture_mode)
        self.run_logger.debug('Using capture mode "%s"', capture_mode)

        if self.queue_only:
            self._emit_queued()
            return
        try:
            with capture_stdout() as self._output_file:
                self._emit_started()
                self._start_heartbeat()
                self._execute_pre_run_hooks()
                self.result = self.main_function(*args)
                self._execute_post_run_hooks()
                if self.result is not None:
                    self.run_logger.info("Result: {}".format(self.result))
                elapsed_time = self._stop_time()
                self.run_logger.info("Completed after %s", elapsed_time)
                self._get_captured_output()
            self._stop_heartbeat()
            self._emit_completed(self.result)
        except (SacredInterrupt, KeyboardInterrupt) as e:
            self._stop_heartbeat()
            status = getattr(e, "STATUS", "INTERRUPTED")
            self._emit_interrupted(status)
            raise
        except BaseException:
            exc_type, exc_value, trace = sys.exc_info()
            self._stop_heartbeat()
            self._emit_failed(exc_type, exc_value, trace.tb_next)
            raise
        finally:
            self._warn_about_failed_observers()
            self._wait_for_observers()

        return self.result

    def _get_captured_output(self):
        if self._output_file.closed:
            return
        text = self._output_file.get()
        if isinstance(text, bytes):
            text = text.decode("utf-8", "replace")
        if self.captured_out:
            text = self.captured_out + text
        if self.captured_out_filter is not None:
            text = self.captured_out_filter(text)
        self.captured_out = text

    def _start_heartbeat(self):
        self.run_logger.debug("Starting Heartbeat")
        if self.beat_interval > 0:
            self._stop_heartbeat_event, self._heartbeat = IntervalTimer.create(
                self._emit_heartbeat, self.beat_interval
            )
            self._heartbeat.start()

    def _stop_heartbeat(self):
        self.run_logger.debug("Stopping Heartbeat")
        # only stop if heartbeat was started
        if self._heartbeat is not None:
            self._stop_heartbeat_event.set()
            self._heartbeat.join(timeout=2)

    def _emit_queued(self):
        self.status = "QUEUED"
        queue_time = datetime.datetime.utcnow()
        self.meta_info["queue_time"] = queue_time
        command = join_paths(
            self.main_function.prefix, self.main_function.signature.name
        )
        self.run_logger.info("Queuing-up command '%s'", command)
        for observer in self.observers:
            _id = observer.queued_event(
                ex_info=self.experiment_info,
                command=command,
                host_info=self.host_info,
                queue_time=queue_time,
                config=self.config,
                meta_info=self.meta_info,
                _id=self._id,
            )
            if self._id is None:
                self._id = _id
            # do not catch any exceptions on startup:
            # the experiment SHOULD fail if any of the observers fails

        if self._id is None:
            self.run_logger.info("Queued")
        else:
            self.run_logger.info('Queued-up run with ID "{}"'.format(self._id))

    def _emit_started(self):
        self.status = "RUNNING"
        self.start_time = datetime.datetime.utcnow()
        command = join_paths(
            self.main_function.prefix, self.main_function.signature.name
        )
        self.run_logger.info("Running command '%s'", command)
        for observer in self.observers:
            _id = observer.started_event(
                ex_info=self.experiment_info,
                command=command,
                host_info=self.host_info,
                start_time=self.start_time,
                config=self.config,
                meta_info=self.meta_info,
                _id=self._id,
            )
            if self._id is None:
                self._id = _id
            # do not catch any exceptions on startup:
            # the experiment SHOULD fail if any of the observers fails
        if self._id is None:
            self.run_logger.info("Started")
        else:
            self.run_logger.info('Started run with ID "{}"'.format(self._id))

    def _emit_heartbeat(self):
        beat_time = datetime.datetime.utcnow()
        self._get_captured_output()
        # Read all measured metrics since last heartbeat
        logged_metrics = self._metrics.get_last_metrics()
        metrics_by_name = linearize_metrics(logged_metrics)
        for observer in self.observers:
            self._safe_call(
                observer, "log_metrics", metrics_by_name=metrics_by_name, info=self.info
            )
            self._safe_call(
                observer,
                "heartbeat_event",
                info=self.info,
                captured_out=self.captured_out,
                beat_time=beat_time,
                result=self.result,
            )

    def _stop_time(self):
        self.stop_time = datetime.datetime.utcnow()
        elapsed_time = datetime.timedelta(
            seconds=round((self.stop_time - self.start_time).total_seconds())
        )
        return elapsed_time

    def _emit_completed(self, result):
        self.status = "COMPLETED"
        for observer in self.observers:
            self._final_call(
                observer, "completed_event", stop_time=self.stop_time, result=result
            )

    def _emit_interrupted(self, status):
        self.status = status
        elapsed_time = self._stop_time()
        self.run_logger.warning("Aborted after %s!", elapsed_time)
        for observer in self.observers:
            self._final_call(
                observer,
                "interrupted_event",
                interrupt_time=self.stop_time,
                status=status,
            )

    def _emit_failed(self, exc_type, exc_value, trace):
        self.status = "FAILED"
        elapsed_time = self._stop_time()
        self.run_logger.error("Failed after %s!", elapsed_time)
        self.fail_trace = tb.format_exception(exc_type, exc_value, trace)
        for observer in self.observers:
            self._final_call(
                observer,
                "failed_event",
                fail_time=self.stop_time,
                fail_trace=self.fail_trace,
            )

    def _emit_resource_added(self, filename):
        for observer in self.observers:
            self._safe_call(observer, "resource_event", filename=filename)

    def _emit_artifact_added(self, name, filename, metadata, content_type):
        for observer in self.observers:
            self._safe_call(
                observer,
                "artifact_event",
                name=name,
                filename=filename,
                metadata=metadata,
                content_type=content_type,
            )

    def _safe_call(self, obs, method, **kwargs):
        if obs not in self._failed_observers:
            try:
                getattr(obs, method)(**kwargs)
            except Exception as e:
                self._failed_observers.append(obs)
                self.run_logger.warning(
                    "An error ocurred in the '{}' " "observer: {}".format(obs, e)
                )

    def _final_call(self, observer, method, **kwargs):
        try:
            getattr(observer, method)(**kwargs)
        except Exception:
            # Feels dirty to catch all exceptions, but it is just for
            # finishing up, so we don't want one observer to kill the
            # others
            self.run_logger.error(tb.format_exc())

    def _wait_for_observers(self):
        """Block until all observers finished processing."""
        for observer in self.observers:
            self._safe_call(observer, "join")

    def _warn_about_failed_observers(self):
        for observer in self._failed_observers:
            self.run_logger.warning(
                "The observer '{}' failed at some point "
                "during the run.".format(observer)
            )

    def _execute_pre_run_hooks(self):
        for pr in self.pre_run_hooks:
            pr()

    def _execute_post_run_hooks(self):
        for pr in self.post_run_hooks:
            pr()

    def warn_if_unobserved(self):
        if not self.observers and not self.debug and not self.unobserved:
            self.run_logger.warning("No observers have been added to this run")

    def log_scalar(self, metric_name, value, step=None):
        """
        Add a new measurement.

        The measurement will be processed by the MongoDB observer
        during a heartbeat event.
        Other observers are not yet supported.

        :param metric_name: The name of the metric, e.g. training.loss
        :param value: The measured value
        :param step: The step number (integer), e.g. the iteration number
                    If not specified, an internal counter for each metric
                    is used, incremented by one.
        """
        # Method added in change https://github.com/chovanecm/sacred/issues/4
        # The same as Experiment.log_scalar (if something changes,
        # update the docstring too!)

        self._metrics.log_scalar_metric(metric_name, value, step) def convert_to_nested_dict(dotted_dict):
    """Convert a dict with dotted path keys to corresponding nested dict."""
    nested_dict = {}
    for k, v in iterate_flattened(dotted_dict):
        set_by_dotted_path(nested_dict, k, v)
    return nested_dict def create_basic_stream_logger():
    """Sets up a basic stream logger.

    Configures the root logger to use a
    `logging.StreamHandler` and sets the logging level to `logging.INFO`.

    Notes
    -----
        This does not change the logger configuration if the root logger
        already is configured (i.e. `len(getLogger().handlers) > 0`)
    """
    logging.basicConfig(
        level=logging.INFO, format="%(levelname)s - %(name)s - %(message)s"
    )
    return logging.getLogger("") def get_by_dotted_path(d, path, default=None):
    """
    Get an entry from nested dictionaries using a dotted path.

    Example
    -------
    >>> get_by_dotted_path({'foo': {'a': 12}}, 'foo.a')
    12
    """
    if not path:
        return d
    split_path = path.split(".")
    current_option = d
    for p in split_path:
        if p not in current_option:
            return default
        current_option = current_option[p]
    return current_option def is_prefix(pre_path, path):
    """Return True if pre_path is a path-prefix of path."""
    pre_path = pre_path.strip(".")
    path = path.strip(".")
    return not pre_path or path.startswith(pre_path + ".") def rel_path(base, path):
    """Return path relative to base."""
    if base == path:
        return ""
    assert is_prefix(base, path), "{} not a prefix of {}".format(base, path)
    return path[len(base) :].strip(".") def iterate_flattened(d):
    """
    Recursively iterate over the items of a dictionary.

    Provides a full dotted paths for every leaf.
    """
    for key in sorted(d.keys()):
        value = d[key]
        if isinstance(value, dict) and value:
            for k, v in iterate_flattened(d[key]):
                yield join_paths(key, k), v
        else:
            yield key, value def set_by_dotted_path(d, path, value):
    """
    Set an entry in a nested dict using a dotted path.

    Will create dictionaries as needed.

    Examples
    --------
    >>> d = {'foo': {'bar': 7}}
    >>> set_by_dotted_path(d, 'foo.bar', 10)
    >>> d
    {'foo': {'bar': 10}}
    >>> set_by_dotted_path(d, 'foo.d.baz', 3)
    >>> d
    {'foo': {'bar': 10, 'd': {'baz': 3}}}

    """
    split_path = path.split(".")
    current_option = d
    for p in split_path[:-1]:
        if p not in current_option:
            current_option[p] = dict()
        current_option = current_option[p]
    current_option[split_path[-1]] = value def recursive_update(d, u):
    """
    Given two dictionaries d and u, update dict d recursively.

    E.g.:
    d = {'a': {'b' : 1}}
    u = {'c': 2, 'a': {'d': 3}}
    => {'a': {'b': 1, 'd': 3}, 'c': 2}
    """
    for k, v in u.items():
        if isinstance(v, collections.abc.Mapping):
            r = recursive_update(d.get(k, {}), v)
            d[k] = r
        else:
            d[k] = u[k]
    return d def iter_prefixes(path):
    """
    Iterate through all (non-empty) prefixes of a dotted path.

    Example
    -------
    >>> list(iter_prefixes('foo.bar.baz'))
    ['foo', 'foo.bar', 'foo.bar.baz']
    """
    split_path = path.split(".")
    for i in range(1, len(split_path) + 1):
        yield join_paths(*split_path[:i]) def join_paths(*parts):
    """Join different parts together to a valid dotted path."""
    return ".".join(str(p).strip(".") for p in parts if p) class NamedConfigNotFoundError(SacredError):
    """A named config is not found."""

    def __init__(
        self,
        named_config,
        message="Named config not found:",
        available_named_configs=(),
        print_traceback=False,
        filter_traceback="default",
        print_usage=False,
    ):
        message = '{} "{}". Available config values are: {}'.format(
            message, named_config, available_named_configs
        )
        super().__init__(
            message,
            print_traceback=print_traceback,
            filter_traceback=filter_traceback,
            print_usage=print_usage,
        ) class ConfigAddedError(ConfigError):
    SPECIAL_ARGS = {"_log", "_config", "_seed", "__doc__", "config_filename", "_run"}
    """Special args that show up in the captured args but can never be set
    by the user"""

    def __init__(
        self,
        conflicting_configs,
        message="Added new config entry that is not used anywhere",
        captured_args=(),
        print_conflicting_configs=True,
        print_traceback=False,
        filter_traceback="default",
        print_usage=False,
        print_suggestions=True,
        config=None,
    ):
        super().__init__(
            message,
            conflicting_configs=conflicting_configs,
            print_conflicting_configs=print_conflicting_configs,
            print_traceback=print_traceback,
            filter_traceback=filter_traceback,
            print_usage=print_usage,
            config=config,
        )
        self.captured_args = captured_args
        self.print_suggestions = print_suggestions

    def __str__(self):
        s = super().__str__()
        if self.print_suggestions:
            possible_keys = set(self.captured_args) - self.SPECIAL_ARGS
            if possible_keys:
                s += "\nPossible config keys are: {}".format(possible_keys)
        return s SETTINGS = FrozenKeyMunch.fromDict(
    {
        "CONFIG": {
            # make sure all config keys are compatible with MongoDB
            "ENFORCE_KEYS_MONGO_COMPATIBLE": True,
            # make sure all config keys are serializable with jsonpickle
            # THIS IS IMPORTANT. Only deactivate if you know what you're doing.
            "ENFORCE_KEYS_JSONPICKLE_COMPATIBLE": True,
            # make sure all config keys are valid python identifiers
            "ENFORCE_VALID_PYTHON_IDENTIFIER_KEYS": False,
            # make sure all config keys are strings
            "ENFORCE_STRING_KEYS": False,
            # make sure no config key contains an equals sign
            "ENFORCE_KEYS_NO_EQUALS": True,
            # if true, all dicts and lists in the configuration of a captured
            # function are replaced with a read-only container that raises an
            # Exception if it is attempted to write to those containers
            "READ_ONLY_CONFIG": True,
            # regex patterns to filter out certain IDE or linter directives
            # from inline comments in the documentation
            "IGNORED_COMMENTS": ["^pylint:", "^noinspection"],
            # if true uses the numpy legacy API, i.e. _rnd in captured functions is
            # a numpy.random.RandomState rather than numpy.random.Generator.
            # numpy.random.RandomState became legacy with numpy v1.19.
            "NUMPY_RANDOM_LEGACY_API": version.parse(opt.np.__version__)
            < version.parse("1.19")
            if opt.has_numpy
            else False,
        },
        "HOST_INFO": {
            # Collect information about GPUs using the nvidia-smi tool
            "INCLUDE_GPU_INFO": True,
            # List of ENVIRONMENT variables to store in host-info
            "CAPTURED_ENV": [],
        },
        "COMMAND_LINE": {
            # disallow string fallback, if parsing a value from command-line
            # failed
            "STRICT_PARSING": False,
            # show command line options that are disabled (e.g. unmet
            # dependencies)
            "SHOW_DISABLED_OPTIONS": True,
        },
        # configure how stdout/stderr are captured. ['no', 'sys', 'fd']
        "CAPTURE_MODE": "sys" if platform.system() == "Windows" else "fd",
        # configure how dependencies are discovered. [none, imported, sys, pkg]
        "DISCOVER_DEPENDENCIES": "imported",
        # configure how source-files are discovered. [none, imported, sys, dir]
        "DISCOVER_SOURCES": "imported",
    },
)
import os
from collections import OrderedDict, defaultdict
from copy import copy, deepcopy
from sacred.config import (
    ConfigDict,
    chain_evaluate_config_scopes,
    dogmatize,
    load_config_file,
    undogmatize,
)
from sacred.config.config_summary import ConfigSummary
from sacred.config.custom_containers import make_read_only
from sacred.host_info import get_host_info
from sacred.randomness import create_rnd, get_seed
from sacred.run import Run
from sacred.utils import (
    convert_to_nested_dict,
    create_basic_stream_logger,
    get_by_dotted_path,
    is_prefix,
    rel_path,
    iterate_flattened,
    set_by_dotted_path,
    recursive_update,
    iter_prefixes,
    join_paths,
    NamedConfigNotFoundError,
    ConfigAddedError,
)
from sacred.settings import SETTINGS
#!/usr/bin/env python
# coding=utf-8




class Scaffold:
    def __init__(
        self,
        config_scopes,
        subrunners,
        path,
        captured_functions,
        commands,
        named_configs,
        config_hooks,
        generate_seed,
    ):
        self.config_scopes = config_scopes
        self.named_configs = named_configs
        self.subrunners = subrunners
        self.path = path
        self.generate_seed = generate_seed
        self.config_hooks = config_hooks
        self.config_updates = {}
        self.named_configs_to_use = []
        self.config = {}
        self.fallback = None
        self.presets = {}
        self.fixture = None  # TODO: rename
        self.logger = None
        self.seed = None
        self.rnd = None
        self._captured_functions = captured_functions
        self.commands = commands
        self.config_mods = None
        self.summaries = []
        self.captured_args = {
            join_paths(cf.prefix, n)
            for cf in self._captured_functions
            for n in cf.signature.arguments
        }
        self.captured_args.add("__doc__")  # allow setting the config docstring

    def set_up_seed(self, rnd=None):
        if self.seed is not None:
            return

        self.seed = self.config.get("seed")
        if self.seed is None:
            self.seed = get_seed(rnd)

        self.rnd = create_rnd(self.seed)

        if self.generate_seed:
            self.config["seed"] = self.seed

        if "seed" in self.config and "seed" in self.config_mods.added:
            self.config_mods.modified.add("seed")
            self.config_mods.added -= {"seed"}

        # Hierarchically set the seed of proper subrunners
        for subrunner_path, subrunner in reversed(list(self.subrunners.items())):
            if is_prefix(self.path, subrunner_path):
                subrunner.set_up_seed(self.rnd)

    def gather_fallbacks(self):
        fallback = {"_log": self.logger}
        for sr_path, subrunner in self.subrunners.items():
            if self.path and is_prefix(self.path, sr_path):
                path = sr_path[len(self.path) :].strip(".")
                set_by_dotted_path(fallback, path, subrunner.config)
            else:
                set_by_dotted_path(fallback, sr_path, subrunner.config)

        # dogmatize to make the subrunner configurations read-only
        self.fallback = dogmatize(fallback)
        self.fallback.revelation()

    def run_named_config(self, config_name):
        if os.path.isfile(config_name):
            nc = ConfigDict(load_config_file(config_name))
        else:
            if config_name not in self.named_configs:
                raise NamedConfigNotFoundError(
                    named_config=config_name,
                    available_named_configs=tuple(self.named_configs.keys()),
                )
            nc = self.named_configs[config_name]

        cfg = nc(
            fixed=self.get_config_updates_recursive(),
            preset=self.presets,
            fallback=self.fallback,
        )

        return undogmatize(cfg)

    def set_up_config(self):
        self.config, self.summaries = chain_evaluate_config_scopes(
            self.config_scopes,
            fixed=self.config_updates,
            preset=self.config,
            fallback=self.fallback,
        )

        self.get_config_modifications()

    def run_config_hooks(self, config, command_name, logger):
        final_cfg_updates = {}
        for ch in self.config_hooks:
            cfg_upup = ch(deepcopy(config), command_name, logger)
            if cfg_upup:

                recursive_update(final_cfg_updates, cfg_upup)