Add automatic dataflow saving when working in KPM

Signed-off-by: bbrzyski <[email protected]>

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Support Python 3.13 and dropped support for Python 3.8
 - Added to config option for choosing a location where KPM server should be built
 - All in one command `topwrap gui` for building, starting the KPM server and connecting the client to it.
+- Automatic dataflow saving with each change to the graph in KPM. This ensures that the state is preserved when the page is reloaded.
 
 ### Changed
 

docs/source/advanced_options.md

@@ -31,6 +31,11 @@ An example block design in the Topwrap GUI for the PWM project may look like thi
 :dataflow: ../build/kpm_jsons/data_pwm.json
 ```
 
+:::{important}
+With each graph change, Topwrap will save the current dataflow to ensure it's not lost, e.g. during an accidental page refresh.
+The file is located at `$XDG_DATA_HOME/topwrap/dataflow_latest_save.json`.
+:::
+
 More information about this example can be found [here](https://antmicro.github.io/topwrap/examples.html#pwm)
 
 ## Command Line Interface (CLI)

tests/data/data_kpm/conversions/complex/specification_complex.json

@@ -55,6 +55,7 @@
         "stopName": "Stop"
       }
     ],
+    "notifyWhenChanged": true,
     "twoColumn": true
   },
   "nodes": [

tests/data/data_kpm/examples/hdmi/specification_hdmi.json

@@ -75,6 +75,7 @@
         "stopName": "Stop"
       }
     ],
+    "notifyWhenChanged": true,
     "twoColumn": true
   },
   "nodes": [

tests/data/data_kpm/examples/hierarchy/specification_hierarchy.json

@@ -55,6 +55,7 @@
         "stopName": "Stop"
       }
     ],
+    "notifyWhenChanged": true,
     "twoColumn": true
   },
   "nodes": [

tests/data/data_kpm/examples/pwm/specification_pwm.json

@@ -65,6 +65,7 @@
         "stopName": "Stop"
       }
     ],
+    "notifyWhenChanged": true,
     "twoColumn": true
   },
   "nodes": [

tests/tests_kpm/common.py

@@ -1,23 +1,7 @@
 # Copyright (c) 2023-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
 
-import json
-from pathlib import Path
-
-from topwrap.util import JsonType
-
 AXI_NAME = "axi_bridge"
 PS7_NAME = "ps7"
 PWM_NAME = "litex_pwm_top"
 TEST_DATA_PATH = "tests/data/data_kpm/"
-
-
-def read_json_file(json_file_path: Path) -> JsonType:
-    with open(json_file_path, "r") as json_file:
-        json_contents = json.load(json_file)
-    return json_contents
-
-
-def save_file_to_json(file_path: Path, file_name: str, file_content: JsonType):
-    with open(Path(file_path / file_name), "w") as json_file:
-        json.dump(file_content, json_file)

tests/tests_kpm/conftest.py

@@ -8,9 +8,7 @@
 import pytest
 
 from topwrap.design import DesignDescription
-from topwrap.util import JsonType
-
-from .common import read_json_file
+from topwrap.util import JsonType, read_json_file
 
 
 def pwm_ipcores_yamls_data() -> List[Path]:

tests/tests_kpm/test_kpm_specification.py

@@ -9,10 +9,9 @@
 from referencing import Registry
 from referencing.jsonschema import DRAFT201909
 
+from topwrap.util import read_json_file
 from topwrap.yamls_to_kpm_spec_parser import ipcore_yamls_to_kpm_spec
 
-from .common import read_json_file
-
 SPEC_METANODES = 5  # Unique metanodes: Input, Output, Inout, Constant, Subgraph
 PWM_UNIQUE_IPCORE_NODES = 3  # Unique IP Cores from examples/pwm/project.yaml
 PWM_ALL_UNIQUE_NODES = PWM_UNIQUE_IPCORE_NODES + SPEC_METANODES

tests/tests_kpm/test_kpm_validation.py

@@ -18,9 +18,7 @@
     _check_parameters_values,
     _check_unconnected_ports_interfaces,
 )
-from topwrap.util import JsonType
-
-from .common import read_json_file
+from topwrap.util import JsonType, read_json_file
 
 
 def get_dataflow_test(test_name: str) -> JsonType:

topwrap/kpm_topwrap_client.py

@@ -2,11 +2,12 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
+import os
 import threading
 from base64 import b64encode
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, List, Literal, Optional, Tuple, TypedDict, Union
+from typing import Any, Dict, List, Literal, Optional, Tuple, TypedDict, Union
 
 import yaml
 from pipeline_manager_backend_communication.communication_backend import (
@@ -23,6 +24,7 @@
 from .kpm_common import RPCparams
 from .kpm_dataflow_parser import kpm_dataflow_to_design
 from .kpm_dataflow_validator import validate_kpm_design
+from .util import read_json_file, save_file_to_json
 from .yamls_to_kpm_spec_parser import ipcore_yamls_to_kpm_spec
 
 
@@ -43,6 +45,10 @@ def __init__(self, params: RPCparams, client: Optional[CommunicationBackend] = N
         self.build_dir = params.build_dir
         self.design = params.design
         self.client = client
+        # Use the $XDG_DATA_HOME as a destination for saving the dataflow, which defaults to ~/.local/share
+        xdg_data_home_var = Path(os.environ.get("XDG_DATA_HOME", "~/.local/share")).expanduser()
+        self.default_save_file = xdg_data_home_var / "topwrap/dataflow_latest_save.json"
+        self.initial_load = True
 
     def app_capabilities_get(self) -> Dict[Literal["stoppable_methods"], List[str]]:
         return {"stoppable_methods": ["dataflow_run"]}
@@ -101,15 +107,42 @@ def dataflow_import(
     async def frontend_on_connect(self):
         """Gets run when frontend connects, loads initial design"""
         logging.debug("frontend on connect")
-        if self.design is not None:
+        if self.client is None:
+            logging.debug("The client to send a request to is not defined")
+            return
+        if self.default_save_file.exists() and not self.initial_load:
+            latest_dataflow = read_json_file(self.default_save_file)
+            await self.client.request("graph_change", {"dataflow": latest_dataflow})
+        elif self.design is not None:
+            self.initial_load = False
             with open(self.design) as design_file:
                 read_file = design_file.read()
                 dataflow = _kpm_import_handler(read_file, self.yamlfiles)
-                if self.client is None:
-                    logging.debug("There client to send request to is not defined")
-                    return
                 await self.client.request("graph_change", {"dataflow": dataflow})
 
+    async def nodes_on_change(self, **kwargs: Any):
+        await _kpm_handle_graph_change(self)
+
+    async def properties_on_change(self, **kwargs: Any):
+        await _kpm_handle_graph_change(self)
+
+    async def connections_on_change(self, **kwargs: Any):
+        await _kpm_handle_graph_change(self)
+
+    async def position_on_change(self, **kwargs: Any):
+        await _kpm_handle_graph_change(self)
+
+
+async def _kpm_handle_graph_change(rpc_object: RPCMethods):
+    if rpc_object.client is None:
+        return
+    current_graph = await rpc_object.client.request("graph_get")
+    save_file_to_json(
+        rpc_object.default_save_file.parent,
+        rpc_object.default_save_file.name,
+        current_graph["result"]["dataflow"],
+    )
+
 
 def _kpm_import_handler(data: str, yamlfiles: List[Path]) -> JsonType:
     specification = ipcore_yamls_to_kpm_spec(yamlfiles)
@@ -122,7 +155,7 @@ def _design_from_kpm_data(data: JsonType, yamlfiles: List[Path]) -> DesignDescri
     return kpm_dataflow_to_design(data, specification)
 
 
-def _kpm_run_handler(data: JsonType, yamlfiles: List[Path], build_dir: Path) -> list:
+def _kpm_run_handler(data: JsonType, yamlfiles: List[Path], build_dir: Path) -> List[str]:
     """Parse information about design from KPM dataflow format into Topwrap's
     internal representation and build the design.
     """

topwrap/util.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
 
+import json
 from collections import defaultdict
 from enum import Enum
 from pathlib import Path
@@ -82,6 +83,18 @@ def __init__(self, *args: object) -> None:
         super().__init__("Stepped into a code path marked as unreachable", *args)
 
 
+def read_json_file(json_file_path: Path) -> JsonType:
+    with open(json_file_path, "r") as json_file:
+        json_contents = json.load(json_file)
+    return json_contents
+
+
+def save_file_to_json(file_path: Path, file_name: str, file_content: JsonType):
+    file_path.mkdir(parents=True, exist_ok=True)
+    with open(Path(file_path / file_name), "w") as json_file:
+        json.dump(file_content, json_file)
+
+
 def path_relative_to(org_path: Path, rel_to: Path) -> Path:
     """Return the `org_path` that is converted to be relative to `rel_to`.
 

topwrap/yamls_to_kpm_spec_parser.py

@@ -265,6 +265,7 @@ def add_metadata_to_specification(
         "backgroundSize": 15,
         "layout": "CytoscapeEngine - grid",
         "twoColumn": True,
+        "notifyWhenChanged": True,
         "layers": [{"name": lr.value, "nodeLayers": [lr.value]} for lr in LayerType],
         "navbarItems": [
             {