Tsp report

cornflow-dags/DAG/activate_dags.py

@@ -1,14 +1,14 @@
 import cornflow_client.airflow.dag_utilities as utils
-from airflow import DAG
 from airflow.operators.python import PythonOperator
 from airflow.secrets.environment_variables import EnvironmentVariablesBackend
+from airflow.decorators import dag, task
+from airflow.models.baseoperator import chain
+from airflow.exceptions import AirflowSkipException
 
 from update_all_schemas import get_new_apps
 
 
 def create_dag(app):
-    def solve(**kwargs):
-        return utils.cf_solve_app(app, EnvironmentVariablesBackend(), **kwargs)
 
     if app.default_args is not None:
         default_args = app.default_args
@@ -19,26 +19,44 @@ def solve(**kwargs):
     if app.extra_args is not None:
         kwargs = app.extra_args
 
-    dag = DAG(
+    @dag(
         app.name,
         description=app.description,
         default_args=default_args,
         schedule_interval=None,
         tags=["model"],
         **kwargs
     )
-    with dag:
-        notify = getattr(app, "notify", True)
-        if not notify:
-            t1 = PythonOperator(task_id=app.name, python_callable=solve)
-        else:
-            t1 = PythonOperator(
-                task_id=app.name,
-                python_callable=solve,
-                on_failure_callback=utils.callback_email,
+    def taskflow_dag():
+        @task
+        def solve_app():
+            def solve(**kwargs):
+                return utils.cf_solve_app(app, EnvironmentVariablesBackend(), **kwargs)
+
+            notify = getattr(app, "notify", True)
+            if not notify:
+                return PythonOperator(task_id=app.name, python_callable=solve)
+            else:
+                return PythonOperator(
+                    task_id=app.name,
+                    python_callable=solve,
+                    on_failure_callback=utils.callback_email,
+                )
+
+        @task
+        def run_report():
+            def run(**kwargs):
+                return utils.cf_report(app, EnvironmentVariablesBackend(), **kwargs)
+
+            file_name = PythonOperator(
+                task_id=app.name + "_report", python_callable=run
             )
+            return dict(file_name=file_name)
 
-    return dag
+        # Define dependencies and call task functions
+        chain(solve_app(), run_report())
+
+    return taskflow_dag
 
 
 for app in get_new_apps():

cornflow-dags/DAG/tsp/core/experiment.py

@@ -6,12 +6,35 @@
 from .instance import Instance
 from .solution import Solution
 
+import json, tempfile
+from quarto import render
+
 
 class Experiment(ExperimentCore):
     schema_checks = load_json(
         os.path.join(os.path.dirname(__file__), "../schemas/solution_checks.json")
     )
 
+    def to_dict(self) -> dict:
+        return dict(instance=self.instance.to_dict(), solution=self.solution.to_dict())
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        return cls(
+            Instance.from_dict(data["instance"]), Solution.from_dict(data["solution"])
+        )
+
+    @classmethod
+    def from_json(cls, path: str) -> "Experiment":
+        with open(path, "r") as f:
+            data_json = json.load(f)
+        return cls.from_dict(data_json)
+
+    def to_json(self, path: str) -> None:
+        data = self.to_dict()
+        with open(path, "w") as f:
+            json.dump(data, f, indent=4, sort_keys=True)
+
     @property
     def instance(self) -> Instance:
         return super().instance
@@ -25,31 +48,52 @@ def solution(self, value):
         self._solution = value
 
     def get_objective(self) -> float:
+        # if solution is empty, we return 0
+        if len(self.solution.data["route"]) == 0:
+            return 0
         # we get a sorted list of nodes by position
-        route = (
-            TupList(self.solution.data["route"])
-            .sorted(key=lambda v: v["pos"])
-            .vapply(lambda v: v["node"])
-        )
-        weight = {(el["n1"], el["n2"]): el["w"] for el in self.instance.data["arcs"]}
-        # we sum all arcs in the solution
-        return (
-            sum([weight[n1, n2] for n1, n2 in zip(route, route[1:])])
-            + weight[route[-1], route[0]]
-        )
+        arcs = self.solution.get_used_arcs()
+
+        # we sum all arc weights in the solution
+        return sum(self.get_used_arc_weights().values())
+
+    def get_used_arc_weights(self) -> dict:
+        arcs = self.solution.get_used_arcs()
+        weight = self.instance.get_indexed_arcs()
+        return arcs.to_dict(None).kapply(lambda k: weight[k]["w"])
 
     def check_missing_nodes(self):
         nodes_in = TupList(v["n1"] for v in self.instance.data["arcs"]).to_set()
         nodes_out = TupList(n["node"] for n in self.solution.data["route"]).to_set()
-        return [{"node": n} for n in (nodes_in - nodes_out)]
+        return TupList({"node": n} for n in (nodes_in - nodes_out))
 
     def check_missing_positions(self):
         nodes_in = TupList(v["n1"] for v in self.instance.data["arcs"]).to_set()
         positions = TupList(n["pos"] for n in self.solution.data["route"]).to_set()
-        return [{"position": p} for p in set(range(len(nodes_in))) - positions]
+        return TupList({"position": p} for p in set(range(len(nodes_in))) - positions)
 
     def check_solution(self, *args, **kwargs) -> SuperDict:
         return SuperDict(
             missing_nodes=self.check_missing_nodes(),
             missing_positions=self.check_missing_positions(),
         )
+
+    def generate_report(self, report_path: str, report_name="report") -> None:
+        # a user may give the full "report.qmd" name.
+        # We want to take out the extension
+        report_base = os.path.splitext(report_name)[0]
+        path_without_ext = os.path.join(
+            os.path.dirname(__file__), "../report/", report_base
+        )
+        path_to_qmd = path_without_ext + ".qmd"
+        path_to_output = path_without_ext + ".html"
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "experiment.json")
+            # write a json with instance and solution to temp file
+            self.to_json(path)
+            # pass the path to the report to render
+            # it generates a report with path = path_to_output
+            render(input=path_to_qmd, execute_params=dict(file_name=path))
+        # quarto always writes the report in the .qmd directory.
+        # thus, we need to move it where we want to:
+        os.replace(path_to_output, report_path)

cornflow-dags/DAG/tsp/core/instance.py

@@ -4,23 +4,24 @@
 from cornflow_client import InstanceCore, get_empty_schema
 from cornflow_client.core.tools import load_json
 from pytups import TupList, SuperDict
+import networkx as nx
 
 
 class Instance(InstanceCore):
     schema = load_json(os.path.join(os.path.dirname(__file__), "../schemas/input.json"))
     schema_checks = get_empty_schema()
 
-    def __init__(self, data):
+    def __init__(self, data: dict):
         data = SuperDict(data)
         data["arcs"] = TupList(data["arcs"])
         super().__init__(data)
 
     @classmethod
-    def from_tsplib_file(cls, path):
+    def from_tsplib_file(cls, path: str):
         return cls.from_tsplib95(tsp.load(path))
 
     @classmethod
-    def from_tsplib95(cls, problem):
+    def from_tsplib95(cls, problem: tsp.models.StandardProblem):
         nodes = list(problem.get_nodes())
         edge_to_dict = lambda e: dict(
             n1=nodes[e[0]], n2=nodes[e[1]], w=problem.get_weight(*e)
@@ -29,7 +30,7 @@ def from_tsplib95(cls, problem):
         return cls(dict(arcs=arcs))
 
     def to_tsplib95(self):
-        arcs = TupList(self.data["arcs"])
+        arcs = TupList(self.get_arcs())
         nodes = (arcs.take("n1") + arcs.take("n2")).unique()
         pos = {k: v for v, k in enumerate(nodes)}
         arc_dict = arcs.to_dict(
@@ -57,7 +58,27 @@ def to_tsplib95(self):
             edge_weight_format=edge_weight_format,
             edge_weights=arc_weights,
         )
+
         return tsp.models.StandardProblem(**dict_data)
 
     def get_arcs(self) -> TupList:
         return self.data["arcs"]
+
+    def get_indexed_arcs(self) -> TupList:
+        return self.data["arcs"].to_dict(
+            result_col=None, indices=["n1", "n2"], is_list=False
+        )
+
+    def get_nodes(self) -> TupList:
+        arcs = self.get_arcs()
+        return (arcs.take("n1") + arcs.take("n2")).unique()
+
+    def get_graph(self) -> nx.Graph:
+        nodes = self.get_nodes()
+        arcs = self.get_arcs()
+        G = nx.DiGraph()
+        for node in nodes:
+            G.add_node(node)
+        for arc in arcs:
+            G.add_edge(arc["n1"], arc["n2"], weight=arc["w"])
+        return G

cornflow-dags/DAG/tsp/core/solution.py

@@ -1,9 +1,25 @@
 import os
 from cornflow_client import SolutionCore
 from cornflow_client.core.tools import load_json
+import pytups as pt
 
 
 class Solution(SolutionCore):
     schema = load_json(
         os.path.join(os.path.dirname(__file__), "../schemas/output.json")
     )
+
+    def get_route(self):
+        return self.data["route"]
+
+    def get_tour(self):
+        return pt.TupList(self.get_route()).sorted(key=lambda v: v["pos"]).take("node")
+
+    def get_used_arcs(self):
+        tour = self.get_tour()
+
+        if len(tour) <= 1:
+            return pt.TupList()
+        edges = pt.TupList(zip(tour, tour[1:]))
+        edges.append((tour[-1], tour[0]))
+        return edges

cornflow-dags/DAG/tsp/report/.gitignore

@@ -0,0 +1 @@
+/.quarto/