Add info about project versions to allow replays when debugging

workpackages/wp_mergin.py

@@ -21,6 +21,7 @@
 
 import getpass
 import glob
+import json
 import time
 import mergin
 import mergin.client_push
@@ -32,6 +33,7 @@
 from .version import __version__
 from .wp_utils import download_project_with_cache, ProjectPadlock
 from .wp import load_config_from_yaml, make_work_packages, WPConfig
+from .wp_replay import Replay
 
 
 class MerginWPContext:
@@ -227,6 +229,12 @@ def prepare_work_package(wp):
     if not ctx.skip_lock:
         ctx.project_padlock.lock(ctx.master_dir)
 
+    # output which versions we are about to use
+    replay = Replay.from_context(ctx, wp_config, wp_new)
+    replay_json_dump = json.dumps(replay.to_dict(), indent=2)
+    print("INPUT PROJECTS:")
+    print(replay_json_dump)
+
     return wp_config, wp_new, gpkg_path, master_project_files
 
 

workpackages/wp_replay.py

@@ -0,0 +1,70 @@
+"""
+This module contains Replay class to allow debugging of work package runs.
+When data get downloaded from a MM server, we capture the data to allow
+replay as a JSON, so that at some point later we can use the same project
+versions.
+"""
+
+import mergin
+import os
+
+from .wp import WPConfig
+
+
+class ReplayProject:
+    """Contains replay information about a single project (master or WP)"""
+
+    def __init__(self, wp_name: str, project_full_name: str, version: str):
+        self.wp_name = wp_name
+        self.project_full_name = project_full_name
+        self.version = version  # empty string if the project does not exist yet
+
+    def to_dict(self):
+        return {
+            "wp-name": self.wp_name,
+            "project-full-name": self.project_full_name,
+            "version": self.version,
+        }
+
+    @staticmethod
+    def from_dict(project_dict: dict) -> "ReplayProject":
+        return ReplayProject(project_dict["wp"], project_dict["project-full-name"], project_dict["version"])
+
+    @staticmethod
+    def from_project_directory(directory: str, wp_name: str = None) -> "ReplayProject":
+        mp = mergin.MerginProject(directory)
+        return ReplayProject(wp_name, mp.project_full_name(), mp.version())
+
+
+class Replay:
+    """Contains replay information about master project and all WP projects"""
+
+    def __init__(self, master_project: ReplayProject, wp_projects: list[ReplayProject]):
+        self.master_project = master_project
+        self.wp_projects = wp_projects
+
+    def to_dict(self) -> str:
+        return {"master": self.master_project.to_dict(), "wp": [project.to_dict() for project in self.wp_projects]}
+
+    @staticmethod
+    def from_dict(replay_dict) -> "Replay":
+        return Replay(
+            ReplayProject.from_dict(replay_dict["master"]),
+            [ReplayProject.from_dict(project_dict for project_dict in replay_dict["wp"])],
+        )
+
+    @staticmethod
+    def from_context(ctx, wp_config: WPConfig, wp_new: set[str]) -> "Replay":
+
+        master_project = ReplayProject.from_project_directory(ctx.master_dir)
+
+        wp_projects = []
+        for wp_name in wp_config.wp_names:
+            wp_dir = os.path.join(ctx.tmp_dir, "wp-" + wp_name.name)
+            if wp_name.name in wp_new:
+                # the MM project does not exist yet
+                wp_projects.append(ReplayProject(wp_name.name, wp_name.mergin_project, ""))
+            else:
+                wp_projects.append(ReplayProject.from_project_directory(wp_dir, wp_name.name))
+
+        return Replay(master_project, wp_projects)

workpackages/wp_utils.py

@@ -16,16 +16,20 @@ def escape_double_quotes(name):
 
 
 def download_project_with_cache(mc, project_path, directory, cache_dir, version=None, server_latest_version=None):
+    """Download project or pull latest changes if the project exists in the cache"""
     if not cache_dir:
+        print(f"Downloading '{project_path}' (cache not enabled)")
         mc.download_project(project_path, directory, version=version)
         return
     project_namespace, project_name = project_path.split("/")
     project_cache_dir = os.path.join(cache_dir, f"{project_namespace}_{project_name}")
     if not os.path.exists(project_cache_dir):
+        print(f"Downloading '{project_path}' project (cache miss)")
         mc.download_project(project_path, project_cache_dir, version=version)
     else:
         mp = mergin.MerginProject(project_cache_dir)
         if server_latest_version is None or mp.version() != server_latest_version:
+            print(f"Pulling '{project_path}' project from {mp.version()} to {server_latest_version}")
             mc.pull_project(project_cache_dir)
         else:
             print(f"Local and server project versions are the same. Pulling '{project_path}' project skipped")