App: implement managing multiple projects (#107)

ultratrace2/app.py

@@ -1,3 +1,4 @@
+from argparse import Namespace
 from tkinter.filedialog import askdirectory as choose_dir
 from typing import Optional
 
@@ -6,27 +7,30 @@
 
 
 class App:
-    def __init__(self, args):  # FIXME: be more granular here
+    def __init__(self, args: Namespace):
 
-        if args.path is None:
+        headless = getattr(args, "headless", False)
+
+        path: Optional[str] = getattr(args, "path", None)
+        if path is None and not headless:
             path = choose_dir()
-            if not path:
-                raise ValueError("You must choose a directory to open")
-        else:
-            path = args.path
+        if not path:
+            raise ValueError("You must choose a directory to open")
+
+        self.project: Project = Project.get_by_path(path)
 
-        self.project = Project(path)
-        self.gui = GUI(theme=args.theme)
+        if not headless:
+            self.gui = GUI(theme=args.theme)
 
-    def main(self):
+    def main(self) -> None:
         pass
 
 
 # singleton
 app: Optional[App] = None
 
 
-def initialize_app(args) -> App:  # FIXME: be more granular here
+def initialize_app(args: Namespace) -> App:
     global app
     app = App(args)
     return app

ultratrace2/model/files/bundle.py

@@ -1,7 +1,7 @@
 import logging
 import os
 
-from typing import Dict, List, Set
+from typing import Dict, Sequence, FrozenSet
 
 from .impls import Sound, Alignment, ImageSet
 
@@ -36,35 +36,48 @@ def __repr__(self):
 
 class FileBundleList:
 
-    exclude_dirs: Set[str] = set(
+    exclude_dirs: FrozenSet[str] = frozenset(
         [
             ".git",
             "node_modules",
             "__pycache__",
+            ".ultratrace",
             # FIXME: add more ignoreable dirs
         ]
     )
 
-    def __init__(self, path: str, extra_exclude_dirs: List[str] = []):
+    def __init__(self, bundles: Dict[str, FileBundle]):
 
-        # FIXME: implement `extra_exclude_dirs` as a command-line arg
-        for extra_exclude_dir in extra_exclude_dirs:
-            self.exclude_dirs.add(extra_exclude_dir)
+        self.current_bundle = None
+        self.bundles: Dict[str, FileBundle] = bundles
 
-        self.path = path
-        self.has_alignment_impl = False
-        self.has_image_impl = False
-        self.has_sound_impl = False
+        self.has_alignment_impl: bool = False
+        self.has_image_impl: bool = False
+        self.has_sound_impl: bool = False
 
-        self.current_bundle = None
+        for bundle in bundles.values():
+            if not self.has_alignment_impl and bundle.alignment_file.has_impl():
+                self.has_alignment_impl = True
+            if not self.has_image_impl and bundle.image_file.has_impl():
+                self.has_image_impl = True
+            if not self.has_sound_impl and bundle.sound_file.has_impl():
+                self.has_sound_impl = True
+
+    @classmethod
+    def build_from_dir(
+        cls, root_path: str, extra_exclude_dirs: Sequence[str] = []
+    ) -> "FileBundleList":
+
+        # FIXME: implement `extra_exclude_dirs` as a command-line arg
+        exclude_dirs = cls.exclude_dirs.union(extra_exclude_dirs)
 
         bundles: Dict[str, FileBundle] = {}
 
         # NB: `topdown=True` increases runtime cost from O(n) -> O(n^2), but it allows us to
         #     modify `dirs` in-place so that we can skip certain directories.  For more info,
         #     see https://stackoverflow.com/questions/19859840/excluding-directories-in-os-walk
-        for path, dirs, filenames in os.walk(path, topdown=True):
-            dirs[:] = [d for d in dirs if d not in self.exclude_dirs]
+        for path, dirs, filenames in os.walk(root_path, topdown=True):
+            dirs[:] = [d for d in dirs if d not in exclude_dirs]
 
             for filename in filenames:
 
@@ -83,12 +96,4 @@ def __init__(self, path: str, extra_exclude_dirs: List[str] = []):
                 if not bundles[name].interpret(filepath):
                     logger.warning(f"unrecognized filetype: {filepath}")
 
-        # FIXME: do this when we add to our data structure
-        for filename, bundle in bundles.items():
-            # build up self.bundles here
-            if not self.has_alignment_impl and bundle.alignment_file.has_impl():
-                self.has_alignment_impl = True
-            if not self.has_image_impl and bundle.image_file.has_impl():
-                self.has_image_impl = True
-            if not self.has_sound_impl and bundle.sound_file.has_impl():
-                self.has_sound_impl = True
+        return cls(bundles)

ultratrace2/model/files/impls/image.py

@@ -54,7 +54,7 @@ def load(self) -> None:
                     "Invalid DICOM ({self.path}), unknown shape {pixels.shape}"
                 )
 
-            os.mkdir(self.png_path)
+            os.mkdir(self.png_path, mode=0o755)
             for i in tqdm(range(frames), desc="converting to PNG"):
                 filename = os.path.join(self.png_path, f"{i:06}.png")
                 arr = pixels[i, :, :] if is_greyscale else pixels[i, :, :, :]

ultratrace2/model/project.py

@@ -1,23 +1,68 @@
+import logging
 import os
+import pickle
 
 from .trace import TraceList
 from .files.bundle import FileBundleList
 
+logger = logging.getLogger(__name__)
+
 
 class Project:
-    def __init__(self, path: str):
-        if not os.path.exists(path):
-            raise ValueError(f"cannot initialize project at {path}")
-        self.root_path = os.path.realpath(os.path.abspath(path))  # absolute path
-        self.traces = TraceList()
-        self.files = FileBundleList(self.root_path)
+    def __init__(self, traces: TraceList, files: FileBundleList):
+        """
+        Internal function: to construct a Project, either call ::get_by_path()
+        """
+        self.traces = traces
+        self.files = files
 
     def save(self):
         raise NotImplementedError()
 
     @classmethod
-    def load(cls):
-        raise NotImplementedError()
+    def load(cls, save_file) -> "Project":
+        try:
+            with open(save_file, "rb") as fp:
+                project = pickle.load(fp)
+                assert isinstance(project, Project)
+                return project
+        except Exception as e:
+            logger.error(e)
+            raise RuntimeError(str(e))
+
+    @classmethod
+    def get_by_path(cls, root_path) -> "Project":
+
+        root_path = os.path.realpath(os.path.abspath(root_path))  # absolute path
+        if not os.path.exists(root_path) or not os.path.isdir(root_path):
+            raise ValueError(
+                f"cannot initialize project at {root_path}: not a directory"
+            )
+
+        save_dir = Project.get_save_dir(root_path)
+        if not os.path.exists(save_dir):
+            os.mkdir(save_dir, mode=0o755)
+
+        save_file = Project.get_save_file(root_path)
+        try:
+            return Project.load(save_file)
+        except RuntimeError:
+            logger.info(
+                f"Unable to find existing project at {root_path}, creating new one..."
+            )
+
+        traces = TraceList()
+        file_bundles = FileBundleList.build_from_dir(root_path)
+        return Project(traces, file_bundles)
+
+    @staticmethod
+    def get_save_dir(path: str) -> str:
+        return os.path.join(path, ".ultratrace")
+
+    @staticmethod
+    def get_save_file(path: str) -> str:
+        save_dir = Project.get_save_dir(path)
+        return os.path.join(save_dir, "project.pkl")
 
     def filepath(self):
         raise NotImplementedError()

ultratrace2/model/trace.py

@@ -17,7 +17,7 @@ class Trace:
     def __init__(self, name: str, color: Color):
         self.id: UUID = uuid4()
         self.is_visible: bool = True
-        self.xhairs: Dict[FileBundle, Dict[int, Set[XHair]]] = {}
+        self.xhairs: Dict["FileBundle", Dict[int, Set[XHair]]] = {}
         self.name = name
         self.color = color
 
@@ -45,7 +45,7 @@ def show(self) -> None:
     def hide(self) -> None:
         self.is_visible = False
 
-    def add_xhair(self, bundle: FileBundle, frame: int, x: float, y: float) -> None:
+    def add_xhair(self, bundle: "FileBundle", frame: int, x: float, y: float) -> None:
         if bundle not in self.xhairs:
             self.xhairs[bundle] = {}
         if frame not in self.xhairs[bundle]:

ultratrace2/model/xhair.py

@@ -7,7 +7,7 @@
 
 
 class XHair:
-    def __init__(self, trace: Trace, x: float, y: float):
+    def __init__(self, trace: "Trace", x: float, y: float):
 
         self.id = uuid4()
         self.trace = trace
@@ -49,5 +49,5 @@ def move(self, x: float, y: float) -> None:
         self.x = x
         self.y = y
 
-    def get_color(self) -> Color:
+    def get_color(self) -> "Color":
         return self.trace.get_color()

ultratrace2/tests/test_app.py

@@ -0,0 +1,36 @@
+from argparse import Namespace
+from pathlib import Path
+
+import pytest
+
+from .. import app as app_py
+from ..app import App, initialize_app
+
+
+@pytest.mark.parametrize(
+    "args,expected",
+    [
+        (Namespace(headless=True), ValueError),
+        (Namespace(headless=True, path="/path/to/nowhere"), ValueError),
+        (Namespace(headless=True, path="/dev/null"), ValueError),  # not a directory
+        (Namespace(headless=True, path="/"), PermissionError),  # not writeable
+    ],
+)
+def test_initialize_app_invalid(args: Namespace, expected: Exception, tmpdir) -> None:
+    app_py.app = None  # overwrite global object
+    with pytest.raises(expected):
+        initialize_app(args)
+
+
+@pytest.mark.parametrize("args", [(Namespace(headless=True)),])  # noqa: E231
+def test_initialize_app_valid(args: Namespace, tmp_path: Path) -> None:
+    # overwrite global object
+    app_py.app = None
+    # initialize an empty dir
+    path = tmp_path / "ultratrace-test-app"
+    path.mkdir()
+    # save that to the Namespace
+    args.path = path
+    # initialize
+    app = initialize_app(args)
+    assert isinstance(app, App)