Add some file types

pasch/__init__.py

@@ -1,11 +1,10 @@
-from . import modules
+from . import cmd, file, modules
-from .cmd import run_capture, run_execute
 from .orchestrator import Module, Orchestrator
 
 __all__: list[str] = [
     "Module",
     "Orchestrator",
+    "cmd",
+    "file",
     "modules",
-    "run_capture",
-    "run_execute",
 ]

pasch/file/__init__.py

@@ -0,0 +1,12 @@
+from .binary import BinaryFile
+from .file import TAG, File
+from .json import JsonFile
+from .text import TextFile
+
+__all__: list[str] = [
+    "TAG",
+    "BinaryFile",
+    "File",
+    "JsonFile",
+    "TextFile",
+]

pasch/file/binary.py

@@ -0,0 +1,9 @@
+from .file import File
+
+
+class BinaryFile(File):
+    def __init__(self, data: bytes) -> None:
+        self.data = data
+
+    def to_bytes(self) -> bytes:
+        return self.data

pasch/file/file.py

@@ -0,0 +1,8 @@
+from abc import ABC, abstractmethod
+
+TAG = "This file was generated by pasch."
+
+
+class File(ABC):
+    @abstractmethod
+    def to_bytes(self) -> bytes: ...

pasch/file/json.py

@@ -0,0 +1,51 @@
+from typing import Self
+import json
+from dataclasses import dataclass
+from typing import Any
+
+from .file import TAG, File
+from .text import TextFile
+
+
+@dataclass
+class JsonFileProxy:
+    file: "JsonFile"
+    path: tuple[str, ...]
+
+    def at(self, *path: str) -> Self:
+        return JsonFileProxy(self.file, self.path + path)
+
+    def set(self, value: Any) -> None:
+        if not self.path:
+            self.file.set(value)
+
+        data = self.file.data
+        *parts, last = self.path
+        for part in parts:
+            data = data[part]
+        data[last] = value
+
+    def tag_here(self, tag: str = TAG) -> None:
+        self.set(tag)
+
+
+class JsonFile(File):
+    def __init__(self, data: Any = {}) -> None:
+        self.data = data
+
+    def at(self, *path: str) -> JsonFileProxy:
+        return JsonFileProxy(self, path)
+
+    def set(self, value: Any) -> None:
+        self.data = value
+
+    def tag(self, tag: str = TAG, key: str | list[str] = "_tag") -> None:
+        if isinstance(key, str):
+            self.at(key).tag_here(tag=tag)
+        self.at(*key).tag_here(tag=tag)
+
+    def to_text(self) -> TextFile:
+        return TextFile(json.dumps(self.data))
+
+    def to_bytes(self) -> bytes:
+        return self.to_text().to_bytes()

pasch/file/text.py

@@ -0,0 +1,37 @@
+from .binary import BinaryFile
+from .file import TAG, File
+
+
+class TextFile(File):
+    def __init__(self, data: str = "") -> None:
+        self.data = data
+
+    def prepend(self, line: str, newline: bool = True) -> None:
+        if newline:
+            line = f"{line}\n"
+        self.data = line + self.data
+
+    def append(self, line: str, newline: bool = True) -> None:
+        if newline:
+            line = f"{line}\n"
+        self.data = self.data + line
+
+    def tag(
+        self,
+        tag: str = TAG,
+        comment: str | None = None,
+        newline: bool = True,
+        prepend: bool = True,
+    ) -> None:
+        if comment is not None:
+            tag = f"{comment} {tag}"
+        if prepend:
+            self.prepend(tag, newline=newline)
+        else:
+            self.append(tag, newline=newline)
+
+    def to_binary(self) -> BinaryFile:
+        return BinaryFile(self.data.encode("utf-8"))
+
+    def to_bytes(self) -> bytes:
+        return self.to_binary().to_bytes()

pasch/modules/files.py

@@ -6,6 +6,7 @@ from pathlib import Path
 
 from rich.markup import escape
 
+from pasch.file.file import File
 from pasch.orchestrator import Module, Orchestrator
 
 
@@ -101,9 +102,9 @@ class Files(Module):
     def _read_path(self, path: Path | str) -> Path:
         return self._root / path
 
-    def add(self, path: Path | str, content: bytes) -> None:
+    def add(self, path: Path | str, content: File) -> None:
         path = self._read_path(path)
-        self._files[path_to_str(path)] = content
+        self._files[path_to_str(path)] = content.to_bytes()
 
     def realize(self) -> None:
         for path, content in sorted(self._files.items()):