quast_decisiontree.core.runner

quast_decisiontree.core.runner

module_dir module-attribute

module_dir = Path(__file__).parent

default_config_path module-attribute

default_config_path = module_dir / 'standard_config.yaml'

global_log_path module-attribute

global_log_path = (
    module_dir.parent.parent.parent / "global.log"
).resolve()

logger module-attribute

logger = logging.getLogger('dt_logger')

handler module-attribute

handler = logging.StreamHandler(sys.stdout)

formatter module-attribute

formatter = logging.Formatter(
    fmt="%(asctime)s - %(levelname)s: %(message)s",
    datefmt="%d.%m.%y - %H:%M:%S",
)

filehandler module-attribute

filehandler = logging.handlers.RotatingFileHandler(
    global_log_path, mode="a", maxBytes=5 * 1024 * 1024
)

warnings_logger module-attribute

warnings_logger = logging.getLogger('py.warnings')

DecisionTree

used to create instances of decision trees through a yaml file. It provides the following attributes and methods:

• init(config): constructs a decision tree instance from the specified config (either as yaml path or python dict)
• validate(): validates the current tree setup
• run(problem_instance, config): runs the decisiontree. This replaces the functionality formerly contained in the run() function from decisiontree.interactive.main.

Source code in src/quast_decisiontree/core/runner.py

class DecisionTree:
    """used to create instances of decision trees through a yaml file. It provides the following
    attributes and methods:

    - __init__(config): constructs a decision tree instance from the specified config (either as
        yaml path or python dict)
    - validate(): validates the current tree setup
    - run(problem_instance, config): runs the decisiontree. This replaces the functionality
        formerly contained in the run() function from decisiontree.interactive.main.
    """

    _valid_requests = ["", "backends", "backend", "requires"]

    def __init__(
        self,
        config: dict | str | Path,
        validate_setup: bool = True,
        default_config: str = default_config_path,
        log_level: str | None = None,
    ) -> None:
        if isinstance(config, str | Path):
            self.source_file = config
            config_dict = load_config(config)
        elif isinstance(config, dict):
            config_dict = config
            self.source_file = "Custom dictionary"
        else:
            raise ValueError("Invalid configuration passed to DecisionTree constructor")
        self._logger = logger
        self.default_config = load_config(default_config)
        self._config = self.default_config.copy()
        self.config = config_dict  # only updates the _config dict
        # set environment variables so they can be retrieved by all classes
        os.environ["QDT_MODE"] = self.config["recommendation_mode"]
        self._mode = self.config["recommendation_mode"]

        if log_level in ["debug", "info", "warning", "error", "critical"]:
            self.config["log_level"] = log_level

        self.history = []
        self.set_log_level(self.config["log_level"])
        self.log("Configuration loaded, now constructing decision tree")
        self._nodes, self._children, self._root = import_tree(config_dict)
        self.set_request_funcs()
        self.backend_provider = load_backends(self.nodes)
        self.log(f"Succesfully loaded {len(self)} nodes into the decision tree.")

        self._graph = None  # only created on request
        if validate_setup:
            self.log("Validating tree setup.")
            if not self.validate():
                raise InvalidDecisionTreeSetupError(
                    "Validation for decision tree failed. To ignore this error, pass"
                    " validate_setup=False to the constructor"
                )
        else:
            self.log("Skipping validation.", log_level="debug")
        self.log("Decision Tree initialization complete.")

    @property
    def core_version(self):
        return importlib.metadata.version(self.__module__.split(".")[0])

    @property
    def logger(self):
        return self._logger

    def set_log_level(self, log_level: str):
        if log_level == "info":
            self.logger.setLevel(logging.INFO)
        elif log_level == "debug":
            self.logger.setLevel(logging.DEBUG)
        elif log_level == "warning":
            self.logger.setLevel(logging.WARNING)
        elif log_level == "error":
            self.logger.setLevel(logging.ERROR)
        elif log_level == "critical":
            self.logger.setLevel(logging.CRITICAL)
        else:
            raise ValueError(f"Unknown log level {log_level}.")

    @property
    def config(self):
        return self._config

    @config.setter
    def config(self, config_dict):
        self._config.update(config_dict)
        # update environment variables if necessary
        if "recommendation_mode" in config_dict:
            self._mode = config_dict["recommendation_mode"]
            os.environ["QDT_MODE"] = config_dict["recommendation_mode"]
        if "log_level" in config_dict:
            self.set_log_level(config_dict["log_level"])

    @property
    def mode(self):
        return self._mode

    @mode.setter
    def mode(self, mode: str):
        self.config = dict(recommendation_mode=mode)

    @property
    def root(self):
        return self._root

    @property
    def nodes(self):
        return self._nodes

    @property
    def children(self):
        return self._children

    @property
    def graph(self):
        if self._graph is None:
            self._graph = nx.DiGraph()
            for node, children in self.children.items():
                for child in children:
                    self._graph.add_edge(node, child)
        return self._graph

    @property
    def parents(self):
        parents = dict()
        for parent, children in self.children.items():
            for child in children:
                if child not in parents:
                    parents[child] = []
                parents[child].append(parent)
        # Ensure all nodes are included, even those without parents
        for node in self.graph:
            if node not in parents:
                parents[node] = []
        return parents

    def get_backend(self, name, filter):
        return self.backend_provider.get_backend(name, filter)

    def __len__(self):
        return len(self.nodes)

    def log(self, msg: str, log_level: str = "info"):
        """shortcut for logging"""
        if log_level == "info":
            self.logger.info(msg)
        elif log_level == "debug":
            self.logger.debug(msg)
        elif log_level == "warning":
            self.logger.warning(msg)
        elif log_level == "error":
            self.logger.error(msg)
        elif log_level == "critical":
            self.logger.critical(msg)

    def validate(self, strict=False) -> bool:
        """should be used to validate that the node setup is valid in terms of the problem data
        entries it creates
        """
        out_keys = {node: None for node in self.nodes.keys()}
        errors = []
        current_node_name = self.root
        out_keys[self.root] = set(self.nodes[self.root].creates)
        node_stack = deque()
        nodes_tested = set()
        missed_nodes = set()

        self.log("Starting validation of decision tree.")
        while len(nodes_tested) + len(missed_nodes) < len(self):
            self.logger.debug("Checking node %s", current_node_name)
            current_node = self.nodes[current_node_name]

            # issue warning for unknown child nodes
            if not current_node.check_children():
                self.logger.warning(
                    "There is more than one child to %s, and not all children are known.",
                    current_node.__class__.__name__,
                )
                self.logger.debug("Children: %s", current_node.children)
                self.logger.debug("Known children: %s", current_node._known_children)
                self.logger.warning("Path selection might not work.")

            out_keys[current_node_name] = out_keys[current_node_name] | set(current_node.creates)
            for child in current_node.children:
                self.logger.debug("Checking child node %s", child)
                self.logger.debug("Keys on current path: %s", out_keys[current_node_name])
                if strict:
                    out_keys[child] = intersect_if_not_none(
                        out_keys[child], out_keys[current_node_name]
                    )
                else:
                    out_keys[child] = union_if_not_none(
                        out_keys[child], out_keys[current_node_name]
                    )

                mismatches = self.nodes[child].violated_requirements(out_keys[current_node_name])
                if mismatches:
                    self.logger.debug("Mismatches found: %s", mismatches)
                    errors.append((current_node_name, child, mismatches))
            nodes_tested.add(current_node_name)
            for child in self.nodes[current_node_name].children:
                if all([parent in nodes_tested for parent in self.parents[child]]):
                    node_stack.append(child)
            # fetch next node to check
            while current_node_name in nodes_tested:  # fetch until a new node occurs
                try:
                    current_node_name = node_stack.popleft()
                except IndexError:  # node stack empty
                    if len(nodes_tested) == len(self):  # all good
                        break
                    else:
                        missed_nodes = set(self.nodes.keys()).difference(nodes_tested)
                        break
        if not all(
            self.nodes[node].silent for node in missed_nodes
        ):  # disconnected nodes must be silent
            raise DisconnectedDecisionTreeError(
                "It seems there are disconnected nodes in the decisiontree. Nodes"
                f" {set(self.nodes.keys()).difference(nodes_tested)} couldn't be reached."
            )
        elif missed_nodes:
            self.log(f"Silent nodes contained in decisiontree: {missed_nodes}", "debug")

        if errors:
            self.log("Validation failed, mismatches found:")
            for line in fmt_mismatches(errors):
                self.log(line)
            self.logger.debug("Outgoing keys:")
            for node in self.nodes:
                self.logger.debug("%s : %s", node, out_keys[node])
            return False
        else:
            self.log("Tree successfully validated")
            return True

    def update_config(self, config_dict: dict = None, config_path: str = None):
        """updates the configuration of a decisiontree on an existing instance

        The tree structure itself should be immutable, so only other keys can be udpated.
        """
        if config_path is not None:
            update_dict = load_config(config_path)
        else:
            update_dict = dict()
        if config_dict is not None:
            update_dict.update(config_dict)
        if "tree" in update_dict:
            self.logger.warning(
                "You cannot update the tree structure of an existing decision tree instance."
                " Create a new one instead. Ignoring the 'tree' key of the update config."
            )
            del update_dict["tree"]
        self.config = update_dict

    def _is_request_valid(self, request, **kwargs):
        """checks whether a request is valid"""
        if request not in self._valid_requests:
            return False
        else:
            if request == "backend" and "name" not in kwargs:
                return False
            if request == "requires" and "node" not in kwargs:
                return False
            else:
                return True
        return True

    def get_data(self, request: str | None, **kwargs):
        """function that serves as a mean to Nodes to retrieve information from the decisiontree

        Can be expanded in the future to implement access restrictions etc.. kwargs can be used to
        modify the output. The general type of information is specified in the "request" argument.

        Here's a list of possible requests, along with their kwargs:

        "backends"      will return a BackendProvider of available backends
        Optional kwargs:    filter      an filter passed on to self.backend_provider.filter

        "backend"       will return a single backend of the given name
        Required kwargs:    name        name string of the backend
        Optional kwargs:    filter      a filter passed to self.get_backend

        "requires"      will return the requirement specification of the node of the given name
        Required kwargs:    node        name of the node to fetch the requirements from

        """
        if not self._is_request_valid(request, **kwargs):
            raise ValueError(f"Request string {request!r} not valid with kwargs {kwargs}.")

        if request == "backend":
            return self.get_backend(kwargs["name"], kwargs.get("filter"))
        elif request == "backends":
            return self.backend_provider.filter(kwargs.get("filter"))
        elif request == "requires":
            return self.nodes[kwargs["node"]].requires

    def set_request_funcs(self):
        """set self.get_data as request_func to all nodes"""
        for node in self.nodes.values():
            node.request_info = self.get_data

    def add_run_log(self, remove_previous=False):
        while len(self.logger.handlers) > 2:  # should have only stream and global file handler
            self.logger.handlers.pop()

        logpath = os.path.join(self.config["experiment_folder"], "run.log")
        filehandler = logging.FileHandler(logpath)
        filehandler.setLevel(logging.DEBUG)
        formatter = logging.Formatter(
            fmt="%(asctime)s - %(levelname)s: %(message)s",
            datefmt="%d.%m.%y - %H:%M:%S",
        )
        filehandler.setFormatter(formatter)
        self.logger.addHandler(filehandler)

    def run(
        self, problem_instance=None, config_overwrite=None, name=None, config_path=None, path=None
    ) -> None:
        if config_overwrite is not None or config_path is not None:
            self.update_config(config_dict=config_overwrite, config_path=config_path)
        if name is None:
            name = datetime.now().strftime("%y%m%d_%H%M%S")

        self.config["experiment_folder"] = os.path.realpath(
            os.path.expanduser(os.path.join(self.config["data_folder"], name))
        )
        os.makedirs(self.config["experiment_folder"], exist_ok=True)
        os.environ["QDT_EXPERIMENT_FOLDER"] = self.config["experiment_folder"]
        self.history.append(self.config["experiment_folder"])

        self.add_run_log(remove_previous=True)

        self.log(f"Starting decision tree run with name {name}")
        self.logger.debug("Run-specific config changes: %s", config_overwrite)
        self.logger.debug("Run-specific config changes from path: %s", config_path)
        if path is None:
            path = dict()
        elif len(path) > 0:
            self.log("Received additional path info.")
            self.logger.debug("Additional path info: %s", path)

        if problem_instance is not None:
            path["LoadProblemNode"] = dict(problem_instance=problem_instance)

        if self.logger.getEffectiveLevel() < 15:  # only construct strings if on DEBUG
            self.logger.debug("Full configuration for current run:")
            self.logger.debug("\n".join(fmt_dictionary(self.config, title="Config")))

            self.logger.debug("Backends:")
            self.logger.debug("\n".join(f"{backend}" for backend in self.backend_provider))

        problem_data = DecisiontreeProblemData()

        current_node_name = self.root
        current_node = self.nodes[self.root]
        path_taken = [current_node_name]
        next_node_info = []
        while True:
            self.log(f"Current node: {current_node_name}")
            node_data = problem_data.extract(
                chain.from_iterable(current_node.requires), exclude="optional:"
            )
            self.logger.debug("Data passed to node: %s", node_data)

            node_path_info = path.get(current_node_name, dict())
            if len(node_path_info) > 0:
                self.logger.info("Path info passed to node: %s", node_path_info)
            else:
                self.logger.debug("Path info passed to node: %s", node_path_info)
            next_node_info.append(
                current_node.execute(problem_data=node_data, path_info=node_path_info)
            )
            self.logger.debug("Data after node execution: %s", node_data)
            path[current_node_name] = node_path_info
            problem_data.update(node_data)
            if current_node.final:
                self.logger.debug("Node is final, finishing execution loop.")
                break
            current_node_name = self.next_node(current_node_name, next_node_info[-1])
            current_node = self.nodes[current_node_name]
            path_taken.append(current_node_name)

        result = next_node_info[-1]
        self.log("Forward pass finished.")
        self.logger.debug("Result:\n %s", result)

        self.log("Starting backward pass.")
        for current_node_name, current_path_info in zip(
            path_taken[::-1], next_node_info[::-1], strict=False
        ):
            self.log(f"Current node: {current_node_name}")
            current_node = self.nodes[current_node_name]
            data = problem_data.extract(
                chain.from_iterable(current_node.requires), exclude="optional:"
            )
            self.logger.debug("Data passed to node: %s", data)
            try:
                result = current_node.interpret_result(
                    result, data, current_path_info, self.config
                )
            except Exception:
                self.logger.error(
                    "interpret_result for node %s failed; passing result through unchanged.",
                    current_node_name,
                    exc_info=True,
                )
            self.logger.debug("Result after node execution: %s", result)
        self.log("Backward pass finished.")

        # Save the config for the whole experiment
        self.log("Saving problem data, result and run configuration.")
        self.save_dt_data(result, problem_data, path)
        self.log(
            f"Saving complete. Files generated: {os.listdir(self.config['experiment_folder'])}"
        )
        self.log("Execution finished.")
        while len(self.logger.handlers) > 2:  # remove run-specific handlers
            self.logger.handlers.pop()

    def next_node(self, node_name: str, next_node_info: dict) -> str:
        if len(self.children[node_name]) == 1:
            self.logger.debug("Only one child available: %s", self.children[node_name][0])
            return self.children[node_name][0]
        else:
            self.logger.debug("Multiple children available: %s", self.children[node_name])
            self.logger.debug(
                "Next node: %s", self.nodes[node_name].next_node(next_node_info=next_node_info)
            )
            return self.nodes[node_name].next_node(next_node_info=next_node_info)

    def save_dt_data(
        self,
        result: dict,
        problem_data: dict,
        path: dict,
    ) -> None:
        static_no_save = set(self.config["no_save_keys"])

        def omit_keys(data: dict) -> set:
            """Static no-save keys, this dict's own ``_no_save`` marker, and the marker itself."""
            return static_no_save | set(data.get("_no_save", set())) | {"_no_save"}

        def log_unsaved(label: str, unsaved: list) -> None:
            if unsaved:
                self.logger.debug(
                    "%s: %s items couldn't be saved to dict: %s",
                    label,
                    len(unsaved),
                    unsaved,
                )

        log_unsaved(
            "Problem Data dict",
            save_dict(
                problem_data,
                self.config["experiment_folder"],
                "problem_data.json",
                no_save_keys=omit_keys(problem_data),
            ),
        )

        log_unsaved(
            "Result dict",
            save_dict(
                result,
                self.config["experiment_folder"],
                "result.json",
                no_save_keys=omit_keys(result),
            ),
        )

        log_unsaved(
            "Config dict",
            save_dict(
                self.config,
                self.config["experiment_folder"],
                "run_config.yaml",
                filetype="yaml",
                no_save_keys=omit_keys(self.config),
            ),
        )

        log_unsaved(
            "Path dict",
            save_dict(
                path,
                self.config["experiment_folder"],
                "path.yaml",
                filetype="yaml",
            ),
        )

    def show(self, **kwargs):
        """creates a figure displaying the decision tree in a tree layout.

        The kwargs will be passed directly to the update_layout() method of the resulting
        plotly figure. For possible adjustments, see the plotly documentation.
        """

        xpos_max = 100
        ypos_max = 100

        layer_height = 60
        node_width = 280

        # compute node positions
        layers = list(nx.topological_generations(self.graph))
        num_layers = len(layers)
        max_width = max(len(layer) for layer in layers)
        ypos = np.linspace(0, ypos_max, num=num_layers, endpoint=True)[::-1]
        xpos = [
            np.linspace(0, xpos_max, num=width, endpoint=True) for width in range(max_width + 1)
        ]
        xpos[1] = np.array([xpos_max / 2])
        pos = dict()
        for ind, layer in enumerate(layers):
            y = ypos[ind]
            for node_ind, node in enumerate(layer):
                x = xpos[len(layer)][node_ind]
                pos[node] = (x, y)
        im_height = num_layers * layer_height
        im_width = max_width * node_width

        # create edge trace
        edge_x = []
        edge_y = []
        for edge in self.graph.edges():
            x0, y0 = pos[edge[0]]
            x1, y1 = pos[edge[1]]
            edge_x += [x0, x1, None]
            edge_y += [y0, y1, None]
        edge_trace = go.Scatter(
            x=edge_x, y=edge_y, line=dict(width=1, color="Black"), hoverinfo="none", mode="lines"
        )

        # create node trace
        node_txt = list(pos.keys())
        node_x = [node[0] for node in pos.values()]
        node_y = [node[1] for node in pos.values()]

        node_trace = go.Scatter(
            x=node_x,
            y=node_y,
            mode="markers+text",
            hoverinfo="text",
            line_width=5,
            text=node_txt,
            textposition="top center",
            marker=dict(size=12, line=dict(width=2, color="Black"), color="Blue"),
            textfont=dict(size=16),
        )

        fig = go.Figure(data=[edge_trace, node_trace])
        fig.update_layout(
            annotations=[
                dict(
                    text=f"QuaST Decision Tree v{self.core_version}",
                    x=0,
                    y=0,
                    xref="paper",
                    yref="paper",
                    xanchor="left",
                    yanchor="bottom",
                    showarrow=False,
                    font=dict(size=12, color="grey"),
                ),
                dict(
                    text=f"Source File: {self.source_file.split(sep=os.sep)[-1]}",
                    x=1,
                    y=0,
                    xref="paper",
                    yref="paper",
                    xanchor="right",
                    yanchor="bottom",
                    showarrow=False,
                    font=dict(size=12, color="grey"),
                ),
            ]
        )
        fig.update_layout(
            showlegend=False,
            plot_bgcolor="rgba(0,0,0,0)",
            xaxis_visible=False,
            yaxis_visible=False,
            height=im_height,
            width=im_width,
            xaxis=dict(range=[-20, 120]),
        )
        fig.update_layout(**kwargs)
        return fig

    def generate_path_spec(self, path: str | Path | None = None) -> dict[str, dict[str, dict]]:
        """Collects the path-key specification of every node in the tree.

        Walks the nodes in topological order and gathers each node's ``_path_keys``
        (via :meth:`Node.path_spec`). The result maps node names to a
        ``{key: {dtype, options}}`` dictionary, mirroring the ``path.yaml`` layout but
        with type/option specs instead of concrete values. Nodes without path keys
        map to an empty dict.

        If ``path`` is given, the spec is written there as a YAML file; otherwise it is
        dumped as YAML to stdout.
        """
        spec: dict[str, dict[str, dict]] = {}
        for layer in nx.topological_generations(self.graph):
            for node_name in layer:
                spec[node_name] = self.nodes[node_name].path_spec()

        # include nodes not reachable in the graph (e.g. silent, disconnected)
        for node_name, node in self.nodes.items():
            spec.setdefault(node_name, node.path_spec())

        if path is None:
            print(yaml.safe_dump(spec, sort_keys=False, default_flow_style=False))
        else:
            path = Path(path)
            unsaved = save_dict(spec, str(path.parent), path.name, filetype="yaml")
            if unsaved:
                self.logger.debug(
                    "Path spec: %s items couldn't be saved: %s", len(unsaved), unsaved
                )
            self.log(f"Path spec written to {path!r}")

        return spec

source_file instance-attribute

source_file = config

default_config instance-attribute

default_config = load_config(default_config)

history instance-attribute

history = []

backend_provider instance-attribute

backend_provider = load_backends(self.nodes)

core_version property

core_version

logger property

logger

config property writable

config

mode property writable

mode

root property

root

nodes property

nodes

children property

children

graph property

graph

parents property

parents

__init__

__init__(
    config,
    validate_setup=True,
    default_config=default_config_path,
    log_level=None,
)

Source code in src/quast_decisiontree/core/runner.py

def __init__(
    self,
    config: dict | str | Path,
    validate_setup: bool = True,
    default_config: str = default_config_path,
    log_level: str | None = None,
) -> None:
    if isinstance(config, str | Path):
        self.source_file = config
        config_dict = load_config(config)
    elif isinstance(config, dict):
        config_dict = config
        self.source_file = "Custom dictionary"
    else:
        raise ValueError("Invalid configuration passed to DecisionTree constructor")
    self._logger = logger
    self.default_config = load_config(default_config)
    self._config = self.default_config.copy()
    self.config = config_dict  # only updates the _config dict
    # set environment variables so they can be retrieved by all classes
    os.environ["QDT_MODE"] = self.config["recommendation_mode"]
    self._mode = self.config["recommendation_mode"]

    if log_level in ["debug", "info", "warning", "error", "critical"]:
        self.config["log_level"] = log_level

    self.history = []
    self.set_log_level(self.config["log_level"])
    self.log("Configuration loaded, now constructing decision tree")
    self._nodes, self._children, self._root = import_tree(config_dict)
    self.set_request_funcs()
    self.backend_provider = load_backends(self.nodes)
    self.log(f"Succesfully loaded {len(self)} nodes into the decision tree.")

    self._graph = None  # only created on request
    if validate_setup:
        self.log("Validating tree setup.")
        if not self.validate():
            raise InvalidDecisionTreeSetupError(
                "Validation for decision tree failed. To ignore this error, pass"
                " validate_setup=False to the constructor"
            )
    else:
        self.log("Skipping validation.", log_level="debug")
    self.log("Decision Tree initialization complete.")

set_log_level

set_log_level(log_level)

Source code in src/quast_decisiontree/core/runner.py

def set_log_level(self, log_level: str):
    if log_level == "info":
        self.logger.setLevel(logging.INFO)
    elif log_level == "debug":
        self.logger.setLevel(logging.DEBUG)
    elif log_level == "warning":
        self.logger.setLevel(logging.WARNING)
    elif log_level == "error":
        self.logger.setLevel(logging.ERROR)
    elif log_level == "critical":
        self.logger.setLevel(logging.CRITICAL)
    else:
        raise ValueError(f"Unknown log level {log_level}.")

get_backend

get_backend(name, filter)

Source code in src/quast_decisiontree/core/runner.py

def get_backend(self, name, filter):
    return self.backend_provider.get_backend(name, filter)

__len__

__len__()

Source code in src/quast_decisiontree/core/runner.py

def __len__(self):
    return len(self.nodes)

log

log(msg, log_level='info')

shortcut for logging

Source code in src/quast_decisiontree/core/runner.py

def log(self, msg: str, log_level: str = "info"):
    """shortcut for logging"""
    if log_level == "info":
        self.logger.info(msg)
    elif log_level == "debug":
        self.logger.debug(msg)
    elif log_level == "warning":
        self.logger.warning(msg)
    elif log_level == "error":
        self.logger.error(msg)
    elif log_level == "critical":
        self.logger.critical(msg)

validate

validate(strict=False)

should be used to validate that the node setup is valid in terms of the problem data entries it creates

Source code in src/quast_decisiontree/core/runner.py

def validate(self, strict=False) -> bool:
    """should be used to validate that the node setup is valid in terms of the problem data
    entries it creates
    """
    out_keys = {node: None for node in self.nodes.keys()}
    errors = []
    current_node_name = self.root
    out_keys[self.root] = set(self.nodes[self.root].creates)
    node_stack = deque()
    nodes_tested = set()
    missed_nodes = set()

    self.log("Starting validation of decision tree.")
    while len(nodes_tested) + len(missed_nodes) < len(self):
        self.logger.debug("Checking node %s", current_node_name)
        current_node = self.nodes[current_node_name]

        # issue warning for unknown child nodes
        if not current_node.check_children():
            self.logger.warning(
                "There is more than one child to %s, and not all children are known.",
                current_node.__class__.__name__,
            )
            self.logger.debug("Children: %s", current_node.children)
            self.logger.debug("Known children: %s", current_node._known_children)
            self.logger.warning("Path selection might not work.")

        out_keys[current_node_name] = out_keys[current_node_name] | set(current_node.creates)
        for child in current_node.children:
            self.logger.debug("Checking child node %s", child)
            self.logger.debug("Keys on current path: %s", out_keys[current_node_name])
            if strict:
                out_keys[child] = intersect_if_not_none(
                    out_keys[child], out_keys[current_node_name]
                )
            else:
                out_keys[child] = union_if_not_none(
                    out_keys[child], out_keys[current_node_name]
                )

            mismatches = self.nodes[child].violated_requirements(out_keys[current_node_name])
            if mismatches:
                self.logger.debug("Mismatches found: %s", mismatches)
                errors.append((current_node_name, child, mismatches))
        nodes_tested.add(current_node_name)
        for child in self.nodes[current_node_name].children:
            if all([parent in nodes_tested for parent in self.parents[child]]):
                node_stack.append(child)
        # fetch next node to check
        while current_node_name in nodes_tested:  # fetch until a new node occurs
            try:
                current_node_name = node_stack.popleft()
            except IndexError:  # node stack empty
                if len(nodes_tested) == len(self):  # all good
                    break
                else:
                    missed_nodes = set(self.nodes.keys()).difference(nodes_tested)
                    break
    if not all(
        self.nodes[node].silent for node in missed_nodes
    ):  # disconnected nodes must be silent
        raise DisconnectedDecisionTreeError(
            "It seems there are disconnected nodes in the decisiontree. Nodes"
            f" {set(self.nodes.keys()).difference(nodes_tested)} couldn't be reached."
        )
    elif missed_nodes:
        self.log(f"Silent nodes contained in decisiontree: {missed_nodes}", "debug")

    if errors:
        self.log("Validation failed, mismatches found:")
        for line in fmt_mismatches(errors):
            self.log(line)
        self.logger.debug("Outgoing keys:")
        for node in self.nodes:
            self.logger.debug("%s : %s", node, out_keys[node])
        return False
    else:
        self.log("Tree successfully validated")
        return True

update_config

update_config(config_dict=None, config_path=None)

updates the configuration of a decisiontree on an existing instance

The tree structure itself should be immutable, so only other keys can be udpated.

Source code in src/quast_decisiontree/core/runner.py

def update_config(self, config_dict: dict = None, config_path: str = None):
    """updates the configuration of a decisiontree on an existing instance

    The tree structure itself should be immutable, so only other keys can be udpated.
    """
    if config_path is not None:
        update_dict = load_config(config_path)
    else:
        update_dict = dict()
    if config_dict is not None:
        update_dict.update(config_dict)
    if "tree" in update_dict:
        self.logger.warning(
            "You cannot update the tree structure of an existing decision tree instance."
            " Create a new one instead. Ignoring the 'tree' key of the update config."
        )
        del update_dict["tree"]
    self.config = update_dict

get_data

get_data(request, **kwargs)

function that serves as a mean to Nodes to retrieve information from the decisiontree

Can be expanded in the future to implement access restrictions etc.. kwargs can be used to modify the output. The general type of information is specified in the "request" argument.

Here's a list of possible requests, along with their kwargs:

"backends" will return a BackendProvider of available backends Optional kwargs: filter an filter passed on to self.backend_provider.filter

"backend" will return a single backend of the given name Required kwargs: name name string of the backend Optional kwargs: filter a filter passed to self.get_backend

"requires" will return the requirement specification of the node of the given name Required kwargs: node name of the node to fetch the requirements from

Source code in src/quast_decisiontree/core/runner.py

def get_data(self, request: str | None, **kwargs):
    """function that serves as a mean to Nodes to retrieve information from the decisiontree

    Can be expanded in the future to implement access restrictions etc.. kwargs can be used to
    modify the output. The general type of information is specified in the "request" argument.

    Here's a list of possible requests, along with their kwargs:

    "backends"      will return a BackendProvider of available backends
    Optional kwargs:    filter      an filter passed on to self.backend_provider.filter

    "backend"       will return a single backend of the given name
    Required kwargs:    name        name string of the backend
    Optional kwargs:    filter      a filter passed to self.get_backend

    "requires"      will return the requirement specification of the node of the given name
    Required kwargs:    node        name of the node to fetch the requirements from

    """
    if not self._is_request_valid(request, **kwargs):
        raise ValueError(f"Request string {request!r} not valid with kwargs {kwargs}.")

    if request == "backend":
        return self.get_backend(kwargs["name"], kwargs.get("filter"))
    elif request == "backends":
        return self.backend_provider.filter(kwargs.get("filter"))
    elif request == "requires":
        return self.nodes[kwargs["node"]].requires

set_request_funcs

set_request_funcs()

set self.get_data as request_func to all nodes

Source code in src/quast_decisiontree/core/runner.py

def set_request_funcs(self):
    """set self.get_data as request_func to all nodes"""
    for node in self.nodes.values():
        node.request_info = self.get_data

add_run_log

add_run_log(remove_previous=False)

Source code in src/quast_decisiontree/core/runner.py

def add_run_log(self, remove_previous=False):
    while len(self.logger.handlers) > 2:  # should have only stream and global file handler
        self.logger.handlers.pop()

    logpath = os.path.join(self.config["experiment_folder"], "run.log")
    filehandler = logging.FileHandler(logpath)
    filehandler.setLevel(logging.DEBUG)
    formatter = logging.Formatter(
        fmt="%(asctime)s - %(levelname)s: %(message)s",
        datefmt="%d.%m.%y - %H:%M:%S",
    )
    filehandler.setFormatter(formatter)
    self.logger.addHandler(filehandler)

run

run(
    problem_instance=None,
    config_overwrite=None,
    name=None,
    config_path=None,
    path=None,
)

Source code in src/quast_decisiontree/core/runner.py

def run(
    self, problem_instance=None, config_overwrite=None, name=None, config_path=None, path=None
) -> None:
    if config_overwrite is not None or config_path is not None:
        self.update_config(config_dict=config_overwrite, config_path=config_path)
    if name is None:
        name = datetime.now().strftime("%y%m%d_%H%M%S")

    self.config["experiment_folder"] = os.path.realpath(
        os.path.expanduser(os.path.join(self.config["data_folder"], name))
    )
    os.makedirs(self.config["experiment_folder"], exist_ok=True)
    os.environ["QDT_EXPERIMENT_FOLDER"] = self.config["experiment_folder"]
    self.history.append(self.config["experiment_folder"])

    self.add_run_log(remove_previous=True)

    self.log(f"Starting decision tree run with name {name}")
    self.logger.debug("Run-specific config changes: %s", config_overwrite)
    self.logger.debug("Run-specific config changes from path: %s", config_path)
    if path is None:
        path = dict()
    elif len(path) > 0:
        self.log("Received additional path info.")
        self.logger.debug("Additional path info: %s", path)

    if problem_instance is not None:
        path["LoadProblemNode"] = dict(problem_instance=problem_instance)

    if self.logger.getEffectiveLevel() < 15:  # only construct strings if on DEBUG
        self.logger.debug("Full configuration for current run:")
        self.logger.debug("\n".join(fmt_dictionary(self.config, title="Config")))

        self.logger.debug("Backends:")
        self.logger.debug("\n".join(f"{backend}" for backend in self.backend_provider))

    problem_data = DecisiontreeProblemData()

    current_node_name = self.root
    current_node = self.nodes[self.root]
    path_taken = [current_node_name]
    next_node_info = []
    while True:
        self.log(f"Current node: {current_node_name}")
        node_data = problem_data.extract(
            chain.from_iterable(current_node.requires), exclude="optional:"
        )
        self.logger.debug("Data passed to node: %s", node_data)

        node_path_info = path.get(current_node_name, dict())
        if len(node_path_info) > 0:
            self.logger.info("Path info passed to node: %s", node_path_info)
        else:
            self.logger.debug("Path info passed to node: %s", node_path_info)
        next_node_info.append(
            current_node.execute(problem_data=node_data, path_info=node_path_info)
        )
        self.logger.debug("Data after node execution: %s", node_data)
        path[current_node_name] = node_path_info
        problem_data.update(node_data)
        if current_node.final:
            self.logger.debug("Node is final, finishing execution loop.")
            break
        current_node_name = self.next_node(current_node_name, next_node_info[-1])
        current_node = self.nodes[current_node_name]
        path_taken.append(current_node_name)

    result = next_node_info[-1]
    self.log("Forward pass finished.")
    self.logger.debug("Result:\n %s", result)

    self.log("Starting backward pass.")
    for current_node_name, current_path_info in zip(
        path_taken[::-1], next_node_info[::-1], strict=False
    ):
        self.log(f"Current node: {current_node_name}")
        current_node = self.nodes[current_node_name]
        data = problem_data.extract(
            chain.from_iterable(current_node.requires), exclude="optional:"
        )
        self.logger.debug("Data passed to node: %s", data)
        try:
            result = current_node.interpret_result(
                result, data, current_path_info, self.config
            )
        except Exception:
            self.logger.error(
                "interpret_result for node %s failed; passing result through unchanged.",
                current_node_name,
                exc_info=True,
            )
        self.logger.debug("Result after node execution: %s", result)
    self.log("Backward pass finished.")

    # Save the config for the whole experiment
    self.log("Saving problem data, result and run configuration.")
    self.save_dt_data(result, problem_data, path)
    self.log(
        f"Saving complete. Files generated: {os.listdir(self.config['experiment_folder'])}"
    )
    self.log("Execution finished.")
    while len(self.logger.handlers) > 2:  # remove run-specific handlers
        self.logger.handlers.pop()

next_node

next_node(node_name, next_node_info)

Source code in src/quast_decisiontree/core/runner.py

def next_node(self, node_name: str, next_node_info: dict) -> str:
    if len(self.children[node_name]) == 1:
        self.logger.debug("Only one child available: %s", self.children[node_name][0])
        return self.children[node_name][0]
    else:
        self.logger.debug("Multiple children available: %s", self.children[node_name])
        self.logger.debug(
            "Next node: %s", self.nodes[node_name].next_node(next_node_info=next_node_info)
        )
        return self.nodes[node_name].next_node(next_node_info=next_node_info)

save_dt_data

save_dt_data(result, problem_data, path)

Source code in src/quast_decisiontree/core/runner.py

def save_dt_data(
    self,
    result: dict,
    problem_data: dict,
    path: dict,
) -> None:
    static_no_save = set(self.config["no_save_keys"])

    def omit_keys(data: dict) -> set:
        """Static no-save keys, this dict's own ``_no_save`` marker, and the marker itself."""
        return static_no_save | set(data.get("_no_save", set())) | {"_no_save"}

    def log_unsaved(label: str, unsaved: list) -> None:
        if unsaved:
            self.logger.debug(
                "%s: %s items couldn't be saved to dict: %s",
                label,
                len(unsaved),
                unsaved,
            )

    log_unsaved(
        "Problem Data dict",
        save_dict(
            problem_data,
            self.config["experiment_folder"],
            "problem_data.json",
            no_save_keys=omit_keys(problem_data),
        ),
    )

    log_unsaved(
        "Result dict",
        save_dict(
            result,
            self.config["experiment_folder"],
            "result.json",
            no_save_keys=omit_keys(result),
        ),
    )

    log_unsaved(
        "Config dict",
        save_dict(
            self.config,
            self.config["experiment_folder"],
            "run_config.yaml",
            filetype="yaml",
            no_save_keys=omit_keys(self.config),
        ),
    )

    log_unsaved(
        "Path dict",
        save_dict(
            path,
            self.config["experiment_folder"],
            "path.yaml",
            filetype="yaml",
        ),
    )

show

show(**kwargs)

creates a figure displaying the decision tree in a tree layout.

The kwargs will be passed directly to the update_layout() method of the resulting plotly figure. For possible adjustments, see the plotly documentation.

Source code in src/quast_decisiontree/core/runner.py

def show(self, **kwargs):
    """creates a figure displaying the decision tree in a tree layout.

    The kwargs will be passed directly to the update_layout() method of the resulting
    plotly figure. For possible adjustments, see the plotly documentation.
    """

    xpos_max = 100
    ypos_max = 100

    layer_height = 60
    node_width = 280

    # compute node positions
    layers = list(nx.topological_generations(self.graph))
    num_layers = len(layers)
    max_width = max(len(layer) for layer in layers)
    ypos = np.linspace(0, ypos_max, num=num_layers, endpoint=True)[::-1]
    xpos = [
        np.linspace(0, xpos_max, num=width, endpoint=True) for width in range(max_width + 1)
    ]
    xpos[1] = np.array([xpos_max / 2])
    pos = dict()
    for ind, layer in enumerate(layers):
        y = ypos[ind]
        for node_ind, node in enumerate(layer):
            x = xpos[len(layer)][node_ind]
            pos[node] = (x, y)
    im_height = num_layers * layer_height
    im_width = max_width * node_width

    # create edge trace
    edge_x = []
    edge_y = []
    for edge in self.graph.edges():
        x0, y0 = pos[edge[0]]
        x1, y1 = pos[edge[1]]
        edge_x += [x0, x1, None]
        edge_y += [y0, y1, None]
    edge_trace = go.Scatter(
        x=edge_x, y=edge_y, line=dict(width=1, color="Black"), hoverinfo="none", mode="lines"
    )

    # create node trace
    node_txt = list(pos.keys())
    node_x = [node[0] for node in pos.values()]
    node_y = [node[1] for node in pos.values()]

    node_trace = go.Scatter(
        x=node_x,
        y=node_y,
        mode="markers+text",
        hoverinfo="text",
        line_width=5,
        text=node_txt,
        textposition="top center",
        marker=dict(size=12, line=dict(width=2, color="Black"), color="Blue"),
        textfont=dict(size=16),
    )

    fig = go.Figure(data=[edge_trace, node_trace])
    fig.update_layout(
        annotations=[
            dict(
                text=f"QuaST Decision Tree v{self.core_version}",
                x=0,
                y=0,
                xref="paper",
                yref="paper",
                xanchor="left",
                yanchor="bottom",
                showarrow=False,
                font=dict(size=12, color="grey"),
            ),
            dict(
                text=f"Source File: {self.source_file.split(sep=os.sep)[-1]}",
                x=1,
                y=0,
                xref="paper",
                yref="paper",
                xanchor="right",
                yanchor="bottom",
                showarrow=False,
                font=dict(size=12, color="grey"),
            ),
        ]
    )
    fig.update_layout(
        showlegend=False,
        plot_bgcolor="rgba(0,0,0,0)",
        xaxis_visible=False,
        yaxis_visible=False,
        height=im_height,
        width=im_width,
        xaxis=dict(range=[-20, 120]),
    )
    fig.update_layout(**kwargs)
    return fig

generate_path_spec

generate_path_spec(path=None)

Collects the path-key specification of every node in the tree.

Walks the nodes in topological order and gathers each node's _path_keys (via :meth:Node.path_spec). The result maps node names to a {key: {dtype, options}} dictionary, mirroring the path.yaml layout but with type/option specs instead of concrete values. Nodes without path keys map to an empty dict.

If path is given, the spec is written there as a YAML file; otherwise it is dumped as YAML to stdout.

Source code in src/quast_decisiontree/core/runner.py

def generate_path_spec(self, path: str | Path | None = None) -> dict[str, dict[str, dict]]:
    """Collects the path-key specification of every node in the tree.

    Walks the nodes in topological order and gathers each node's ``_path_keys``
    (via :meth:`Node.path_spec`). The result maps node names to a
    ``{key: {dtype, options}}`` dictionary, mirroring the ``path.yaml`` layout but
    with type/option specs instead of concrete values. Nodes without path keys
    map to an empty dict.

    If ``path`` is given, the spec is written there as a YAML file; otherwise it is
    dumped as YAML to stdout.
    """
    spec: dict[str, dict[str, dict]] = {}
    for layer in nx.topological_generations(self.graph):
        for node_name in layer:
            spec[node_name] = self.nodes[node_name].path_spec()

    # include nodes not reachable in the graph (e.g. silent, disconnected)
    for node_name, node in self.nodes.items():
        spec.setdefault(node_name, node.path_spec())

    if path is None:
        print(yaml.safe_dump(spec, sort_keys=False, default_flow_style=False))
    else:
        path = Path(path)
        unsaved = save_dict(spec, str(path.parent), path.name, filetype="yaml")
        if unsaved:
            self.logger.debug(
                "Path spec: %s items couldn't be saved: %s", len(unsaved), unsaved
            )
        self.log(f"Path spec written to {path!r}")

    return spec

fmt_mismatches

fmt_mismatches(mismatches)

Source code in src/quast_decisiontree/core/runner.py

def fmt_mismatches(mismatches):
    out = []
    for in_node, out_node, mismatch in mismatches:
        if len(mismatch) == 1:
            out.append(f"Edge {in_node}, {out_node}: Missing one of keys {mismatch[0]}")
        else:
            out.append(f"Edge {in_node}, {out_node}: Missing one key in each of the sets:")
            for mis in mismatch:
                out.append(mis)
    return out

intersect_if_not_none

intersect_if_not_none(keys, new_keys)

Source code in src/quast_decisiontree/core/runner.py

def intersect_if_not_none(keys, new_keys):
    if keys is None:
        return new_keys
    else:
        return set(keys) & set(new_keys)

union_if_not_none

union_if_not_none(keys, new_keys)

Source code in src/quast_decisiontree/core/runner.py

def union_if_not_none(keys, new_keys):
    if keys is None:
        return new_keys
    else:
        return set(keys) | set(new_keys)

load_backends

load_backends(nodes)

checks whether the nodes define backends, and if so, puts them into a BackendProvider

Source code in src/quast_decisiontree/core/runner.py

def load_backends(nodes) -> BackendProvider:
    """checks whether the nodes define backends, and if so, puts them into
    a BackendProvider
    """
    out = BackendProvider()
    for node in nodes.values():
        if node.provides_backend:
            backends = node.get_backends()
            out.extend(backends)
    return out

import_tree

import_tree(config)

Source code in src/quast_decisiontree/core/runner.py

def import_tree(config: dict) -> tuple[dict[str, Node], dict[str, list[str]]]:
    try:
        tree = config["tree"]
    except KeyError as exc:
        raise InvalidConfigValueError from exc

    # extract the required information from the config
    children = _children_dict(tree)  # childrens of the nodes
    init_args = _get_init_args(tree)  # additional initialization kwargs for the nodes
    package_list = _get_pkg_source(config)  # packages where nodes are contained
    root_node = _find_root(tree)

    # do the actual import and instantiation, fingers crossed
    node_names = list(tree.keys())
    nodes = _import_nodes(package_list, node_names, children, init_args)

    return nodes, children, root_node