Introduce DirectoryObject and FileObject in Workflow

pyiron_contrib/workflow/files.py

@@ -0,0 +1,99 @@
+from pathlib import Path
+
+
+def delete_files_and_directories_recursively(path):
+    if not path.exists():
+        return
+    for item in path.rglob("*"):
+        if item.is_file():
+            item.unlink()
+        else:
+            delete_files_and_directories_recursively(item)
+    path.rmdir()
+
+
+def categorize_folder_items(folder_path):
+    types = [
+        "dir",
+        "file",
+        "mount",
+        "symlink",
+        "block_device",
+        "char_device",
+        "fifo",
+        "socket",
+    ]
+    results = {t: [] for t in types}
+
+    for item in folder_path.iterdir():
+        for tt in types:
+            try:
+                if getattr(item, f"is_{tt}")():
+                    results[tt].append(str(item))
+            except NotImplementedError:
+                pass
+    return results
+
+
+class DirectoryObject:
+    def __init__(self, directory):
+        self.path = Path(directory)
+        self.create()
+
+    def create(self):
+        self.path.mkdir(parents=True, exist_ok=True)
+
+    def delete(self):
+        delete_files_and_directories_recursively(self.path)
+
+    def list_content(self):
+        return categorize_folder_items(self.path)
+
+    def __len__(self):
+        return sum([len(cc) for cc in self.list_content().values()])
+
+    def __repr__(self):
+        return f"DirectoryObject(directory='{self.path}')\n{self.list_content()}"
+
+    def get_path(self, file_name):
+        return self.path / file_name
+
+    def file_exists(self, file_name):
+        return self.get_path(file_name).is_file()
+
+    def write(self, file_name, content, mode="w"):
+        with self.get_path(file_name).open(mode=mode) as f:
+            f.write(content)
+
+    def create_subdirectory(self, path):
+        return DirectoryObject(self.path / path)
+
+    def create_file(self, file_name):
+        return FileObject(file_name, self)
+
+
+class FileObject:
+    def __init__(self, file_name: str, directory: DirectoryObject):
+        self.directory = directory
+        self._file_name = file_name
+
+    @property
+    def file_name(self):
+        return self._file_name
+
+    @property
+    def path(self):
+        return self.directory.path / Path(self._file_name)
+
+    def write(self, content, mode="x"):
+        self.directory.write(file_name=self.file_name, content=content, mode=mode)
+
+    def read(self, mode="r"):
+        with open(self.path, mode=mode) as f:
+            return f.read()
+
+    def is_file(self):
+        return self.directory.file_exists(self.file_name)
+
+    def delete(self):
+        self.path.unlink()

pyiron_contrib/workflow/node.py

@@ -369,6 +369,8 @@ def __init__(
         if update_on_instantiation:
             self.update()
 
+        self._working_directory = None
+
     @property
     def _input_args(self):
         return inspect.signature(self.node_function).parameters
@@ -560,6 +562,19 @@ def to_dict(self):
             "signals": self.signals.to_dict(),
         }
 
+    @property
+    def working_directory(self):
+        if self._working_directory is None:
+            if self.parent is None:
+                raise ValueError(
+                    "working directory is available only if the node is"
+                    " attached to a workflow"
+                )
+            self._working_directory = self.parent.working_directory.create_subdirectory(
+                self.label
+            )
+        return self._working_directory
+
 
 class FastNode(Node):
     """

pyiron_contrib/workflow/workflow.py

@@ -4,6 +4,7 @@
 from pyiron_contrib.workflow.has_to_dict import HasToDict
 from pyiron_contrib.workflow.node import Node, node, fast_node, single_value_node
 from pyiron_contrib.workflow.util import DotDict
+from pyiron_contrib.workflow.files import DirectoryObject
 
 
 class _NodeDecoratorAccess:
@@ -121,10 +122,17 @@ class Workflow(HasToDict, HasNodes):
     def __init__(self, label: str, *nodes: Node, strict_naming=True):
         super().__init__(strict_naming=strict_naming)
         self.label = label
+        self._working_directory = None
 
         for node in nodes:
             self.add_node(node)
 
+    @property
+    def working_directory(self):
+        if self._working_directory is None:
+            self._working_directory = DirectoryObject(self.label)
+        return self._working_directory
+
     @property
     def inputs(self):
         return DotDict(

tests/unit/workflow/test_files.py

@@ -0,0 +1,50 @@
+import unittest
+from pyiron_contrib.workflow.files import DirectoryObject, FileObject
+from pathlib import Path
+
+
+class TestFiles(unittest.TestCase):
+    def setUp(cls):
+        cls.directory = DirectoryObject("test")
+
+    def tearDown(cls):
+        cls.directory.delete()
+
+    def test_directory_exists(self):
+        self.assertTrue(Path("test").exists() and Path("test").is_dir())
+
+    def test_write(self):
+        self.directory.write(file_name="test.txt", content="something")
+        self.assertTrue(self.directory.file_exists("test.txt"))
+        self.assertTrue(
+            "test/test.txt" in [
+                ff.replace("\\", "/")
+                for ff in self.directory.list_content()['file']
+            ]
+        )
+        self.assertEqual(len(self.directory), 1)
+
+    def test_create_subdirectory(self):
+        self.directory.create_subdirectory("another_test")
+        self.assertTrue(Path("test/another_test").exists())
+
+    def test_path(self):
+        f = FileObject("test.txt", self.directory)
+        self.assertEqual(str(f.path).replace("\\", "/"), "test/test.txt")
+
+    def test_read_and_write(self):
+        f = FileObject("test.txt", self.directory)
+        f.write("something")
+        self.assertEqual(f.read(), "something")
+
+    def test_is_file(self):
+        f = FileObject("test.txt", self.directory)
+        self.assertFalse(f.is_file())
+        f.write("something")
+        self.assertTrue(f.is_file())
+        f.delete()
+        self.assertFalse(f.is_file())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/unit/workflow/test_node.py

@@ -359,6 +359,12 @@ def my_node(x: int = 0, y: int = 0, z: int = 0):
             msg="After the run, all three should now be waiting for updates again"
         )
 
+    def test_working_directory(self):
+        n_f = Node(plus_one, "output")
+        with self.assertRaises(ValueError):
+            _ = n_f.working_directory
+            #  cf. test_workflow.py for the case that it does notraise an error
+
 
 if __name__ == '__main__':
     unittest.main()

tests/unit/workflow/test_workflow.py

@@ -1,16 +1,17 @@
-from unittest import TestCase, skipUnless
+import unittest
 from sys import version_info
 
 from pyiron_contrib.workflow.node import Node
 from pyiron_contrib.workflow.workflow import Workflow
+from pyiron_contrib.workflow.files import DirectoryObject
 
 
 def fnc(x=0):
     return x + 1
 
 
-@skipUnless(version_info[0] == 3 and version_info[1] >= 10, "Only supported for 3.10+")
-class TestWorkflow(TestCase):
+@unittest.skipUnless(version_info[0] == 3 and version_info[1] >= 10, "Only supported for 3.10+")
+class TestWorkflow(unittest.TestCase):
 
     def test_node_addition(self):
         wf = Workflow("my_workflow")
@@ -107,3 +108,16 @@ def plus_one(x: int = 0) -> int:
             return x + 1
 
         self.assertEqual(plus_one().outputs.y.value, 1)
+
+    def test_working_directory(self):
+        wf = Workflow("wf")
+        self.assertTrue(wf._working_directory is None)
+        self.assertIsInstance(wf.working_directory, DirectoryObject)
+        self.assertTrue(str(wf.working_directory.path).endswith(wf.label))
+        wf.add.Node(fnc, "output")
+        self.assertTrue(str(wf.fnc.working_directory.path).endswith(wf.fnc.label))
+        wf.working_directory.delete()
+
+
+if __name__ == '__main__':
+    unittest.main()