vinca-snapshot --rosdistro-ref: snapshot a ros/rosdistro git ref (e.g. a sync tag)

vinca/snapshot.py

@@ -4,13 +4,47 @@
 import yaml
 from rosdistro import (
     DistributionCache,
+    get_cached_distribution,
     get_distribution_cache_string,
     get_index,
     get_index_url,
 )
+from rosdistro.loader import load_url
 
 from .distro import Distro
 
+ROSDISTRO_RAW_URL = "https://raw.githubusercontent.com/ros/rosdistro"
+
+
+def load_distribution_cache(distro_name, rosdistro_ref=None):
+    """The rosdistro distribution cache, optionally for a git ref of ros/rosdistro.
+
+    The published cache (repo.ros2.org) always follows rosdistro's default branch.
+    With a ref, e.g. a sync tag such as ``jazzy/2026-10-05``, the distribution file
+    of that ref replaces the cached one: package.xml entries whose release changed
+    are dropped and fetched again for the version of the ref, everything else is
+    reused from the published cache.
+    """
+    cache_yaml = get_distribution_cache_string(get_index(get_index_url()), distro_name)
+    cache = DistributionCache(distro_name, yaml.safe_load(cache_yaml))
+    if not rosdistro_ref:
+        return cache
+
+    index = get_index(f"{ROSDISTRO_RAW_URL}/{rosdistro_ref}/index-v4.yaml")
+    urls = index.distributions[distro_name]["distribution"]
+    urls = urls if isinstance(urls, list) else [urls]
+    data = [yaml.safe_load(load_url(url)) for url in urls]
+    cache.update_distribution(data)
+
+    # fetch the package.xml files the update dropped (only changed releases)
+    dist = get_cached_distribution(
+        index, distro_name, cache=cache, allow_lazy_load=True
+    )
+    for pkg_name in sorted(dist.release_packages):
+        if pkg_name not in cache.release_package_xmls:
+            dist.get_release_package_xml(pkg_name)
+    return cache
+
 
 def main():
     parser = argparse.ArgumentParser(
@@ -43,6 +77,15 @@ def main():
         help="Output file to write dependencies to",
         required=False,
     )
+    parser.add_argument(
+        "--rosdistro-ref",
+        type=str,
+        dest="rosdistro_ref",
+        default=None,
+        help="git ref of ros/rosdistro to snapshot, e.g. a sync tag such as "
+        "'jazzy/2026-10-05' (default: the published cache, i.e. the latest state)",
+        required=False,
+    )
     parser.add_argument(
         "-q",
         "--quiet",
@@ -60,9 +103,7 @@ def main():
     # the dependency metadata stored in the snapshot.
     # TODO: Consider referencing immutable cache release assets instead if they
     # become available through https://github.com/ros/rosdistro/pull/50112.
-    index = get_index(get_index_url())
-    cache_yaml = get_distribution_cache_string(index, args.distro)
-    cache = DistributionCache(args.distro, yaml.safe_load(cache_yaml))
+    cache = load_distribution_cache(args.distro, args.rosdistro_ref)
     distro = Distro(args.distro, distribution_cache=cache)
 
     if args.package is None:
@@ -104,7 +145,8 @@ def main():
             print("{0:{2}} {1}".format(dep, version, max_len + 2))
 
     with open(args.output, "w") as f:
+        source = f" at rosdistro {args.rosdistro_ref}" if args.rosdistro_ref else ""
         f.write(
-            f"# Snapshot generated by vinca-snapshot on {utc_time} UTC for distro {args.distro}\n"
+            f"# Snapshot generated by vinca-snapshot on {utc_time} UTC for distro {args.distro}{source}\n"
         )
         yaml.dump(output, f)