Files
GeoData/geodata_pipeline/setup_helpers.py

63 lines
1.9 KiB
Python

from __future__ import annotations
import glob
import os
import shutil
import zipfile
from typing import Iterable
from .config import Config, ensure_default_config
def ensure_directories(cfg: Config) -> None:
for path in _paths_from_config(cfg):
os.makedirs(path, exist_ok=True)
def _paths_from_config(cfg: Config) -> Iterable[str]:
return [
cfg.raw.dgm1_dir,
cfg.raw.dop20_dir,
cfg.raw.citygml_lod1_dir,
cfg.raw.citygml_lod2_dir,
cfg.archives.dgm1_dir,
cfg.archives.dop20_dir,
cfg.archives.citygml_lod1_dir,
cfg.archives.citygml_lod2_dir,
cfg.work.work_dir,
cfg.export.heightmap_dir,
cfg.export.ortho_dir,
]
def materialize_archives(cfg: Config) -> None:
"""Best-effort expansion of archive zips into raw inputs."""
ensure_directories(cfg)
_unpack_all(cfg.archives.dgm1_dir, cfg.raw.dgm1_dir)
_unpack_all(cfg.archives.citygml_lod1_dir, cfg.raw.citygml_lod1_dir)
_unpack_all(cfg.archives.citygml_lod2_dir, cfg.raw.citygml_lod2_dir)
_unpack_all(cfg.archives.dop20_dir, cfg.raw.dop20_dir)
_copy_filelist(
os.path.join(cfg.archives.dop20_dir, "filelist.txt"),
os.path.join(os.path.dirname(cfg.raw.dop20_dir), "filelist.txt"),
)
def _unpack_all(archive_dir: str, dest_dir: str) -> None:
os.makedirs(dest_dir, exist_ok=True)
for zpath in glob.glob(os.path.join(archive_dir, "*.zip")):
print(f"Unpacking {zpath} -> {dest_dir}")
with zipfile.ZipFile(zpath, "r") as zf:
zf.extractall(dest_dir)
def _copy_filelist(src: str, dest: str) -> None:
if not os.path.exists(src):
return
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.copy2(src, dest)
print(f"Copied filelist: {src} -> {dest}")
__all__ = ["ensure_directories", "materialize_archives", "ensure_default_config"]