files

Working with file inputs and outputs

BadArchiveError

Bases: OSError

The archive contains a bad file

Source code in bag3d/common/utils/files.py

class BadArchiveError(OSError):
    """The archive contains a bad file"""

    pass

bag3d_dir(root_dir)

The 3D BAG data directory

Source code in bag3d/common/utils/files.py

def bag3d_dir(root_dir: os.PathLike) -> Path:
    """The 3D BAG data directory"""
    return Path(root_dir) / "3DBAG"

bag3d_export_dir(root_dir, version)

Create the 3DBAG export directory if does not exist

Source code in bag3d/common/utils/files.py

def bag3d_export_dir(root_dir: os.PathLike, version: str) -> Path:
    """Create the 3DBAG export directory if does not exist"""
    export_dir = bag3d_dir(root_dir) / f"export_{version}"
    export_dir.mkdir(exist_ok=True)
    return export_dir

check_export_results(path_quadtree_tsv, path_tiles_dir)

Parse the quadtree.tsv written by tyler, check if all formats exists for each tile, add the tile WKT.

Returns:

Type	Description
Iterator[ExportResult]	Generator of ExportResult

Source code in bag3d/common/utils/files.py

def check_export_results(
    path_quadtree_tsv: Path, path_tiles_dir: Path
) -> Iterator[ExportResult]:
    """Parse the `quadtree.tsv` written by *tyler*, check if all formats exists for each
    tile, add the tile WKT.

    Returns:
         Generator of ExportResult
    """
    with path_quadtree_tsv.open("r") as fo:
        csvreader = csv.DictReader(fo, delimiter="\t")
        for row in csvreader:
            if row["leaf"] == "true" and int(row["nr_items"]) > 0:
                leaf_id = row["id"]
                leaf_id_in_filename = leaf_id.replace("/", "-")
                leaf_dir = path_tiles_dir.joinpath(leaf_id)
                if leaf_dir.exists():
                    obj_paths = tuple(
                        p for p in leaf_dir.iterdir() if p.suffix == ".obj"
                    )
                    basename = path_tiles_dir.joinpath(leaf_id, leaf_id_in_filename)
                    yield ExportResult(
                        tile_id=leaf_id,
                        cityjson_path=basename.with_suffix(".city.json"),
                        gpkg_path=basename.with_suffix(".gpkg"),
                        obj_paths=obj_paths,
                        wkt=row["wkt"],
                    )

geoflow_crop_dir(root_dir)

Directory for the Geoflow crop-reconstruct output

Source code in bag3d/common/utils/files.py

def geoflow_crop_dir(root_dir: os.PathLike) -> Path:
    """Directory for the Geoflow crop-reconstruct output"""
    return bag3d_dir(root_dir) / "crop_reconstruct"

get_export_tile_ids()

Get the IDs of the distribution tiles from the file system. It reads the quadtree.tsv output from tyler and extracts the IDs of the leaf tiles.

Fixme

• Currently we read the root data dir from the BAG3D_EXPORT_DIR environment variable, or use /data/3DBAG/export_{version} as default. Maybe we could consolidate all resource configurations to .env files and load from there in all places. But need to be able to load different .env files and production and dev setup.

Returns:

Type	Description
Sequence[str]	List of tile IDs

Source code in bag3d/common/utils/files.py

def get_export_tile_ids() -> Sequence[str]:
    """Get the IDs of the distribution tiles from the file system.
    It reads the `quadtree.tsv` output from *tyler* and extracts the IDs of the
    leaf tiles.

    Fixme:
        * Currently we read the root data dir from the BAG3D_EXPORT_DIR environment
        variable, or use /data/3DBAG/export_{version} as default. Maybe we could consolidate all resource
        configurations to .env files and load from there in all places. But need to be
        able to load different .env files and production and dev setup.

    Returns:
        List of tile IDs
    """
    tileids = []
    version = os.getenv("BAG3D_RELEASE_VERSION", "test_version")
    export_dir = Path(os.getenv("BAG3D_EXPORT_DIR", f"/data/3DBAG/export_{version}"))
    if export_dir.exists():
        path_tiles_dir = export_dir.joinpath("tiles")
        path_quadtree_tsv = export_dir.joinpath("quadtree.tsv")
        if path_quadtree_tsv.exists():
            tileids = [
                er.tile_id
                for er in check_export_results(path_quadtree_tsv, path_tiles_dir)
            ]
    else:
        raise FileNotFoundError(f"""Export directory {export_dir} does not exist.
                                You need to set the BAG3D_EXPORT_DIR variable.""")

    return tileids

unzip(file, dest, remove=True)

Uncompress the whole zip archive and optionally delete the zip.

Parameters:

Name	Type	Description	Default
file	Path	The Path to the zip.	required
dest	Path	The Path to the destination directory.	required
remove	bool	Whether to remove the zip.	True

Raises:

Type	Description
BadArchiveError	The archive contains at least one bad file

Source code in bag3d/common/utils/files.py

def unzip(file: Path, dest: Path, remove: bool = True) -> None:
    """Uncompress the whole zip archive and optionally delete the zip.

    Args:
        file: The Path to the zip.
        dest: The Path to the destination directory.
        remove: Whether to remove the zip.

    Raises:
        BadArchiveError: The archive contains at least one bad file
    """
    logger = get_dagster_logger()
    logger.info(f"Uncompressing {file} to {dest}")
    with ZipFile(file, "r") as ezip:
        first_bad_file = ezip.testzip()
        if first_bad_file:
            raise BadArchiveError(
                f"The archive contains at least one bad file: {first_bad_file}"
            )
        ezip.extractall(path=dest)
    if remove:
        logger.info(f"Deleting {file}")
        file.unlink()