Skip to content

quast_decisiontree.nodes.algorithm_execute

quast_decisiontree.nodes.algorithm_execute

logger module-attribute 

logger = logging.getLogger('dt_logger')

ClassicalAlgorithmExecuteNode

Bases: FinalNode

Executes a ClassicalAlgorithm instance after it has been set up.

Expects problem_data to contain
  • solver: a ClassicalAlgorithm instance (e.g. BruteForce, TabuSolver)
  • solver_input (or qubo_matrix as fallback): the optimization problem (typically a QUBO matrix) passed to execute()

The node uses the generic ClassicalAlgorithm.execute(opt_problem) interface, making it compatible with any ClassicalAlgorithm subclass.

Modifications at runtime: - execute: - True : executes algorithm without confirmation - False : prompts the user for confirmation before execution

Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class ClassicalAlgorithmExecuteNode(FinalNode):
    """Executes a ClassicalAlgorithm instance after it has been set up.

    Expects problem_data to contain:
        - solver: a ClassicalAlgorithm instance (e.g. BruteForce, TabuSolver)
        - solver_input (or qubo_matrix as fallback): the optimization problem
          (typically a QUBO matrix) passed to execute()

    The node uses the generic ClassicalAlgorithm.execute(opt_problem) interface,
    making it compatible with any ClassicalAlgorithm subclass.

    Modifications at runtime:
    - execute:
        - True : executes algorithm without confirmation
        - False : prompts the user for confirmation before execution
    """

    _known_children = []
    _path_keys = dict(execute=PathKey(bool))

    def __init__(self, children: list, auto_execute: bool = False) -> None:
        if len(children) > 0:
            logger.warning("Cannot create FinalNode with children - argument will be ignored.")
        super().__init__(
            requires=[["solver_input", "qubo_matrix"], ["solver"]],
            creates=["solved"],
            children=[],
        )
        self.query = MultiChoiceQuery(
            question="Do you want to start execution of the algorithm now?",
            answers=dict(yes="Yes", no="No"),
            name="execute",
        )
        self.auto_execute = auto_execute

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        execute = path_info.get("execute", self.auto_execute)
        self.query.set_default("yes" if execute else "no")
        answer = self.query.input()
        path_info["execute"] = execute

        if answer == "yes":
            algorithm = problem_data["solver"]
            opt_problem = problem_data.get("solver_input")
            if opt_problem is None:
                opt_problem = problem_data["qubo_matrix"]

            if not algorithm.check_input(opt_problem):
                raise ValueError(
                    f"Cannot execute {algorithm.__class__.__name__}: "
                    f"invalid input for this algorithm."
                )

            logger.info("Executing %s.", algorithm.__class__.__name__)

            start = datetime.now()
            raw_result = algorithm.execute(opt_problem)
            elapsed = (datetime.now() - start).total_seconds()

            result = dict(raw=raw_result, total_solver_time=elapsed)
            problem_data["solved"] = True

            logger.info(
                "%s execution completed in %.3f seconds.",
                algorithm.__class__.__name__,
                elapsed,
            )

        elif answer == "no":
            problem_data["solved"] = False
            result = dict(raw=None, total_solver_time=0.0)
            logger.info("Execution cancelled by user.")

        else:
            problem_data["solved"] = False
            result = dict(raw=None, total_solver_time=0.0)
            logger.warning("Invalid answer %r — algorithm not executed.", answer)

        return result

query instance-attribute 

query = MultiChoiceQuery(
    question="Do you want to start execution of the algorithm now?",
    answers=dict(yes="Yes", no="No"),
    name="execute",
)

auto_execute instance-attribute 

auto_execute = auto_execute

__init__ 

__init__(children, auto_execute=False)
Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(self, children: list, auto_execute: bool = False) -> None:
    if len(children) > 0:
        logger.warning("Cannot create FinalNode with children - argument will be ignored.")
    super().__init__(
        requires=[["solver_input", "qubo_matrix"], ["solver"]],
        creates=["solved"],
        children=[],
    )
    self.query = MultiChoiceQuery(
        question="Do you want to start execution of the algorithm now?",
        answers=dict(yes="Yes", no="No"),
        name="execute",
    )
    self.auto_execute = auto_execute

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def execute(self, problem_data: dict, path_info: dict) -> dict:
    execute = path_info.get("execute", self.auto_execute)
    self.query.set_default("yes" if execute else "no")
    answer = self.query.input()
    path_info["execute"] = execute

    if answer == "yes":
        algorithm = problem_data["solver"]
        opt_problem = problem_data.get("solver_input")
        if opt_problem is None:
            opt_problem = problem_data["qubo_matrix"]

        if not algorithm.check_input(opt_problem):
            raise ValueError(
                f"Cannot execute {algorithm.__class__.__name__}: "
                f"invalid input for this algorithm."
            )

        logger.info("Executing %s.", algorithm.__class__.__name__)

        start = datetime.now()
        raw_result = algorithm.execute(opt_problem)
        elapsed = (datetime.now() - start).total_seconds()

        result = dict(raw=raw_result, total_solver_time=elapsed)
        problem_data["solved"] = True

        logger.info(
            "%s execution completed in %.3f seconds.",
            algorithm.__class__.__name__,
            elapsed,
        )

    elif answer == "no":
        problem_data["solved"] = False
        result = dict(raw=None, total_solver_time=0.0)
        logger.info("Execution cancelled by user.")

    else:
        problem_data["solved"] = False
        result = dict(raw=None, total_solver_time=0.0)
        logger.warning("Invalid answer %r — algorithm not executed.", answer)

    return result

HybridAlgorithmExecuteNode

Bases: FinalNode

Executes a HybridAlgorithm instance after it has been set up.

Expects problem_data to contain
  • hybrid_algorithm: a HybridAlgorithm instance (e.g. QrispQAOA)
  • hybrid_inputs: dict mapping INPUT_KEYS to their values
  • backend: the Backend to run on

The node uses the generic HybridAlgorithm.execute(backend, **inputs) interface, making it compatible with any HybridAlgorithm subclass.

Modifications at runtime: - execute: - True : executes algorithm without confirmation - False : prompts the user for confirmation before execution

Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class HybridAlgorithmExecuteNode(FinalNode):
    """Executes a HybridAlgorithm instance after it has been set up.

    Expects problem_data to contain:
        - hybrid_algorithm: a HybridAlgorithm instance (e.g. QrispQAOA)
        - hybrid_inputs: dict mapping INPUT_KEYS to their values
        - backend: the Backend to run on

    The node uses the generic HybridAlgorithm.execute(backend, **inputs) interface,
    making it compatible with any HybridAlgorithm subclass.

    Modifications at runtime:
    - execute:
        - True : executes algorithm without confirmation
        - False : prompts the user for confirmation before execution
    """

    _known_children = []
    _path_keys = dict(execute=PathKey(bool))

    def __init__(self, children: list, auto_execute: bool = False) -> None:
        if len(children) > 0:
            logger.warning("Cannot create FinalNode with children - argument will be ignored.")
        super().__init__(
            requires=["hybrid_algorithm", "hybrid_inputs", "backend"],
            creates=["solved"],
            children=[],
        )
        self.query = MultiChoiceQuery(
            question="Do you want to start execution of the algorithm now?",
            answers=dict(yes="Yes", no="No"),
            name="execute",
        )
        self.auto_execute = auto_execute

    def execute(self, problem_data: dict, path_info: dict) -> dict:
        execute = path_info.get("execute", self.auto_execute)
        self.query.set_default("yes" if execute else "no")
        answer = self.query.input()
        path_info["execute"] = execute

        if answer == "yes":
            algorithm = problem_data["hybrid_algorithm"]
            backend = problem_data["backend"]
            inputs = problem_data["hybrid_inputs"]

            missing = [
                key for key in algorithm.INPUT_KEYS if key not in inputs or inputs[key] is None
            ]
            if missing:
                raise ValueError(
                    f"Cannot execute {algorithm.__class__.__name__}: "
                    f"missing required inputs {missing} in hybrid_inputs."
                )

            logger.info(
                "Executing %s with %d input keys on backend.",
                algorithm.__class__.__name__,
                len(inputs),
            )

            start = datetime.now()
            raw_result = algorithm.execute(backend, **inputs)
            elapsed = (datetime.now() - start).total_seconds()

            result = dict(raw=raw_result, total_solver_time=elapsed)
            problem_data["solved"] = True

            logger.info(
                "%s execution completed in %.3f seconds.",
                algorithm.__class__.__name__,
                elapsed,
            )

        elif answer == "no":
            problem_data["solved"] = False
            result = dict(raw=None, total_solver_time=0.0)
            logger.info("Execution cancelled by user.")

        else:
            problem_data["solved"] = False
            result = dict(raw=None, total_solver_time=0.0)
            logger.warning("Invalid answer %r — algorithm not executed.", answer)

        return result

query instance-attribute 

query = MultiChoiceQuery(
    question="Do you want to start execution of the algorithm now?",
    answers=dict(yes="Yes", no="No"),
    name="execute",
)

auto_execute instance-attribute 

auto_execute = auto_execute

__init__ 

__init__(children, auto_execute=False)
Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def __init__(self, children: list, auto_execute: bool = False) -> None:
    if len(children) > 0:
        logger.warning("Cannot create FinalNode with children - argument will be ignored.")
    super().__init__(
        requires=["hybrid_algorithm", "hybrid_inputs", "backend"],
        creates=["solved"],
        children=[],
    )
    self.query = MultiChoiceQuery(
        question="Do you want to start execution of the algorithm now?",
        answers=dict(yes="Yes", no="No"),
        name="execute",
    )
    self.auto_execute = auto_execute

execute 

execute(problem_data, path_info)
Source code in src/quast_decisiontree/nodes/algorithm_execute.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def execute(self, problem_data: dict, path_info: dict) -> dict:
    execute = path_info.get("execute", self.auto_execute)
    self.query.set_default("yes" if execute else "no")
    answer = self.query.input()
    path_info["execute"] = execute

    if answer == "yes":
        algorithm = problem_data["hybrid_algorithm"]
        backend = problem_data["backend"]
        inputs = problem_data["hybrid_inputs"]

        missing = [
            key for key in algorithm.INPUT_KEYS if key not in inputs or inputs[key] is None
        ]
        if missing:
            raise ValueError(
                f"Cannot execute {algorithm.__class__.__name__}: "
                f"missing required inputs {missing} in hybrid_inputs."
            )

        logger.info(
            "Executing %s with %d input keys on backend.",
            algorithm.__class__.__name__,
            len(inputs),
        )

        start = datetime.now()
        raw_result = algorithm.execute(backend, **inputs)
        elapsed = (datetime.now() - start).total_seconds()

        result = dict(raw=raw_result, total_solver_time=elapsed)
        problem_data["solved"] = True

        logger.info(
            "%s execution completed in %.3f seconds.",
            algorithm.__class__.__name__,
            elapsed,
        )

    elif answer == "no":
        problem_data["solved"] = False
        result = dict(raw=None, total_solver_time=0.0)
        logger.info("Execution cancelled by user.")

    else:
        problem_data["solved"] = False
        result = dict(raw=None, total_solver_time=0.0)
        logger.warning("Invalid answer %r — algorithm not executed.", answer)

    return result