geodata

add_info(metadata, info)

Add the :py:func:ogrinfo output to the metadata returned by :py:func:download_extract.

Parameters:

Name	Type	Description	Default
metadata	dict	Metadata returned by :py:func:download_extract.	required
info	dict	:py:func:ogrinfo output.	required

Source code in packages/common/src/bag3d/common/utils/geodata.py

def add_info(metadata: dict, info: dict) -> None:
    """Add the :py:func:`ogrinfo` output to the metadata returned by
    :py:func:`download_extract`.

    Args:
        metadata: Metadata returned by :py:func:`download_extract`.
        info: :py:func:`ogrinfo` output.
    """
    for layername, layerinfo in info.items():
        layer = layername.lower()
        metadata.update(layerinfo)
        for dt, ft in metadata["timeliness"].items():
            if layer in ft:
                metadata[f"Timeliness [{layer}]"] = dt
    del metadata["timeliness"]

attributes_dict(attributes_str)

[ { "name": attribute name, "type": value type, "constraints" value constraints }, ... ]

Source code in packages/common/src/bag3d/common/utils/geodata.py

def attributes_dict(attributes_str: str) -> List[dict]:
    """
    [
        {
        "name": attribute name,
        "type": value type,
        "constraints" value constraints
        },
    ...
    ]
    """
    ret = []
    _astr = attributes_str.strip("\n").strip().split("\n")
    for i in _astr:
        # Skip empty lines
        if not i.strip():
            continue

        adict = {}

        # Split on first colon to get attribute name and rest
        parts = i.split(":", 1)

        # reject lines without at least one colon
        if len(parts) != 2:
            continue

        # get the name
        aname = parts[0].strip()
        specs_full = parts[1].strip()

        # Check if there's a comment and remove it
        if ", comment=" in specs_full:
            specs = specs_full.split(", comment=", 1)[0]
        else:
            specs = specs_full

        try:
            # Handle different formats:
            # 'String (0.0) NOT NULL'
            # 'Date'
            # 'Integer(Boolean) (0.0)'
            if ")" in specs and " " in specs:
                # Find the last closing parenthesis and split there
                last_paren = specs.rfind(")")
                atype = specs[: last_paren + 1]
                constraints = specs[last_paren + 1 :].strip()
            else:
                # Simple type like 'Date', 'DateTime'
                atype = specs.strip()
                constraints = ""

        except ValueError:
            # 'inOnderzoek: Integer(Boolean) (0.0)'
            atype = specs.strip()
            constraints = ""

        adict["name"] = aname
        adict["type"] = atype

        if constraints == "" or constraints is None:
            adict["constraints"] = None
        else:
            adict["constraints"] = TableColumnConstraints(other=[constraints.strip()])
        ret.append(adict)
    return ret

bbox_from_wkt(wkt)

Returns the BBOX of a WKT Polygon.

Source code in packages/common/src/bag3d/common/utils/geodata.py

def bbox_from_wkt(wkt):
    """Returns the BBOX of a WKT Polygon."""
    re_polygon = re.compile(r"(?<=polygon).*", re.IGNORECASE)
    m = re_polygon.search(wkt)
    if m:
        ext = m.group().strip().strip("(").strip(")").split(")")[0]
        cstr = ext.split(",")
        fcoord = tuple(map(float, cstr[0].strip(" ").split(" ")))
        minx, miny = fcoord
        maxx, maxy = fcoord
        for coord in cstr:
            x, y = tuple(map(float, coord.strip(" ").split(" ")))
            minx = x if x < minx else minx
            miny = y if y < miny else miny
            maxx = x if x > maxx else maxx
            maxy = y if y > maxy else maxy
        return minx, miny, maxx, maxy
    else:
        return None

ogr2postgres(gdal_runner, dsn, dataset, extract_path, feature_type, xsd, new_table, context=None)

ogr2ogr a layer from zipped data extract from GML into Postgres.

It was developed for loading the TOP10NL and BGT extracts that are downloaded from the PDOK API.

Parameters:

Name	Type	Description	Default
gdal_runner	CommandRunner	CommandRunner for GDAL tools.	required
dsn	str	PostgreSQL connection string.	required
dataset	str	Name of the dataset ('top10nl', 'bgt').	required
extract_path	Path	Local path to the zipped extract.	required
feature_type	str	The feature layer to load from the dataset	required
xsd	str	Path (URL) to the XSD file.	required
new_table	PostgresTableIdentifier	The name of the new Postgres table to load the data into.	required
context	OpExecutionContext	Optional op execution context from Dagster for metadata retrieval.	None

Returns: Runs :py:func:postgrestable_metadata on return and returns a dict of metadata of the new_table loaded with data.

Source code in packages/common/src/bag3d/common/utils/geodata.py

def ogr2postgres(
    gdal_runner: CommandRunner,
    dsn: str,
    dataset: str,
    extract_path: Path,
    feature_type: str,
    xsd: str,
    new_table: PostgresTableIdentifier,
    context: OpExecutionContext = None,
) -> dict:
    """ogr2ogr a layer from zipped data extract from GML into Postgres.

    It was developed for loading the TOP10NL and BGT extracts that are downloaded from
    the PDOK API.

    Args:
        gdal_runner: CommandRunner for GDAL tools.
        dsn: PostgreSQL connection string.
        dataset: Name of the dataset ('top10nl', 'bgt').
        extract_path: Local path to the zipped extract.
        feature_type: The feature layer to load from the ``dataset``
        xsd: Path (URL) to the XSD file.
        new_table: The name of the new Postgres table to load the data into.
        context: Optional op execution context from Dagster for metadata retrieval.
    Returns:
        Runs :py:func:`postgrestable_metadata` on return and returns a dict of metadata
        of the ``new_table`` loaded with data.
    """
    cmd = " ".join(
        [
            "{exe}",
            "--config PG_USE_COPY=YES",
            "-overwrite",
            "-nln {new_table}",
            '-oo XSD="{xsd}"',
            "-oo WRITE_GFS=NO",
            "-lco UNLOGGED=ON",
            "-lco SPATIAL_INDEX=NONE",
            "-lco GEOMETRY_NAME=wkb_geometry",
            '-f PostgreSQL PG:"{dsn}"',
            "/vsizip/{local_path}/{dataset}_{feature_type}.gml {feature_type}",
        ]
    )

    kwargs = {
        "new_table": new_table,
        "feature_type": feature_type,
        "dsn": dsn,
        "xsd": xsd,
        "dataset": dataset,
    }
    result = gdal_runner.run(
        cmd,
        exe_name="ogr2ogr",
        kwargs=kwargs,
        local_path=extract_path,
        context=context,
    )
    if result.success:
        if context:
            return postgrestable_metadata(context, new_table)
        else:
            return {"Database.Schema.Table": f"{new_table.schema}.{new_table.table}"}

ogrinfo(gdal_runner, dataset, extract_path, feature_types, xsd, context=None)

Runs ogrinfo on the zipped extract.

Source code in packages/common/src/bag3d/common/utils/geodata.py

def ogrinfo(
    gdal_runner: CommandRunner,
    dataset: str,
    extract_path: Path,
    feature_types: list,
    xsd: str,
    context: OpExecutionContext = None,
):
    """Runs ogrinfo on the zipped extract."""
    cmd = " ".join(
        [
            "{exe}",
            "-so",
            "-al",
            '-oo XSD="{xsd}"',
            "-oo WRITE_GFS=NO",
            "/vsizip/{local_path}/{dataset}_{feature_type}.gml {feature_type}",
        ]
    )

    info = {}
    for feature_type in feature_types:
        kwargs = {"xsd": xsd, "dataset": dataset, "feature_type": feature_type}
        result = gdal_runner.run(
            cmd,
            exe_name="ogrinfo",
            kwargs=kwargs,
            local_path=extract_path,
            context=context,
        )
        if result.success:
            layername, layerinfo = parse_ogrinfo(result.stdout, feature_type)
            info[str(layername)] = layerinfo
    return info

parse_ogrinfo(ogrinfo_stdout, feature_type)

Parses the stdout of ogrinfo into a dictionary.

Source code in packages/common/src/bag3d/common/utils/geodata.py

def parse_ogrinfo(ogrinfo_stdout: str, feature_type: str) -> (str, dict):
    """Parses the stdout of ogrinfo into a dictionary."""
    layerinfo = {}
    inf = ogrinfo_stdout.split("Layer name: ")[1]
    layername = inf.split("\n")[0].lower()
    if layername != feature_type:
        raise ValueError(
            f"Encountered a {layername} layer, but expected {feature_type}"
        )

    re_feature_count = re.compile(r"(?<=Feature Count: )\d+")
    ft = feature_type.lower()
    layerinfo[f"Feature Count [{ft}]"] = int(re_feature_count.search(inf)[0])
    layerinfo[f"Extent [{ft}]"] = dict(
        (geom, wkt) for geom, wkt in parse_ogrinfo_extent(inf)
    )
    schema = parse_ogrinfo_attributes(inf[inf.find("gml_id") :])
    layerinfo[f"Schema [{ft}]"] = TableSchemaMetadataValue(schema)
    return layername, layerinfo

parse_ogrinfo_attributes(attributes_str)

Parses the attributes list of the ogrinfo stdout into a :py:class:TableSchema

Source code in packages/common/src/bag3d/common/utils/geodata.py

def parse_ogrinfo_attributes(attributes_str: str) -> TableSchema:
    """Parses the attributes list of the ogrinfo stdout into a :py:class:`TableSchema`"""
    alist = attributes_dict(attributes_str)
    schema = attributes_schema(alist)
    return schema

pdal_info(pdal, file_path, with_all=False, verbose=False, context=None)

Run 'pdal info' on a point cloud file.

Parameters:

Name	Type	Description	Default
verbose	bool	Return stdout/stderr from pdal	False
pdal		The pdal runner or AppImage executable.	required
file_path	Path	Path to the point cloud file.	required
with_all	bool	If true, run pdal info --all, else run pdal info --metadata. Defaults to False.	False
context		Optional Dagster context for Pipes support.	None

Returns: A tuple of (pdal's return code, parsed pdal info output)

Source code in packages/common/src/bag3d/common/utils/geodata.py

def pdal_info(
    pdal, file_path: Path, with_all: bool = False, verbose: bool = False, context=None
) -> Tuple[int, dict]:
    """Run 'pdal info' on a point cloud file.

    Args:
        verbose: Return stdout/stderr from pdal
        pdal: The pdal runner or AppImage executable.
        file_path: Path to the point cloud file.
        with_all: If true, run ``pdal info --all``, else run ``pdal info --metadata``.
            Defaults to ``False``.
        context: Optional Dagster context for Pipes support.
    Returns:
        A tuple of (pdal's return code, parsed pdal info output)
    """
    cmd_list = [
        "{exe}",
        "info",
    ]
    cmd_list.append("--all") if with_all else cmd_list.append("--metadata")
    cmd_list.append("{local_path}")

    result = pdal.run(
        " ".join(cmd_list),
        exe_name="pdal",
        local_path=file_path,
        context=context,
    )

    output = result.stdout
    if "Global encoding WKT flag" in str(output):
        logger.warning(f"Pdal failed for tile {file_path} with output {output}.")
        # Remove the first line
        output_lines = output.split("\n")
        if len(output_lines) > 1:
            logger.warning(f"Removing first line from PDAL output : {output_lines[0]}")
            output = "\n".join(output_lines[1:])

    output_processed = output.replace("\\u0000", "")

    try:
        json_data = json.loads(output_processed)
    except JSONDecodeError as e:
        raise Failure(f"Failed to make JSON from pdal output: {output_processed}. {e}")

    return result.returncode, json_data