Upload 7 files

CodeTestingAtomicFlow.py

@@ -0,0 +1,39 @@
+import os
+import time
+from typing import Dict, Any
+from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow
+
+
+class CodeTestingAtomicFlow(InterpreterAtomicFlow):
+    def _wait_for_file_update(self, file_location, last_check_time, check_interval=1):
+        while True:
+            time.sleep(check_interval)
+            current_time = os.path.getmtime(file_location)
+            if current_time != last_check_time:
+                break
+    def _prepare_code_and_lang(self, input_data: Dict[str, Any]):
+        file_location = input_data["temp_code_file_location"]
+        input_data["language"] = "python"
+        with open(file_location, "r") as file:
+            code_str = file.read()
+        input_data["code"] = code_str
+
+    def _check_input(self, input_data: Dict[str, Any]):
+        assert "temp_code_file_location" in input_data, "temp_code_file_location not passed to CodeTestingAtomicFlow"
+        assert "temp_code_file_written_timestamp" in input_data, "temp_code_file_written_timestamp not passed to CodeTestingAtomicFlow"
+
+    def _delete_file(self, file_location):
+        if os.path.exists(file_location):
+            os.remove(file_location)
+
+    def run(
+            self,
+            input_data: Dict[str, Any]):
+        self._check_input(input_data)
+        file_loc = input_data["temp_code_file_location"]
+        last_timestamp = input_data["temp_code_file_written_timestamp"]
+        self._wait_for_file_update(file_loc, last_timestamp)
+        self._prepare_code_and_lang(input_data)
+        response = self._call()
+        self._delete_file(file_loc)
+        return {"test_results": response}

CodeTestingAtomicFlow.yaml

@@ -0,0 +1,10 @@
+name: "CodeTestingAtomicFlow"
+description: "A flow that runs test code"
+
+
+input_interface:
+  - "temp_code_file_location"
+  - "temp_code_file_written_timestamp"
+
+output_interface:
+  - "test_results"

TestCodeFileEditAtomicFlow.py

@@ -0,0 +1,47 @@
+import os
+from typing import Dict, Any
+from flow_modules.Tachi67.CodeFileEditFlowModule import CodeFileEditAtomicFlow
+
+class TestCodeFileEditAtomicFlow(CodeFileEditAtomicFlow):
+    def _generate_import_statement(self, code_lib_location):
+        module_dir = os.path.dirname(code_lib_location)
+        module_name = os.path.basename(code_lib_location).rstrip('.py')
+        import_code = (
+            f"import sys\n"
+            f"sys.path.insert(0, '{module_dir}')\n"
+            f"import {module_name}\n"
+        )
+        return import_code
+
+    def _generate_content(self, code_lib_location, code_str) -> str:
+        import_code_lib_str = self._generate_import_statement(code_lib_location)
+        content = (
+                "# Don't touch this import statement \n"
+                + import_code_lib_str + "\n"
+                "# Here is the code just generated \n" +
+                code_str + "\n"
+                "# Below, please provide code to test it.\n"
+                "# The simplest form could be just calling it with appropriate parameters. \n"
+                "# You could also assert the output, anyway, the test results will be informed to JARVIS. \n"
+                "# If you do not write anything, JARVIS just checks if the syntax is alright. \n"
+                "###########\n"
+                "\n # Test Code:\n" +
+                "\n############\n"
+        )
+        return content
+
+    def _generate_temp_file_location(self, code_lib_location):
+        directory = os.path.dirname(code_lib_location)
+        ret = os.path.join(directory, 'temp_tests.py')
+        return ret
+
+    def _check_input(self, input_data: Dict[str, Any]):
+        assert "code" in input_data, "code is not passed to TestCodeFileEditAtomicFlow"
+        assert "code_library" in input_data, "code_library is not passed to TestCodeFileEditAtomicFlow"
+
+    def _generate_input_to_writer(self, input_data: Dict[str, Any]):
+        code_str = input_data['code']
+        code_lib_location = input_data["code_library"]
+        content_to_write = self._generate_content(code_lib_location, code_str)
+        file_location_to_write = self._generate_temp_file_location(code_lib_location)
+        return content_to_write, file_location_to_write

TestCodeFileEditAtomicFlow.yaml

@@ -0,0 +1,11 @@
+name: "TestCodeFileEditorAtomicFlow"
+description: "A flow that writes tests code to a temp file"
+
+input_interface:
+  - "code"
+  - "code_library"
+
+output_interface:
+  - "code_editor_output"
+  - "temp_code_file_location"
+  - "temp_code_file_written_timestamp"

TestCodeFlow.py

@@ -0,0 +1,52 @@
+from copy import deepcopy
+from typing import Any, Dict
+
+from flows.base_flows import SequentialFlow
+from flows.utils import logging
+
+logging.set_verbosity_debug()
+log = logging.get_logger(__name__)
+
+class TestCodeFlow(SequentialFlow):
+    REQUIRED_KEYS_CONFIG = ["max_rounds", "early_exit_key", "topology", "memory_files"]
+
+    def __init__(
+            self,
+            memory_files: Dict[str, Any],
+            **kwargs
+    ):
+        super().__init__(**kwargs)
+        self.memory_files = memory_files
+
+    @classmethod
+    def instantiate_from_config(cls, config):
+        flow_config = deepcopy(config)
+
+        kwargs = {"flow_config": flow_config}
+
+        # ~~~ Set up memory file ~~~
+        memory_files = flow_config["memory_files"]
+        kwargs.update({"memory_files": memory_files})
+
+        # ~~~ Set up subflows ~~~
+        kwargs.update({"subflows": cls._set_up_subflows(flow_config)})
+
+        # ~~~ Instantiate flow ~~~
+        return cls(**kwargs)
+
+    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        # ~~~ sets the input_data in the flow_state dict ~~~
+        self._state_update_dict(update_data=input_data)
+
+        # ~~~ set the memory file to the flow state ~~~
+        self._state_update_dict(update_data={"memory_files": self.memory_files})
+
+        max_rounds = self.flow_config.get("max_rounds", 1)
+        if max_rounds is None:
+            log.info(f"Running {self.flow_config['name']} without `max_rounds` until the early exit condition is met.")
+
+        self._sequential_run(max_rounds=max_rounds)
+
+        output = self._get_output_from_state()
+
+        return output

TestCodeFlow.yaml

@@ -0,0 +1,49 @@
+name: "TestCodeFlow"
+description: "Test the code generated by InteractiveCodeGenFlow"
+
+_target_: Tachi67.TestCodeFlowModule.TestCodeFlow.instantiate_from_default_config
+
+memory_files: ???
+
+input_interface:
+  - "code"
+
+output_interface:
+  - "test_results"
+
+subflows_config:
+  MemoryReading:
+    _target_: Tachi67.MemoryReadingFlowModule.MemoryReadingAtomicFlow.instantiate_from_default_config
+
+  TestCodeFileEdit:
+    _target_: TestCodeFileEditAtomicFlow.TestCodeFileEditAtomicFlow.instantiate_from_default_config
+
+  CodeTesting:
+    _target_: CodeTestingAtomicFlow.CodeTestingAtomicFlow.instantiate_from_default_config
+
+early_exit_key: "EARLY_EXIT"
+
+topology:
+  - goal: "Read in necessary memory"
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+    flow: MemoryReading
+    reset: false
+
+  - goal: "Write the code & instructions for testing to a temp file"
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+    flow: TestCodeFileEdit
+    reset: false
+
+  - goal: "Test the code"
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+    flow: CodeTesting
+    reset: false

__init__.py

@@ -0,0 +1,13 @@
+dependencies = [
+    {"url": "Tachi67/MemoryReadingFlowModule", "revision": "main"},
+    {"url": "Tachi67/CodeFileEditFlowModule", "revision": "main"},
+    {"url": "Tachi67/InterpreterFlowModule", "revision": "main"},
+]
+
+from flows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+from .CodeTestingAtomicFlow import CodeTestingAtomicFlow
+from .TestCodeFileEditAtomicFlow import TestCodeFileEditAtomicFlow
+from .TestCodeFlow import TestCodeFlow