Add snapshot system

setup.cfg

@@ -42,6 +42,7 @@ console_scripts =
     vinca-gha = vinca.generate_gha:main
     vinca-azure = vinca.generate_azure:main
     vinca-migrate = vinca.migrate:main
+    vinca-snapshot = vinca.snapshot:main
 
 [flake8]
 import-order-style = google

vinca/distro.py

@@ -6,10 +6,11 @@
 
 
 class Distro(object):
-    def __init__(self, distro_name, python_version=None):
+    def __init__(self, distro_name, python_version=None, snapshot=None):
         index = get_index(get_index_url())
         self._distro = get_cached_distribution(index, distro_name)
         self.distro_name = distro_name
+        self.snapshot = snapshot
         # set up ROS environments
         if python_version is None:
             python_version = index.distributions[distro_name]["python_version"]
@@ -60,20 +61,29 @@ def get_depends(self, pkg, ignore_pkgs=None):
         return dependencies
 
     def get_released_repo(self, pkg_name):
+        if self.snapshot and pkg_name in self.snapshot:
+            return (
+                self.snapshot[pkg_name].get("url", None),
+                self.snapshot[pkg_name].get("tag", None),
+            )
+
         pkg = self._distro.release_packages[pkg_name]
         repo = self._distro.repositories[pkg.repository_name].release_repository
         release_tag = get_release_tag(repo, pkg_name)
         return repo.url, release_tag
 
     def check_package(self, pkg_name):
         if pkg_name in self._distro.release_packages:
-            return True
+            return self.snapshot is None or pkg_name in self.snapshot
         elif pkg_name in self.build_packages:
             return True
         else:
             return False
 
     def get_version(self, pkg_name):
+        if self.snapshot and pkg_name in self.snapshot:
+            return self.snapshot[pkg_name].get("version", None)
+
         pkg = self._distro.release_packages[pkg_name]
         repo = self._distro.repositories[pkg.repository_name].release_repository
         return repo.version.split("-")[0]
@@ -86,3 +96,6 @@ def check_ros1(self):
 
     def get_python_version(self):
         return self._python_version
+
+    def get_package_names(self):
+        return self._distro.release_packages.keys()

vinca/main.py

@@ -126,6 +126,13 @@ def parse_command_line(argv):
         default=None,
         help="The conda platform to check existing recipes for.",
     )
+    parser.add_argument(
+        "-z",
+        "--snapshot",
+        dest="snapshot",
+        default=None,
+        help="The version snapshot file (default: None)."
+    )
     arguments = parser.parse_args(argv[1:])
     global selected_platform
     config.parsed_args = arguments
@@ -208,6 +215,15 @@ def read_vinca_yaml(filepath):
     return vinca_conf
 
 
+def read_snapshot(filepath):
+    if not filepath:
+        return None
+
+    yaml = ruamel.yaml.YAML()
+    snapshot = yaml.load(open(filepath, "r"))
+    return snapshot
+
+
 def generate_output(pkg_shortname, vinca_conf, distro, version, all_pkgs=None):
     if not all_pkgs:
         all_pkgs = []
@@ -528,21 +544,14 @@ def get_pkg(pkg_name):
     return outputs
 
 
-def get_version(distro, vinca_conf, pkg_shortname):
-    version = distro.get_version(pkg_shortname)
-    if (
-        vinca_conf.get("package_version")
-        and vinca_conf["package_version"][pkg_shortname]
-    ):
-        version = vinca_conf["package_version"][pkg_shortname]["version"]
-
-    return version
-
-
 def generate_outputs_version(distro, vinca_conf):
     outputs = []
     for pkg_shortname in vinca_conf["_selected_pkgs"]:
-        version = get_version(distro, vinca_conf, pkg_shortname)
+        if not distro.check_package(pkg_shortname):
+            print(f"Could not generate output for {pkg_shortname}")
+            continue
+
+        version = distro.get_version(pkg_shortname)
         output = generate_output(pkg_shortname, vinca_conf, distro, version)
         if output is not None:
             outputs.append(output)
@@ -562,7 +571,7 @@ def generate_source(distro, vinca_conf):
         entry["git_url"] = url
         entry["git_rev"] = version
         pkg_names = resolve_pkgname(pkg_shortname, vinca_conf, distro)
-        pkg_version = get_version(distro, vinca_conf, pkg_shortname)
+        pkg_version = distro.get_version(pkg_shortname)
         print("Checking ", pkg_shortname, pkg_version)
         if not pkg_names:
             continue
@@ -602,12 +611,6 @@ def generate_source_version(distro, vinca_conf):
             continue
 
         url, version = distro.get_released_repo(pkg_shortname)
-        if (
-            vinca_conf["package_version"]
-            and vinca_conf["package_version"][pkg_shortname]
-        ):
-            url = vinca_conf["package_version"][pkg_shortname]["url"]
-            version = vinca_conf["package_version"][pkg_shortname]["version"]
 
         entry = {}
         entry["git_url"] = url
@@ -865,6 +868,7 @@ def main():
     base_dir = os.path.abspath(arguments.dir)
     vinca_yaml = os.path.join(base_dir, "vinca.yaml")
     vinca_conf = read_vinca_yaml(vinca_yaml)
+    snapshot = read_snapshot(arguments.snapshot)
 
     from .template import generate_bld_ament_cmake
     from .template import generate_bld_ament_python
@@ -879,7 +883,7 @@ def main():
     generate_bld_colcon_merge()
     generate_bld_catkin_merge()
     generate_activate_hook()
-    
+
     if arguments.trigger_new_versions:
         vinca_conf["trigger_new_versions"] = True
     else:
@@ -892,7 +896,7 @@ def main():
         if "python_version" in vinca_conf:
             python_version = vinca_conf["python_version"]
 
-        distro = Distro(vinca_conf["ros_distro"], python_version)
+        distro = Distro(vinca_conf["ros_distro"], python_version, snapshot)
         additional_pkgs, parsed_pkgs = [], []
         for f in pkg_files:
             parsed_pkg = catkin_pkg.package.parse_package(f)
@@ -994,7 +998,7 @@ def main():
         if "python_version" in vinca_conf:
             python_version = vinca_conf["python_version"]
 
-        distro = Distro(vinca_conf["ros_distro"], python_version)
+        distro = Distro(vinca_conf["ros_distro"], python_version, snapshot)
 
         selected_pkgs = get_selected_packages(distro, vinca_conf)
 

vinca/snapshot.py

@@ -0,0 +1,75 @@
+import argparse
+import yaml
+from .distro import Distro
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Dependency snapshotting tool for ROS packages"
+    )
+    parser.add_argument(
+        "-d",
+        "--distro",
+        type=str,
+        dest="distro",
+        default="humble",
+        help="ROS distribution to use (default: humble)",
+        required=False,
+    )
+    parser.add_argument(
+        "-p",
+        "--package",
+        type=str,
+        dest="package",
+        default=None,
+        help="ROS package to get dependencies for (default: ALL)",
+        required=False,
+    )
+    parser.add_argument(
+        "-o",
+        "--output",
+        type=str,
+        dest="output",
+        default="snapshot.yaml",
+        help="Output file to write dependencies to",
+        required=False,
+    )
+    parser.add_argument(
+        "-q",
+        "--quiet",
+        dest="quiet",
+        action="store_true",
+        help="Suppress output to stdout",
+        required=False,
+    )
+    args = parser.parse_args()
+
+    distro = Distro(args.distro)
+
+    if args.package is None:
+        deps = distro.get_package_names()
+    else:
+        deps = distro.get_depends(args.package)
+        deps.add(args.package)
+
+    if not args.quiet:
+        max_len = max([len(dep) for dep in deps])
+        print("\033[1m{0:{2}} {1}\033[0m".format("Package", "Version", max_len + 2))
+
+    output = {}
+
+    for dep in deps:
+        try:
+            url, tag = distro.get_released_repo(dep)
+            version = distro.get_version(dep)
+        except AttributeError:
+            print("\033[93mPackage '{}' has no version set, skipping...\033[0m".format(dep))
+            continue
+
+        output[dep] = {"url": url, "version": version, "tag": tag}
+
+        if not args.quiet:
+            print("{0:{2}} {1}".format(dep, version, max_len + 2))
+
+    with open(args.output, "w") as f:
+        yaml.dump(output, f)