Merge pull request #161 from 0Hughman0/migrate

Creating migration tools for 0.2.x to 0.3.x

cassini/migrate/V0_1toV0_2.py

@@ -0,0 +1,95 @@
+import nbformat
+import re
+
+from .base import BaseMigrator, cell_processor
+
+
+class V0_1toV0_2(BaseMigrator):
+    def __init__(self, project):
+        self.project = project
+        self.home = project.home
+
+    @cell_processor
+    def fix_imports(cell):
+        text = cell["source"]
+        text = text.replace("from htools.exp import g", "from htools.scify import *")
+        cell["source"] = text
+
+    @cell_processor
+    def fix_g_references(cell):
+        text = cell["source"]
+
+        def replace(match):
+            return f"env.{match.group(1)}"
+
+        cell["source"] = re.sub("g\.(.+)", replace, text)
+
+    @cell_processor
+    def replace_mksmpl(cell):
+        pattern = "%%mksmpl (.+)"
+        text = cell["source"]
+
+        match = re.search(pattern, text)
+        if match:
+            cell["source"] = f"smpl = env('{match.group(1)}')\nsmpl.header()"
+
+    @cell_processor
+    def replace_smplconc(cell):
+        text = cell["source"]
+        cell["source"] = text.replace("%%smplconc", "%%conc")
+
+    @cell_processor
+    def replace_expconc(cell):
+        text = cell["source"]
+        cell["source"] = text.replace("%%expconc", "%%conc")
+
+    @cell_processor
+    def fix_meta_store(cell):
+        text = cell["source"]
+        pattern = "smpl\.write_meta\('(.+)', (.+)\)"
+
+        def replace(match):
+            return f"smpl.{match.group(1)} = {match.group(2)}"
+
+        cell["source"] = re.sub(pattern, replace, text)
+
+    def walk_smpls(self):
+        for wp in self.home:
+            for exp in wp:
+                for smpl in exp:
+                    yield smpl
+
+    def migrate(self):
+        for smpl in self.walk_smpls():
+            if not smpl.file.exists():
+                print(f"No file found for {smpl} - skippiping")
+                continue
+
+            print("Fixing", smpl)
+            with open(smpl.file, "rb") as f:
+                nb = nbformat.read(f, 4)
+
+            print("Backing up old file")
+            try:
+                new_name = smpl.file.rename(
+                    smpl.file.with_name(f"{smpl.name}-old.ipynb")
+                )
+            except FileExistsError:
+                print(f"Already updated {smpl} - skipping")
+                continue
+
+            try:
+                print("Applying processors")
+                for cell in nb["cells"]:
+                    for processor in self.cell_processors:
+                        processor(cell)
+                print("Success")
+                print("Errors: ", nbformat.validate(nb, version=4))
+
+                print("Writing new file")
+                with open(smpl.file, "w", encoding="utf-8") as f:
+                    nbformat.write(nb, f, 4)
+                print("Success")
+            except Exception:
+                print("Exception occured, rolling back")
+                new_name.rename(smpl.file.name)

cassini/migrate/V0_2toV0_3.py

@@ -0,0 +1,44 @@
+import json
+import datetime
+import shutil
+
+from .base import BaseMigrator
+
+
+class V0_2toV0_3(BaseMigrator):
+    def __init__(self, project) -> None:
+        self.project = project
+        self.home = project.home
+
+    def migrate(self):
+        from cassini import NotebookTierBase
+
+        for tier in self.walk_tiers():
+            if not isinstance(tier, NotebookTierBase):
+                continue
+            else:
+                backup_path = shutil.copy(
+                    tier.meta_file, tier.meta_file.with_suffix(".backup")
+                )
+                try:
+                    self.migrate_meta(tier)
+                except Exception:
+                    raise RuntimeError(
+                        "Error occured, please restore from backup", backup_path
+                    )
+                else:
+                    print("Successfully migrated, removing backup")
+                    backup_path.unlink()
+
+    def migrate_meta(self, tier):
+        with open(tier.meta_file, "r") as fs:
+            meta = json.load(fs)
+
+        started = meta.get("started")
+
+        if started:
+            started_dt = datetime.datetime.strptime(started, "%d/%m/%Y")
+            meta["started"] = str(started_dt)
+
+        with open(tier.meta_file, "w") as fs:
+            json.dump(meta, fs)

cassini/migrate/__main__.py

@@ -0,0 +1,63 @@
+import sys
+import argparse
+from typing import List
+
+from cassini.utils import find_project
+
+from .base import BaseMigrator
+from .V0_1toV0_2 import V0_1toV0_2
+from .V0_2toV0_3 import V0_2toV0_3
+
+
+migrators = {("0.1", "0.2"): V0_1toV0_2, ("0.2", "0.3"): V0_2toV0_3}
+
+
+def main(args: List[str]) -> BaseMigrator:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Migration tool for cassini projects. "
+            "It's not that fancy, so sequential updates may be needed."
+        )
+    )
+    parser.add_argument(
+        "old", choices=["0.1", "0.2"], help="which version to migrate from"
+    )
+    parser.add_argument(
+        "new", choices=["0.2", "0.3"], help="which version to migrate to"
+    )
+    parser.add_argument(
+        "--cassini-project",
+        default="project.py:project",
+        help="cassini project to migrate, see cassini.utils.find_project for possible forms",
+    )
+
+    out = parser.parse_args(args)
+
+    project = find_project(out.cassini_project)
+
+    print("Found project", project)
+
+    old, new = out.old, out.new
+
+    print("Looking for migrator for", old, "to", new)
+
+    try:
+        migrator = migrators[(old, new)](project)
+    except KeyError:
+        raise ValueError(
+            "No migator available for this combination, options are", list(migrators)
+        )
+
+    print("Found", migrator)
+
+    print("Running migrator")
+
+    migrator.migrate()
+
+    print("Success")
+
+    return migrator
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])

cassini/migrate/base.py

@@ -0,0 +1,40 @@
+from abc import abstractmethod
+from typing import Callable, List
+
+from nbformat import NotebookNode
+
+CellProcessor = Callable[[NotebookNode], None]
+
+
+class BaseMigrator:
+    cell_processors: List[CellProcessor] = []
+
+    def __init_subclass__(cls) -> None:
+        cls.cell_processors = []
+
+    @classmethod
+    def _cell_processor(cls, f: CellProcessor):
+        cls.cell_processors.append(f)
+        return f
+
+    def walk_tiers(self):
+        yield self.home
+
+        for wp in self.home:
+            yield wp
+
+            for exp in wp:
+                yield exp
+
+                for smpl in exp:
+                    yield smpl
+
+    @abstractmethod
+    def migrate(self):
+        """
+        Perform the migration.
+        """
+
+
+def cell_processor(f):
+    return BaseMigrator._cell_processor(f)