Cache export CLI command (#94)

epymorph/cli/cache.py

@@ -65,6 +65,31 @@ def define_argparser(command_parser: _SubParsersAction):
         help='clear the cache')
     clear_command.set_defaults(handler=lambda args: clear())
 
+    export_command = sp.add_parser(
+        'export',
+        help='export geo as a .geo.tar file')
+    export_command.add_argument(
+        'geo',
+        type=str,
+        help='the name of a geo or the path to a geo spec file')
+    export_command.add_argument(
+        '-o', '--out',
+        type=str,
+        help='(optional) the directory in which to write the file')
+    export_command.add_argument(
+        '-r', '--rename',
+        type=str,
+        help='(optional) an override for the name of the file')
+    export_command.add_argument(
+        '-i', '--ignore_cache',
+        action='store_true',
+        help='(optional) do not add this geo to the local cache')
+    export_command.set_defaults(handler=lambda args: export(
+        geo_name_or_path=args.geo,
+        out=args.out,
+        rename=args.rename,
+        ignore_cache=args.ignore_cache
+    ))
 
 # Exit codes:
 # - 0 success
@@ -103,6 +128,25 @@ def fetch(geo_name_or_path: str, force: bool) -> int:
     return 0  # exit code: success
 
 
+def export(geo_name_or_path: str, out: str | None, rename: str | None, ignore_cache: bool) -> int:
+    """CLI command handler: export compressed geo to a location outside the cache."""
+    # split geo name and path
+    if geo_name_or_path in geo_library_dynamic or os.path.exists(cache.CACHE_PATH / F.to_archive_filename(geo_name_or_path)):
+        geo_name = geo_name_or_path
+        geo_path = cache.CACHE_PATH / F.to_archive_filename(geo_name)
+    elif os.path.exists(Path(geo_name_or_path).expanduser()):
+        geo_path = Path(geo_name_or_path).expanduser()
+        geo_name = geo_path.stem
+    else:
+        raise cache.GeoCacheException("Specified geo not found.")
+
+    cache.export(geo_name, geo_path, out, rename, ignore_cache)
+
+    print("Geo successfully exported.")
+
+    return 0  # exit code: success
+
+
 def remove(geo_name: str) -> int:
     """CLI command handler: remove geo from cache"""
     try:

epymorph/geo/adrio/__init__.py

@@ -3,5 +3,5 @@
 from epymorph.geo.dynamic import ADRIOMaker
 
 adrio_maker_library: dict[str, type[ADRIOMaker]] = {
-    'Census': ADRIOMakerCensus
+    'Census': ADRIOMakerCensus,
 }

epymorph/geo/cache.py

@@ -50,6 +50,56 @@ def fetch(geo_name_or_path: str) -> None:
             raise GeoCacheException(f'spec file at {geo_name_or_path} not found.')
 
 
+def export(geo_name: str, geo_path: Path, out: str | None, rename: str | None, ignore_cache: bool) -> None:
+    """
+    Exports a geo as a .geo.tar file to a location outside the cache.
+    If uncached, geo to export is also cached.
+    User can specify a destination path and new name for exported geo.
+    Raises a GeoCacheException if geo not found.
+    """
+    # check for out path specified
+    if out is not None:
+        if not os.path.exists(out):
+            raise GeoCacheException(f'specified output directory {out} not found.')
+        else:
+            out_dir = Path(out)
+    else:
+        out_dir = Path(os.getcwd())
+
+    # check for geo name specified
+    if rename is not None:
+        geo_exp_name = rename
+    else:
+        geo_exp_name = geo_name
+
+    out_path = out_dir / F.to_archive_filename(geo_exp_name)
+    cache_file_path = CACHE_PATH / F.to_archive_filename(geo_name)
+    cache_out_file_path = CACHE_PATH / F.to_archive_filename(geo_exp_name)
+
+    # if cached, copy cached file
+    if os.path.exists(cache_file_path):
+        geo = load_from_cache(geo_name)
+        if geo is not None:
+            geo.save(out_path)
+
+    # if geo uncached or spec file passed, fetch and export
+    elif geo_name in geo_library_dynamic or os.path.exists(geo_path):
+        if geo_name in geo_library_dynamic:
+            geo_load = geo_library_dynamic.get(geo_name)
+            if geo_load is not None:
+                geo = geo_load()
+        else:
+            geo = DF.load_from_spec(geo_path, adrio_maker_library)
+        with dynamic_geo_messaging(geo):
+            static_geo = convert_to_static_geo(geo)
+        if not ignore_cache:
+            static_geo.save(cache_out_file_path)
+        static_geo.save(out_path)
+
+    else:
+        raise GeoCacheException("Geo to export not found.")
+
+
 def remove(geo_name: str) -> None:
     """
     Removes a geo's data from the cache.
@@ -88,7 +138,7 @@ def save_to_cache(geo: Geo, geo_name: str) -> None:
     F.save_as_archive(static_geo, file_path)
 
 
-def load_from_cache(geo_name: str) -> Geo | None:
+def load_from_cache(geo_name: str) -> StaticGeo | None:
     """Checks whether a dynamic geo has already been cached and returns it if so."""
     file_path = CACHE_PATH / F.to_archive_filename(geo_name)
     if not os.path.exists(file_path):