diff --git a/dcpy/lifecycle/package/esri.py b/dcpy/lifecycle/package/esri.py index f096c664bc..dc30a140a7 100644 --- a/dcpy/lifecycle/package/esri.py +++ b/dcpy/lifecycle/package/esri.py @@ -5,6 +5,11 @@ import yaml import dcpy.models.product.dataset.metadata as models +from dcpy.lifecycle import product_metadata +from dcpy.models.product.metadata import OrgMetadata +from dcpy.utils.geospatial import esri_metadata, fgdb +from dcpy.utils.geospatial.esri_metadata import _create_attr_metadata +from dcpy.utils.geospatial.shapefile import Shapefile from dcpy.utils.logging import logger @@ -83,3 +88,141 @@ def parse_pdf_text( output_path = output_path or Path("columns.yml") with open(output_path, "w") as outfile: yaml.dump(fields, outfile, sort_keys=False) + + +@app.command("write_metadata") +def _write_metadata( + product_name: str = typer.Argument(..., help="Name of product. Example: 'lion'"), + dataset_name: str = typer.Argument( + ..., help="Name of dataset. Example: 'pseudo-lots'" + ), + file_id: str = typer.Argument( + ..., + help="Identifier from within the org. metadata reference. Example: 'mappluto_unclipped_gdb'", + ), + path_to_file: Path = typer.Argument(..., help="Path to file."), + layer: str | None = typer.Argument(None, help="Name of layer within file"), + org_md_path: Path | None = typer.Option( + None, + "--org-md-path", + help="Path to organizational metadata", + ), + zip_subdir: str | None = typer.Option( + None, + "--zip-subdir", + help="Directory structure within zip file, if relevant", + ), +): + write_metadata( + product_name=product_name, + dataset_name=dataset_name, + path_to_file=path_to_file, + layer=layer, + file_id=file_id, + zip_subdir=zip_subdir, + org_md=org_md_path, + ) + + +def write_metadata( + product_name: str, + dataset_name: str, + file_id: str, # refers to product md + path_to_file: Path, + layer: str | None, + org_md: Path | OrgMetadata | None, # Allow passing OrgMetadata for testing purposes + zip_subdir: str | None, +): + """Write product metadata to an Esri metadata XML embedded in a shapefile or geodatabase. + Generates a new XML with defaults and applies product-specific values. + + Args: + product_name (str): Name of product. e.g. "lion" + dataset_name (str): Name of dataset within a product. e.g. "pseudo-lots" + file_id (str): File identifier from within the org. metadata yaml. e.g. "mappluto_unclipped_gdb" + path_to_file (Path): For shapefiles, path to the parent directory or zip file containing + the shapefile. For geodatabases, path to the `.gdb` itself or a zip containing it + at the top level. + layer (str | None): Shapefile filename (required for shapefiles) or GDB feature class name + (optional for single-layer GDBs; inferred automatically if omitted). + org_md (Path | OrgMetadata | None): Metadata reference used to populate the embedded XML. + zip_subdir (str | None): Internal path if shp is nested within a zip file. + Must be None when path is a file geodatabase. + """ + if isinstance(org_md, Path) or not org_md: + org_md = product_metadata.load(org_md_path_override=org_md) + + product_md = org_md.product(product_name).dataset(dataset_name) + + is_gdb = ".gdb" in path_to_file.suffixes + is_shp = layer is not None and ( + ".shp" in path_to_file.suffixes or layer.endswith(".shp") + ) + + if is_gdb: + if zip_subdir is not None: + raise ValueError( + "Nested zipped GDBs are not supported. The GDB must be at the top level of the zip." + ) + layer = fgdb.resolve_layer(path_to_file, layer) + file_metadata = product_md.calculate_layer_dataset_metadata( + file_id=file_id, layer=layer + ) + custom_type_key = "fgdb_data_type" + elif is_shp: + file_metadata = product_md.calculate_file_dataset_metadata(file_id=file_id) + custom_type_key = "shp_data_type" + else: + raise ValueError( + f"Unsupported file type for metadata writing: path='{path_to_file}', layer='{layer}'. " + "Expected a .gdb or .shp path." + ) + + assert ( + layer is not None + ) # guaranteed: GDB branch resolved it, SHP branch required it + + logger.info(f"Wrote metadata to layer '{layer}' in {path_to_file}") + + esri_md = esri_metadata.generate_metadata() + + # Set dataset-level values + # TODO: define DCP organizationally required metadata fields + esri_md.md_hr_lv_name = "dataset" + esri_md.data_id_info.id_citation.res_title = file_metadata.attributes.display_name + esri_md.data_id_info.id_abs = file_metadata.attributes.description + # TODO: map idPurp to a product-metadata field + esri_md.data_id_info.id_credit = file_metadata.attributes.attribution + esri_md.data_id_info.res_const.consts.use_limit = ( + file_metadata.attributes.disclaimer + ) + esri_md.data_id_info.other_keys.keyword = file_metadata.attributes.tags + esri_md.data_id_info.search_keys.keyword = file_metadata.attributes.tags + + if file_metadata.attributes.projection: + authority, code = file_metadata.attributes.projection.split(":") + ref_sys_id = esri_md.ref_sys_info.ref_system.ref_sys_id + ref_sys_id.ident_code.code = int(code) + ref_sys_id.id_code_space.value = authority + # idVersion intentionally omitted: ArcGIS Synchronize Metadata overwrites it from its bundled EPSG dataset + + entity_name = layer.removesuffix(".shp") + esri_md.eainfo.detailed.enttyp.enttypl.value = entity_name + esri_md.eainfo.detailed.enttyp.enttypt.value = "Feature Class" + esri_md.eainfo.detailed.name = entity_name + + # Build attribute metadata for each column + esri_md.eainfo.detailed.attr = [ + _create_attr_metadata(column, custom_type_key=custom_type_key) + for column in file_metadata.columns + ] + + if is_gdb: + fgdb.write_metadata( + gdb=path_to_file, layer=layer, metadata=esri_md, overwrite=True + ) + else: + shp = Shapefile(path=path_to_file, shp_name=layer, zip_subdir=zip_subdir) + shp.write_metadata(esri_md, overwrite=True) + + diff --git a/dcpy/lifecycle/package/shapefiles.py b/dcpy/lifecycle/package/shapefiles.py index 44555833cb..eeb12a2db5 100644 --- a/dcpy/lifecycle/package/shapefiles.py +++ b/dcpy/lifecycle/package/shapefiles.py @@ -6,18 +6,13 @@ import typer -from dcpy.lifecycle import product_metadata -from dcpy.models.data.shapefile_metadata import Attr, Edom +from dcpy.models.dataset import COLUMN_TYPES from dcpy.models.product.dataset.metadata import ( - COLUMN_TYPES, ColumnValue, DatasetAttributes, DatasetColumn, Metadata, ) -from dcpy.models.product.metadata import OrgMetadata -from dcpy.utils.geospatial import shapefile as shp_utils -from dcpy.utils.geospatial.shapefile import Shapefile from dcpy.utils.logging import logger _shapefile_to_dcpy_types: dict[str, COLUMN_TYPES] = { @@ -121,8 +116,9 @@ def parse_shapefile_metadata(file_path: Path) -> Metadata: app = typer.Typer() +# TODO - delete? @app.command("to_metadata") -def _write_metadata( +def _write_shapefile_metadata( shp_xml_path: Path, output_path: Path = typer.Option( None, @@ -134,110 +130,3 @@ def _write_metadata( out_path = output_path or Path("./metadata.yml") parse_shapefile_metadata(shp_xml_path).write_to_yaml(out_path) logger.info(f"Wrote metadata to {out_path}") - - -@app.command("write_metadata") -def _write_shapefile_xml_metadata( - product_name: str, - dataset_name: str, - path: Path, - shp_name: str, - org_md_path: Path | None = typer.Option( - None, - "--org-md-path", - help="Path to organizational metadata", - ), - zip_subdir: str | None = typer.Option( - None, - "--zip-subdir", - help="Directory structure within zip file, if relevant", - ), -): - write_shapefile_xml_metadata( - product_name=product_name, - dataset_name=dataset_name, - path=path, - shp_name=shp_name, - zip_subdir=zip_subdir, - org_md=org_md_path, - ) - logger.info(f"Wrote metadata to {shp_name} in {path}") - - -def write_shapefile_xml_metadata( - product_name: str, - dataset_name: str, - path: Path, - shp_name: str, - zip_subdir: str | None, - org_md: Path | OrgMetadata | None, # Allow passing OrgMetadata for testing purposes -): - """Write product metadata to the shapefile metadata XML. Generates a new XML with defaults, - and applies additional product-specific values. - - Args: - product_name (str): Name of product. e.g. "lion" - dataset_name (str): Name of dataset within a product. e.g. "pseudo-lots" - path (Path): Path to parent directory or zip file containing shapefile. - shp_name (str): Shapefile name, ending in ".shp". e.g. "shapefile_name.shp" - zip_subdir (str | None): Internal path, if shp is nested within a zip file. - org_md (Path | OrgMetadata | None): Metadata reference used to populate shapefile metadata. - """ - if isinstance(org_md, Path) or not org_md: - org_md = product_metadata.load(org_md_path_override=org_md) - - product_md = org_md.product(product_name).dataset(dataset_name) - - metadata = shp_utils.generate_metadata() - - # Set dataset-level values - # TODO: define DCP organizationally required metadata fields - metadata.md_hr_lv_name = product_md.attributes.display_name - metadata.data_id_info.id_abs = product_md.attributes.description - metadata.data_id_info.other_keys.keyword = product_md.attributes.tags - metadata.data_id_info.search_keys.keyword = product_md.attributes.tags - - metadata.eainfo.detailed.name = product_md.id - metadata.eainfo.detailed.enttyp.enttypl.value = product_md.id - metadata.eainfo.detailed.enttyp.enttypt.value = "Feature Class" - - # Build attribute metadata for each column - metadata.eainfo.detailed.attr = [ - _create_attr_metadata(column) for column in product_md.columns - ] - - shp = Shapefile(path=path, shp_name=shp_name, zip_subdir=zip_subdir) - shp.write_metadata(metadata, overwrite=True) - - -def _create_attr_metadata(column: DatasetColumn) -> Attr: - """Create an Attr metadata object from a column specification.""" - attr = Attr() - - attr.attrlabl.value = "FID" if column.id == "uid" else column.id - attr.attalias.value = "FID" if column.name == "uid" else column.name - attr.attrdef.value = column.description - - # TODO: define column-level defaults (e.g. attrdefs = 'Esri' if column.name == 'uid') - # TODO: map DCP types to Esri types (e.g. attrtype = 'OID' if column.name == 'uid') Note DCP types != Esri types - # attr.attrtype.value = column.data_type - # attr.attwidth.value = None - # attr.atprecis.value = None - # attr.attscale.value = None - # attr.attrdefs.value = "" - - # Handle domain values if present - if hasattr(column, "values") and column.values: - attr.attrdomv.edom = [_create_edom_metadata(value) for value in column.values] - - # TODO: handle 'attrdomv.udom', with other esri value defaults - return attr - - -def _create_edom_metadata(column_value: ColumnValue) -> Edom: - """Create an Edom metadata object from a column value specification.""" - edom = Edom() - edom.edomv = column_value.value - edom.edomvd = column_value.description - - return edom diff --git a/dcpy/lifecycle/package/validate.py b/dcpy/lifecycle/package/validate.py index a6dbe11b67..e643c672cd 100644 --- a/dcpy/lifecycle/package/validate.py +++ b/dcpy/lifecycle/package/validate.py @@ -143,6 +143,16 @@ def _is_geom_poly(s): "lot": lambda df, col_name: df.iloc[0:0], "text": lambda df, col_name: df.iloc[0:0], "uid": lambda df, col_name: df.iloc[0:0], + # Esri type aliases: used when a file-entry override sets data_type to the + # column's actual Esri type rather than a DCP semantic type. + "String": lambda df, col_name: df.iloc[0:0], + "Integer": lambda df, col_name: df[~df[col_name].apply(_is_int)], + "SmallInteger": lambda df, col_name: df[~df[col_name].apply(_is_int)], + "Double": lambda df, col_name: df[~df[col_name].apply(_is_number)], + "Float": lambda df, col_name: df[~df[col_name].apply(_is_number)], + "Date": lambda df, col_name: df.iloc[0:0], + "Geometry": lambda df, col_name: df.iloc[0:0], + "OID": lambda df, col_name: df.iloc[0:0], } diff --git a/dcpy/models/product/dataset/metadata.py b/dcpy/models/product/dataset/metadata.py index 39fb422af8..aadd7e687e 100644 --- a/dcpy/models/product/dataset/metadata.py +++ b/dcpy/models/product/dataset/metadata.py @@ -7,7 +7,7 @@ from tabulate import tabulate # type: ignore from dcpy.models.base import SortedSerializedBase, TemplatedYamlReader, YamlWriter -from dcpy.models.dataset import COLUMN_TYPES, Column +from dcpy.models.dataset import Column from dcpy.utils.collections import deep_merge_dict as merge ERROR_MISSING_COLUMN = "MISSING COLUMN" @@ -81,6 +81,11 @@ class DatasetColumn(CustomizableBase, Column): # Note: id isn't intended to be overrideable, but is always required as a # pointer back to the original column. name: str | None = None + # Widened from base Column's COLUMN_TYPES Literal to str so the field can hold + # DCP semantic types and remain nullable in overrides. Format-specific Esri types + # (e.g. "String", "SmallInteger") belong in custom["fgdb_data_type"] or + # custom["shp_data_type"], not here. + data_type: str | None = None # type: ignore[assignment] data_source: str | None = None description: str | None = None limitations: str | None = None @@ -99,6 +104,14 @@ class FileOverrides(CustomizableBase): type: str | None = None +class GdbLayerOverrides(CustomizableBase): + """Per-layer column overrides for a geodatabase file. `layer` must match + an actual layer name in the GDB.""" + + layer: str + overridden_columns: list["DatasetColumn"] = [] + + class File(CustomizableBase): """Describes an actual dataset file, e.g. dataset files or attachments.""" @@ -108,6 +121,7 @@ class File(CustomizableBase): is_metadata: bool | None = ( None # e.g. readmes, data_dictionaries, version_files, etc. ) + layers: list[GdbLayerOverrides] | None = None def override(self, overrides: FileOverrides) -> File: return File( @@ -147,6 +161,7 @@ class DatasetOrgProductAttributesOverride(CustomizableBase): contains_address: bool | None = ( None # `contains_address` refers specifically to addresses containing house, numbers + street names. (ie. not just streets, polys, etc.) ) + current_version: str | None = None data_collection_method: str | None = None data_change_frequency: str | None = None date_made_public: str | None = None @@ -159,7 +174,7 @@ class DatasetOrgProductAttributesOverride(CustomizableBase): publishing_frequency_details: str | None = None publishing_purpose: str | None = None rows_removed: bool | None = None - tags: List[str] | None = [] + tags: List[str] | None = None class DatasetAttributesOverride(DatasetOrgProductAttributesOverride): @@ -322,6 +337,18 @@ def calculate_file_dataset_metadata(self, *, file_id: str) -> Dataset: self.get_file_and_overrides(file_id).dataset_overrides ) + def calculate_layer_dataset_metadata(self, *, file_id: str, layer: str) -> Dataset: + file_dataset = self.calculate_file_dataset_metadata(file_id=file_id) + file = self.get_file_and_overrides(file_id).file + if not file.layers: + return file_dataset + matching = [lo for lo in file.layers if lo.layer == layer] + if not matching: + return file_dataset + layer_overrides = matching[0] + overrides = DatasetOverrides(overridden_columns=layer_overrides.overridden_columns) + return file_dataset.override(overrides) + def calculate_destination_metadata( self, *, file_id: str, destination_id: str ) -> DestinationMetadata: @@ -408,7 +435,7 @@ def validate_consistency(self): return errors def apply_column_defaults( - self, column_defaults: dict[tuple[str, COLUMN_TYPES], DatasetColumn] + self, column_defaults: dict[tuple[str, str], DatasetColumn] ) -> list[DatasetColumn]: return [ c.override(column_defaults[c.id, c.data_type]) diff --git a/dcpy/models/product/metadata.py b/dcpy/models/product/metadata.py index 6a2b9a1921..c8e5b85704 100644 --- a/dcpy/models/product/metadata.py +++ b/dcpy/models/product/metadata.py @@ -11,7 +11,6 @@ from dcpy.models.product.artifacts import Artifact, Artifacts from dcpy.models.product.data_dictionary import DataDictionary from dcpy.models.product.dataset.metadata import ( - COLUMN_TYPES, DatasetColumn, DatasetOrgProductAttributesOverride, ) @@ -48,7 +47,7 @@ class ProductMetadata(SortedSerializedBase, extra="forbid"): root_path: Path metadata: ProductMetadataFile template_vars: dict = {} - column_defaults: dict[tuple[str, COLUMN_TYPES], DatasetColumn] = {} + column_defaults: dict[tuple[str, str], DatasetColumn] = {} org_attributes: DatasetOrgProductAttributesOverride @classmethod @@ -56,7 +55,7 @@ def from_path( cls, root_path: Path, template_vars: dict = {}, - column_defaults: dict[tuple[str, COLUMN_TYPES], DatasetColumn] = {}, + column_defaults: dict[tuple[str, str], DatasetColumn] = {}, org_attributes: DatasetOrgProductAttributesOverride = DatasetOrgProductAttributesOverride(), ) -> ProductMetadata: return ProductMetadata( @@ -187,7 +186,7 @@ class OrgMetadata(SortedSerializedBase, extra="forbid"): root_path: Path template_vars: dict = Field(default_factory=dict) metadata: OrgMetadataFile - column_defaults: dict[tuple[str, COLUMN_TYPES], DatasetColumn] + column_defaults: dict[tuple[str, str], DatasetColumn] data_dictionary: DataDictionary = DataDictionary() @classmethod @@ -205,7 +204,7 @@ def get_string_snippets(cls, path: Path) -> dict: @classmethod def get_column_defaults( cls, path: Path - ) -> dict[tuple[str, COLUMN_TYPES], DatasetColumn]: + ) -> dict[tuple[str, str], DatasetColumn]: c_path = path / "snippets" / "column_defaults.yml" if not c_path.exists(): return {} diff --git a/dcpy/test/lifecycle/package/test_metadata.py b/dcpy/test/lifecycle/package/test_metadata.py new file mode 100644 index 0000000000..6d45b4f294 --- /dev/null +++ b/dcpy/test/lifecycle/package/test_metadata.py @@ -0,0 +1,69 @@ +from dcpy.models.product.dataset.metadata import ( + DatasetAttributes, + DatasetColumn, + DatasetOverrides, + File, + FileAndOverrides, + GdbLayerOverrides, + Metadata, +) + + +def _make_metadata(layers: list[GdbLayerOverrides] | None = None) -> Metadata: + return Metadata( + id="test", + attributes=DatasetAttributes(display_name="Test", each_row_is_a="row"), + columns=[DatasetColumn(id="borough", name="Borough", data_type="text")], + files=[ + FileAndOverrides( + file=File( + id="my_gdb", + filename="my.gdb", + type="geodatabase", + layers=layers, + ), + dataset_overrides=DatasetOverrides( + overridden_columns=[DatasetColumn(id="borough", data_type="String")] + ), + ) + ], + ) + + +def test_calculate_layer_dataset_metadata_three_level_merge(): + """base → file-level → layer-level: all three levels are applied.""" + md = _make_metadata( + layers=[ + GdbLayerOverrides( + layer="my_layer", + overridden_columns=[DatasetColumn(id="borough", description="Layer desc")], + ) + ] + ) + result = md.calculate_layer_dataset_metadata(file_id="my_gdb", layer="my_layer") + col = result.columns[0] + assert col.data_type == "String" # from file-level override + assert col.description == "Layer desc" # from layer-level override + + +def test_calculate_layer_dataset_metadata_unmatched_layer_returns_file_level(): + """Layer name not in file.layers → layer-level overrides not applied.""" + md = _make_metadata( + layers=[ + GdbLayerOverrides( + layer="my_layer", + overridden_columns=[DatasetColumn(id="borough", description="Layer desc")], + ) + ] + ) + result = md.calculate_layer_dataset_metadata(file_id="my_gdb", layer="other_layer") + col = result.columns[0] + assert col.data_type == "String" # file-level override still applied + assert col.description is None # layer-level description not applied + + +def test_calculate_layer_dataset_metadata_no_layers_declared_returns_file_level(): + """file.layers is None → file-level result returned unchanged.""" + md = _make_metadata(layers=None) + result = md.calculate_layer_dataset_metadata(file_id="my_gdb", layer="any_layer") + assert result.columns[0].data_type == "String" diff --git a/dcpy/test/lifecycle/package/test_package_esri.py b/dcpy/test/lifecycle/package/test_package_esri.py new file mode 100644 index 0000000000..8feeef549a --- /dev/null +++ b/dcpy/test/lifecycle/package/test_package_esri.py @@ -0,0 +1,521 @@ +import shutil +import zipfile +from datetime import datetime +from pathlib import Path + +import pytest +from pytest import fixture + +from dcpy.lifecycle.package import esri +from dcpy.models.data.shapefile_metadata import Metadata +from dcpy.models.product.dataset.metadata import ColumnValue, DatasetColumn +from dcpy.models.product.metadata import OrgMetadata +from dcpy.utils.geospatial import fgdb +from dcpy.utils.geospatial import shapefile as shp_utils + +SHP_ZIP_NO_MD = "shapefile_single_pluto_feature_no_metadata.shp.zip" +SHP_ZIP_WITH_MD = "shapefile_single_pluto_feature_with_metadata.shp.zip" + +GDB_ZIP = "geodatabase.gdb.zip" +SPATIAL_LAYER = "mappluto_one_row" +SHP_SUBDIR = "subdir" + + +@fixture +def temp_shp_zip_no_md_path(utils_resources_path, tmp_path): + shutil.copy2( + src=utils_resources_path / SHP_ZIP_NO_MD, + dst=tmp_path / SHP_ZIP_NO_MD, + ) + assert zipfile.is_zipfile(tmp_path / SHP_ZIP_NO_MD), ( + f"'{SHP_ZIP_NO_MD}' should be a valid zip file" + ) + return tmp_path / SHP_ZIP_NO_MD + + +@fixture +def temp_shp_zip_with_subdir_path(utils_resources_path, tmp_path): + """Shapefile zip where the .shp files are nested inside a subdirectory.""" + extract_dir = tmp_path / "extracted" + extract_dir.mkdir() + shutil.unpack_archive( + filename=utils_resources_path / SHP_ZIP_NO_MD, extract_dir=extract_dir + ) + subdir_zip = tmp_path / SHP_ZIP_NO_MD + with zipfile.ZipFile(subdir_zip, "w") as zf: + for f in extract_dir.iterdir(): + zf.write(f, arcname=f"{SHP_SUBDIR}/{f.name}") + return subdir_zip + + +@fixture +def temp_shp_zip_with_md_path(utils_resources_path, tmp_path): + shutil.copy2( + src=utils_resources_path / SHP_ZIP_WITH_MD, + dst=tmp_path / SHP_ZIP_WITH_MD, + ) + assert zipfile.is_zipfile(tmp_path / SHP_ZIP_WITH_MD), ( + f"'{SHP_ZIP_WITH_MD}' should be a valid zip file" + ) + return tmp_path / SHP_ZIP_WITH_MD + + +@fixture +def temp_nonzipped_shp_no_md_path(temp_shp_zip_no_md_path, tmp_path): + shutil.unpack_archive(filename=temp_shp_zip_no_md_path, extract_dir=tmp_path) + shp_path = tmp_path / temp_shp_zip_no_md_path.stem + assert shp_path.is_file(), "Expected a shapefile, but found none" + assert not Path(f"{shp_path}.xml").is_file(), "Expected no file, but found one" + return shp_path + + +@fixture +def temp_nonzipped_shp_with_md_path(temp_shp_zip_with_md_path, tmp_path): + shutil.unpack_archive(filename=temp_shp_zip_with_md_path, extract_dir=tmp_path) + shp_path = tmp_path / temp_shp_zip_with_md_path.stem + assert shp_path.is_file(), "Expected a shapefile, but found none" + assert Path(f"{shp_path}.xml").is_file(), "Expected a file, but found none" + return shp_path + + +@fixture +def temp_gdb_zip_path(utils_resources_path, tmp_path): + shutil.copy2( + src=utils_resources_path / GDB_ZIP, + dst=tmp_path / GDB_ZIP, + ) + assert zipfile.is_zipfile(tmp_path / GDB_ZIP), ( + f"'{GDB_ZIP}' should be a valid zip file" + ) + return tmp_path / GDB_ZIP + + +@fixture +def temp_gdb_nonzipped_path(temp_gdb_zip_path, tmp_path): + shutil.unpack_archive(filename=temp_gdb_zip_path, extract_dir=tmp_path) + gdb_path = tmp_path / temp_gdb_zip_path.stem + assert gdb_path.is_dir(), "Expected a gdb directory, but found none" + return gdb_path + + +def _get_info_from_file_fixture( + request: pytest.FixtureRequest, fixture: str, file_type: str +) -> dict: + """Calculate path and layer name for a given fixture. + Calculation differs between zipped and non-zipped fixtures. + Supports shapefiles and file geodatabases. + + Args: + request (pytest.FixtureRequest): + fixture (str): fixture name + file_type (str): type of fixture - either "zip" or "nonzip" + + Returns: + dict: path and layer name for given fixture + """ + + if file_type not in ["zip", "nonzip"]: + raise Exception(f"Type: {file_type} is an ") + elif file_type == "zip": + path_fixture = request.getfixturevalue(fixture) # Retrieve fixture by name + if ".gdb" in path_fixture.suffixes: + layer = SPATIAL_LAYER + elif ".shp" in path_fixture.suffixes: + layer = path_fixture.stem + path = path_fixture + elif file_type == "nonzip": + path_fixture = request.getfixturevalue(fixture) + if ".gdb" in path_fixture.suffixes: + path = path_fixture # GDB directory is the addressable path itself + layer = SPATIAL_LAYER + elif ".shp" in path_fixture.suffixes: + path = path_fixture.parent + layer = path_fixture.name + return {"path": path, "layer": layer} + + +@fixture +def today_datestamp() -> str: + return datetime.now().strftime("%Y%m%d") + + +@pytest.fixture +def org_metadata(package_and_dist_test_resources): + return package_and_dist_test_resources.org_md + + +@pytest.mark.parametrize( + "path_fixture, file_type, subdir", + [ + pytest.param( + "temp_shp_zip_no_md_path", + "zip", + None, + id="add_md_to_zip_shp_w_no_md", + ), + pytest.param( + "temp_nonzipped_shp_no_md_path", + "nonzip", + None, + id="add_md_to_nonzip_shp_w_no_md", + ), + pytest.param( + "temp_shp_zip_with_md_path", + "zip", + None, + id="add_md_to_zip_shp_with_md", + ), + pytest.param( + "temp_nonzipped_shp_with_md_path", + "nonzip", + None, + id="add_md_to_nonzip_shp_with_md", + ), + pytest.param( + "temp_shp_zip_with_subdir_path", + "zip", + SHP_SUBDIR, + id="add_md_to_zip_shp_with_subdir", + ), + pytest.param( + "temp_gdb_zip_path", + "zip", + None, + id="add_md_to_zip_gdb", + ), + pytest.param( + "temp_gdb_nonzipped_path", + "nonzip", + None, + id="add_md_to_nonzip_gdb", + ), + ], +) +def test_write_metadata( + request, + path_fixture, + file_type, + subdir, + org_metadata: OrgMetadata, +): + fixture_info = _get_info_from_file_fixture( + request, fixture=path_fixture, file_type=file_type + ) + + product_md = org_metadata.product("colp").dataset("colp") + file_metadata = product_md.calculate_file_dataset_metadata( + file_id="primary_shapefile" + ) + + fields = Metadata.model_fields + + # write metadata + esri.write_metadata( + product_name="colp", + dataset_name="colp", + path_to_file=fixture_info["path"], + layer=fixture_info["layer"], + file_id="primary_shapefile", + zip_subdir=subdir, + org_md=org_metadata, + ) + + # read it back + metadata = None + + if ".gdb" in fixture_info["path"].suffixes: + metadata = fgdb.read_metadata( + gdb=fixture_info["path"], layer=fixture_info["layer"] + ) + if ".shp" in fixture_info["layer"]: + shp = shp_utils.from_path( + path=fixture_info["path"], shp_name=fixture_info["layer"], zip_subdir=subdir + ) + metadata = shp.read_metadata() + + if metadata is None: + pytest.fail("Expected metadata to exist") + + # Test default values + assert metadata.md_stan_name == fields["md_stan_name"].default + assert metadata.md_stan_ver == fields["md_stan_ver"].default + # TODO - add helper code to access nested defaults (if this is the direction we end up pursuing) + + # Test product-specific values + assert metadata.md_hr_lv_name == "dataset" + assert ( + metadata.data_id_info.id_citation.res_title + == file_metadata.attributes.display_name + ) + assert metadata.data_id_info.id_abs == file_metadata.attributes.description + assert metadata.data_id_info.id_credit == file_metadata.attributes.attribution + assert ( + metadata.data_id_info.res_const.consts.use_limit + == file_metadata.attributes.disclaimer + ) + assert metadata.data_id_info.other_keys.keyword == file_metadata.attributes.tags + assert metadata.data_id_info.search_keys.keyword == file_metadata.attributes.tags + + assert metadata.eainfo.detailed.name == fixture_info["layer"].removesuffix(".shp") + assert metadata.eainfo.detailed.enttyp.enttypl.value == fixture_info[ + "layer" + ].removesuffix(".shp") + assert metadata.eainfo.detailed.enttyp.enttypt.value == "Feature Class" + + # column 0 has no domain values: attrlabl, attrtype, and udom should all round-trip + col0 = file_metadata.columns[0] + expected_label_0 = col0.name + assert metadata.eainfo.detailed.attr[0].attrlabl.value == expected_label_0 + expected_type_0 = "OID" if col0.id == "uid" else col0.data_type + assert metadata.eainfo.detailed.attr[0].attrtype.value == expected_type_0 + udom_0 = metadata.eainfo.detailed.attr[0].attrdomv.udom + assert udom_0 is not None + assert udom_0.value == col0.description + + assert file_metadata.columns[1].values is not None, "Column values must be defined" + + # column 1 is borough in the colp test fixture — has domain values + assert ( + metadata.eainfo.detailed.attr[1].attrlabl.value == file_metadata.columns[1].name + ) + assert ( + metadata.eainfo.detailed.attr[1].attrtype.value + == file_metadata.columns[1].data_type + ) + udom_1 = metadata.eainfo.detailed.attr[1].attrdomv.udom + assert udom_1 is None or udom_1.value is None + assert ( + metadata.eainfo.detailed.attr[1].attrdomv.edom[0].edomv + == file_metadata.columns[1].values[0].value # "1", when org_md product is colp + ) + assert ( + metadata.eainfo.detailed.attr[1].attrdomv.edom[0].edomvd + == file_metadata.columns[1] + .values[0] + .description # "Manhattan", when org_md product is colp + ) + + +def _make_column(**kwargs) -> DatasetColumn: + defaults = dict( + id="some_field", + name="Some Field", + data_type="text", + description="A description", + ) + return DatasetColumn(**(defaults | kwargs)) + + +def test_create_attr_metadata_basic(): + col = _make_column(name="MyField", description="My desc", data_source="Agency X") + attr = esri._create_attr_metadata(col) + assert attr.attrlabl.value == "MyField" + assert attr.attalias.value == "MyField" + assert attr.attrdef.value == "My desc" + assert attr.attrdefs.value == "Agency X" + + +def test_create_attr_metadata_uid_label_passthrough(): + """uid's label is no longer hardcoded to "FID" — it's driven by column.name, + set via per-format name overrides in product-metadata (FID for SHP, OBJECTID + for GDB).""" + col = _make_column(id="uid", name="FID") + attr = esri._create_attr_metadata(col) + assert attr.attrlabl.value == "FID" + assert attr.attalias.value == "FID" + + col = _make_column(id="uid", name="OBJECTID") + attr = esri._create_attr_metadata(col) + assert attr.attrlabl.value == "OBJECTID" + assert attr.attalias.value == "OBJECTID" + + +def test_create_attr_metadata_no_data_source(): + col = _make_column() + attr = esri._create_attr_metadata(col) + assert attr.attrdefs.value is None + + +def test_create_attr_metadata_with_values(): + col = _make_column( + values=[ + ColumnValue(value="A", description="Alpha"), + ColumnValue(value="B", description="Beta"), + ] + ) + attr = esri._create_attr_metadata(col) + assert len(attr.attrdomv.edom) == 2 + assert attr.attrdomv.edom[0].edomv == "A" + assert attr.attrdomv.edom[0].edomvd == "Alpha" + assert attr.attrdomv.edom[1].edomv == "B" + assert attr.attrdomv.edom[1].edomvd == "Beta" + + +def test_create_attr_metadata_no_values(): + col = _make_column() + attr = esri._create_attr_metadata(col) + assert attr.attrdomv.edom == [] + + +def test_create_attr_metadata_esri_type_passthrough(): + col = _make_column(data_type="String") + assert esri._create_attr_metadata(col).attrtype.value == "String" + + col = _make_column(data_type="SmallInteger") + assert esri._create_attr_metadata(col).attrtype.value == "SmallInteger" + + col = _make_column(data_type=None) + assert esri._create_attr_metadata(col).attrtype.value is None + + +def test_create_attr_metadata_uid_attrtype_is_oid(): + col = _make_column(id="uid", name="uid", data_type="String") + assert esri._create_attr_metadata(col).attrtype.value == "OID" + + +def test_create_attr_metadata_udom_set_when_no_values(): + col = _make_column(description="Free-form text field") + attr = esri._create_attr_metadata(col) + assert attr.attrdomv.udom is not None + assert attr.attrdomv.udom.value == "Free-form text field" + + +def test_create_attr_metadata_udom_none_when_values_present(): + col = _make_column(values=[ColumnValue(value="A", description="Alpha")]) + attr = esri._create_attr_metadata(col) + assert attr.attrdomv.udom is None + + +def test_create_attr_metadata_uses_custom_type_key(): + col = _make_column(data_type="text", custom={"fgdb_data_type": "String"}) + attr = esri._create_attr_metadata(col, custom_type_key="fgdb_data_type") + assert attr.attrtype.value == "String" + + +def test_create_attr_metadata_falls_back_to_data_type_when_custom_key_absent(): + col = _make_column(data_type="text") + attr = esri._create_attr_metadata(col, custom_type_key="fgdb_data_type") + assert attr.attrtype.value == "text" + + +def test_create_attr_metadata_no_custom_key_ignores_custom_dict(): + col = _make_column(data_type="text", custom={"fgdb_data_type": "String"}) + attr = esri._create_attr_metadata(col) # no custom_type_key + assert attr.attrtype.value == "text" + + +def test_write_metadata_gdb_pluto(temp_gdb_zip_path, org_metadata): + """Verifies GDB-specific metadata writing using the pluto test fixture. + + Checks: + - eainfo.detailed.name is product_md.id ("pluto"), not the layer name + - data_type overrides (e.g. "String", "Date") pass through verbatim to attrtype + - full column names are used (no shapefile truncation) + """ + esri.write_metadata( + product_name="pluto", + dataset_name="pluto", + path_to_file=temp_gdb_zip_path, + layer=SPATIAL_LAYER, + file_id="primary_file_geodatabase", + zip_subdir=None, + org_md=org_metadata, + ) + metadata = fgdb.read_metadata(gdb=temp_gdb_zip_path, layer=SPATIAL_LAYER) + if metadata is None: + pytest.fail("Expected metadata to exist after write") + + pluto_md = org_metadata.product("pluto").dataset("pluto") + file_metadata = pluto_md.calculate_file_dataset_metadata( + file_id="primary_file_geodatabase" + ) + + assert metadata.eainfo.detailed.name == SPATIAL_LAYER + assert metadata.eainfo.detailed.enttyp.enttypl.value == SPATIAL_LAYER + + # uid → GDB-specific name override (OBJECTID), attrtype always OID + assert metadata.eainfo.detailed.attr[0].attrlabl.value == "OBJECTID" + assert metadata.eainfo.detailed.attr[0].attrtype.value == "OID" + + # borough — full name, has domain values (edom), no truncation + assert ( + metadata.eainfo.detailed.attr[1].attrlabl.value == file_metadata.columns[1].name + ) + assert metadata.eainfo.detailed.attr[1].attrtype.value == "String" + assert len(metadata.eainfo.detailed.attr[1].attrdomv.edom) == len( + file_metadata.columns[1].values + ) + + # appdate — file-entry override declares Esri type "Date" explicitly + assert metadata.eainfo.detailed.attr[2].attrlabl.value == "APPDate" + assert metadata.eainfo.detailed.attr[2].attrtype.value == "Date" + + # projection — EPSG:2263 maps to refSysInfo + ref_sys_id = metadata.ref_sys_info.ref_system.ref_sys_id + assert ref_sys_id.id_code_space.value == "EPSG" + assert ref_sys_id.ident_code.code == 2263 + + +def test_write_metadata_gdb_layer_none_auto_resolves( + temp_gdb_zip_path, org_metadata, monkeypatch +): + """layer=None auto-resolves when GDB has exactly one layer.""" + monkeypatch.setattr(fgdb, "get_layers", lambda _: [SPATIAL_LAYER]) + esri.write_metadata( + product_name="pluto", + dataset_name="pluto", + path_to_file=temp_gdb_zip_path, + layer=None, + file_id="primary_file_geodatabase", + zip_subdir=None, + org_md=org_metadata, + ) + metadata = fgdb.read_metadata(gdb=temp_gdb_zip_path, layer=SPATIAL_LAYER) + assert metadata is not None + assert metadata.md_hr_lv_name == "dataset" + + +def test_write_metadata_gdb_layer_none_raises_when_ambiguous( + temp_gdb_zip_path, org_metadata +): + """layer=None raises ValueError when GDB has multiple layers.""" + with pytest.raises(ValueError, match="layer must be specified"): + esri.write_metadata( + product_name="pluto", + dataset_name="pluto", + path_to_file=temp_gdb_zip_path, + layer=None, + file_id="primary_file_geodatabase", + zip_subdir=None, + org_md=org_metadata, + ) + + +def test_write_metadata_raises_on_nested_gdb_zip(tmp_path, org_metadata): + gdb_path = tmp_path / "test.gdb" + gdb_path.mkdir() + with pytest.raises(ValueError, match="Nested zipped GDBs are not supported"): + esri.write_metadata( + product_name="colp", + dataset_name="colp", + path_to_file=gdb_path, + layer="some_layer", + file_id="primary_shapefile", + zip_subdir="some_subdir", + org_md=org_metadata, + ) + + +def test_write_metadata_raises_on_unsupported_file_type(tmp_path, org_metadata): + bad_path = tmp_path / "file.csv" + bad_path.touch() + with pytest.raises(ValueError, match="Unsupported file type"): + esri.write_metadata( + product_name="colp", + dataset_name="colp", + path_to_file=bad_path, + layer="some_layer", + file_id="primary_shapefile", + zip_subdir=None, + org_md=org_metadata, + ) diff --git a/dcpy/test/lifecycle/package/test_shapefiles.py b/dcpy/test/lifecycle/package/test_shapefiles.py deleted file mode 100644 index 4118831b31..0000000000 --- a/dcpy/test/lifecycle/package/test_shapefiles.py +++ /dev/null @@ -1,174 +0,0 @@ -import shutil -import zipfile -from datetime import datetime -from pathlib import Path - -import pytest -from pytest import fixture - -from dcpy.lifecycle.package import shapefiles -from dcpy.models.data.shapefile_metadata import Metadata -from dcpy.models.product.metadata import OrgMetadata -from dcpy.utils.geospatial import shapefile as shp_utils - -SHP_ZIP_NO_MD = "shapefile_single_pluto_feature_no_metadata.shp.zip" -SHP_ZIP_WITH_MD = "shapefile_single_pluto_feature_with_metadata.shp.zip" - - -@fixture -def temp_shp_zip_no_md_path(utils_resources_path, tmp_path): - shutil.copy2( - src=utils_resources_path / SHP_ZIP_NO_MD, - dst=tmp_path / SHP_ZIP_NO_MD, - ) - assert zipfile.is_zipfile(tmp_path / SHP_ZIP_NO_MD), ( - f"'{SHP_ZIP_NO_MD}' should be a valid zip file" - ) - return tmp_path / SHP_ZIP_NO_MD - - -@fixture -def temp_shp_zip_with_md_path(utils_resources_path, tmp_path): - shutil.copy2( - src=utils_resources_path / SHP_ZIP_WITH_MD, - dst=tmp_path / SHP_ZIP_WITH_MD, - ) - assert zipfile.is_zipfile(tmp_path / SHP_ZIP_WITH_MD), ( - f"'{SHP_ZIP_WITH_MD}' should be a valid zip file" - ) - return tmp_path / SHP_ZIP_WITH_MD - - -@fixture -def temp_nonzipped_shp_no_md_path(temp_shp_zip_no_md_path, tmp_path): - shutil.unpack_archive(filename=temp_shp_zip_no_md_path, extract_dir=tmp_path) - shp_path = tmp_path / temp_shp_zip_no_md_path.stem - assert shp_path.is_file(), "Expected a shapefile, but found none" - assert not Path(f"{shp_path}.xml").is_file(), "Expected no file, but found one" - return shp_path - - -@fixture -def temp_nonzipped_shp_with_md_path(temp_shp_zip_with_md_path, tmp_path): - shutil.unpack_archive(filename=temp_shp_zip_with_md_path, extract_dir=tmp_path) - shp_path = tmp_path / temp_shp_zip_with_md_path.stem - assert shp_path.is_file(), "Expected a shapefile, but found none" - assert Path(f"{shp_path}.xml").is_file(), "Expected a file, but found none" - return shp_path - - -def _get_info_from_file_fixture( - request: pytest.FixtureRequest, fixture: str, file_type: str -) -> dict: - """Calculate path and shp name for a given shapefile fixture. - Calculation differs between zipped and non-zipped fixtures. - - Args: - request (pytest.FixtureRequest): - fixture (str): fixture name - file_type (str): type of fixture - either "zip" or "nonzip" - - Returns: - dict: path and shapefile name for given fixture - """ - if file_type not in ["zip", "nonzip"]: - raise Exception(f"Type: {file_type} is an ") - elif file_type == "zip": - path = request.getfixturevalue(fixture) # Retrieve fixture by name - shp_name = path.stem - elif file_type == "nonzip": - path_fixture = request.getfixturevalue(fixture) - path = path_fixture.parent # Retrieve fixture by name - shp_name = path_fixture.name - return {"path": path, "shp_name": shp_name} - - -@fixture -def today_datestamp() -> str: - return datetime.now().strftime("%Y%m%d") - - -@pytest.fixture -def org_metadata(package_and_dist_test_resources): - return package_and_dist_test_resources.org_md - - -@pytest.mark.parametrize( - "path_fixture, file_type, subdir", - [ - pytest.param( - "temp_shp_zip_no_md_path", - "zip", - None, - id="add_md_to_zip_shp_w_no_md", - ), - pytest.param( - "temp_nonzipped_shp_no_md_path", - "nonzip", - None, - id="add_md_to_nonzip_shp_w_no_md", - ), - ], -) -def test_write_shapefile_xml_metadata( - request, - path_fixture, - file_type, - subdir, - org_metadata: OrgMetadata, -): - fixture_info = _get_info_from_file_fixture( - request, fixture=path_fixture, file_type=file_type - ) - - product_md = org_metadata.product("colp").dataset("colp") - - fields = Metadata.model_fields - - # write metadata - shapefiles.write_shapefile_xml_metadata( - product_name="colp", - dataset_name="colp", - path=fixture_info["path"], - shp_name=fixture_info["shp_name"], - zip_subdir=subdir, - org_md=org_metadata, - ) - - # read it back - shp = shp_utils.from_path( - path=fixture_info["path"], shp_name=fixture_info["shp_name"], zip_subdir=subdir - ) - metadata = shp.read_metadata() - - if metadata is None: - pytest.fail("Expected metadata to exist") - - # Test default values - assert metadata.md_stan_name == fields["md_stan_name"].default - assert metadata.md_stan_ver == fields["md_stan_ver"].default - # TODO - add helper code to access nested defaults (if this is the direction we end up pursuing) - - # Test product-specific values - assert metadata.md_hr_lv_name == product_md.attributes.display_name - assert metadata.data_id_info.id_abs == product_md.attributes.description - assert metadata.data_id_info.other_keys.keyword == product_md.attributes.tags - assert metadata.data_id_info.search_keys.keyword == product_md.attributes.tags - - assert metadata.eainfo.detailed.name == product_md.id - assert metadata.eainfo.detailed.enttyp.enttypl.value == product_md.id - assert metadata.eainfo.detailed.enttyp.enttypt.value == "Feature Class" - - assert product_md.columns[1].values is not None, "Column values must be defined" - - assert ( - metadata.eainfo.detailed.attr[1].attrdomv.edom[0].edomv - == product_md.columns[1].values[0].value # "1", when org_md product is colp - ) - - assert ( - metadata.eainfo.detailed.attr[1].attrdomv.edom[0].edomvd - == product_md.columns[1] - .values[0] - .description # "Manhattan", when org_md product is colp - ) diff --git a/dcpy/test/resources/package_and_distribute/metadata_repo/metadata.yml b/dcpy/test/resources/package_and_distribute/metadata_repo/metadata.yml index 672bd11afc..2225b91d79 100644 --- a/dcpy/test/resources/package_and_distribute/metadata_repo/metadata.yml +++ b/dcpy/test/resources/package_and_distribute/metadata_repo/metadata.yml @@ -4,8 +4,10 @@ attributes: attribution: DCP attribution_link: https://www.nyc.gov/site/planning/data-maps/open-data.page contact_email: opendata@planning.nyc.gov + projection: EPSG:2263 products: - colp - lion + - pluto - transit_zones diff --git a/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/metadata.yml b/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/metadata.yml new file mode 100644 index 0000000000..704e5ede11 --- /dev/null +++ b/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/metadata.yml @@ -0,0 +1,4 @@ +id: pluto + +datasets: +- pluto diff --git a/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/pluto/metadata.yml b/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/pluto/metadata.yml new file mode 100644 index 0000000000..1b9e6b1587 --- /dev/null +++ b/dcpy/test/resources/package_and_distribute/metadata_repo/products/pluto/pluto/metadata.yml @@ -0,0 +1,45 @@ +id: pluto + +attributes: + display_name: Primary Land Use Tax Lot Output (PLUTO) + description: PLUTO test dataset + each_row_is_a: Tax Lot + +files: + - file: + id: primary_file_geodatabase + filename: mappluto_wi_gdb.zip + type: geodatabase + dataset_overrides: + overridden_columns: + - id: uid + name: OBJECTID + +columns: + - id: uid + name: uid + data_type: text + description: Unique identifier + - id: borough + name: Borough + data_type: text + description: NYC borough + values: + - value: MN + description: Manhattan + - value: BX + description: Bronx + - value: BK + description: Brooklyn + - value: QN + description: Queens + - value: SI + description: Staten Island + custom: + fgdb_data_type: String + - id: appdate + name: APPDate + data_type: date + description: Date of most recent alteration permit application + custom: + fgdb_data_type: Date diff --git a/dcpy/test/utils/geospatial/test_fgdb.py b/dcpy/test/utils/geospatial/test_fgdb.py index 8cb2d04106..89a8cdefb6 100644 --- a/dcpy/test/utils/geospatial/test_fgdb.py +++ b/dcpy/test/utils/geospatial/test_fgdb.py @@ -8,8 +8,8 @@ from dcpy.utils.geospatial import fgdb GDB_ZIP = "geodatabase.gdb.zip" -FEATURE_CLASS = "mappluto_one_row" -TABLE = "pluto_one_row" +SPATIAL_LAYER = "mappluto_one_row" +TABLE_LAYER = "pluto_one_row" METADATA_XML = "esri_metadata.xml" @@ -68,12 +68,34 @@ def path_fixture(request): @gdb_paths def test_get_layers(path_fixture): layers = fgdb.get_layers(path_fixture) - assert layers == [FEATURE_CLASS, TABLE] + assert layers == [SPATIAL_LAYER, TABLE_LAYER] + + +@gdb_paths +def test_resolve_layer_explicit_valid(path_fixture): + assert fgdb.resolve_layer(path_fixture, SPATIAL_LAYER) == SPATIAL_LAYER + + +@gdb_paths +def test_resolve_layer_explicit_invalid(path_fixture): + with pytest.raises(LookupError, match="nonexistent_layer"): + fgdb.resolve_layer(path_fixture, "nonexistent_layer") + + +@gdb_paths +def test_resolve_layer_omitted_ambiguous(path_fixture): + with pytest.raises(ValueError, match="layer must be specified"): + fgdb.resolve_layer(path_fixture) + + +def test_resolve_layer_omitted_unambiguous(temp_gdb_zip_path, monkeypatch): + monkeypatch.setattr(fgdb, "get_layers", lambda _: [SPATIAL_LAYER]) + assert fgdb.resolve_layer(temp_gdb_zip_path) == SPATIAL_LAYER @gdb_paths def test_read_metadata(path_fixture): - md = fgdb.read_metadata(gdb=path_fixture, layer=FEATURE_CLASS) + md = fgdb.read_metadata(gdb=path_fixture, layer=SPATIAL_LAYER) element = "esri" assert hasattr(md, element), f"Expected element '{element}', but found none" @@ -87,13 +109,13 @@ def test_write_metadata(path_fixture, temp_metadata_object): layers_before_md_write = fgdb.get_layers(path_fixture) fgdb.write_metadata( gdb=path_fixture, - layer=FEATURE_CLASS, + layer=SPATIAL_LAYER, metadata=temp_metadata_object, overwrite=True, ) layers_after_md_write = fgdb.get_layers(path_fixture) - md = fgdb.read_metadata(path_fixture, FEATURE_CLASS) + md = fgdb.read_metadata(path_fixture, SPATIAL_LAYER) element = "esri" assert hasattr(md, element), f"Expected element '{element}', but found none" @@ -105,14 +127,14 @@ def test_write_metadata(path_fixture, temp_metadata_object): @gdb_paths def test_metadata_exists(path_fixture): - originally_md_exists = fgdb.metadata_exists(gdb=path_fixture, layer=FEATURE_CLASS) + originally_md_exists = fgdb.metadata_exists(gdb=path_fixture, layer=SPATIAL_LAYER) # remove metadata fgdb.remove_metadata( gdb=path_fixture, - layer=FEATURE_CLASS, + layer=SPATIAL_LAYER, ) md_exists_after_removal = fgdb.metadata_exists( - gdb=path_fixture, layer=FEATURE_CLASS + gdb=path_fixture, layer=SPATIAL_LAYER ) assert originally_md_exists is True, "Expected layer metadata but found none" assert md_exists_after_removal is False, ( @@ -125,11 +147,11 @@ def test_remove_metadata(path_fixture): layers_before_md_removal = fgdb.get_layers(path_fixture) fgdb.remove_metadata( gdb=path_fixture, - layer=FEATURE_CLASS, + layer=SPATIAL_LAYER, ) layers_after_md_removal = fgdb.get_layers(path_fixture) - md = fgdb.read_metadata(path_fixture, FEATURE_CLASS) + md = fgdb.read_metadata(path_fixture, SPATIAL_LAYER) assert md is None # confirm that no gdb layers were lost during md removal assert sorted(layers_before_md_removal) == sorted(layers_after_md_removal) diff --git a/dcpy/utils/geospatial/esri_metadata.py b/dcpy/utils/geospatial/esri_metadata.py index 5801f85db9..c4637e3f66 100644 --- a/dcpy/utils/geospatial/esri_metadata.py +++ b/dcpy/utils/geospatial/esri_metadata.py @@ -1,11 +1,15 @@ from datetime import datetime from dcpy.models.data.shapefile_metadata import ( + Attr, + Edom, Esri, Mddatest, Metadata, Scalerange, + Udom, ) +from dcpy.models.product.dataset.metadata import ColumnValue, DatasetColumn def generate_metadata() -> Metadata: @@ -53,3 +57,37 @@ def _get_esri_timestamp(dt_obj=None): ) return crea_date, crea_time + + +def _create_attr_metadata( + column: DatasetColumn, custom_type_key: str | None = None +) -> Attr: + """Create an Attr metadata object from a column specification.""" + attr = Attr() + + is_uid = column.id == "uid" + attr.attrlabl.value = column.name + attr.attalias.value = column.name + attr.attrdef.value = column.description + attr.attrdefs.value = column.data_source + effective_type = ( + column.custom.get(custom_type_key) if custom_type_key else None + ) or column.data_type + attr.attrtype.value = "OID" if is_uid else effective_type + + if column.values: + attr.attrdomv.udom = None + attr.attrdomv.edom = [_create_edom_metadata(value) for value in column.values] + else: + attr.attrdomv.udom = Udom(value=column.description) + attr.attrdomv.edom = [] + return attr + + +def _create_edom_metadata(column_value: ColumnValue) -> Edom: + """Create an Edom metadata object from a column value specification.""" + edom = Edom() + edom.edomv = column_value.value + edom.edomvd = column_value.description + + return edom diff --git a/dcpy/utils/geospatial/fgdb.py b/dcpy/utils/geospatial/fgdb.py index 6905c7c503..fff28e5693 100644 --- a/dcpy/utils/geospatial/fgdb.py +++ b/dcpy/utils/geospatial/fgdb.py @@ -17,6 +17,21 @@ def get_layers(gdb: Path) -> list[str]: return info["rootGroup"]["layerNames"] +def resolve_layer(gdb: Path, layer: str | None = None) -> str: + layers = get_layers(gdb) + if layer is not None: + if layer not in layers: + raise LookupError( + f"Layer '{layer}' not found in {gdb}. Found layers: {layers}." + ) + return layer + if len(layers) != 1: + raise ValueError( + f"{gdb} has {len(layers)} layers ({layers}); layer must be specified to disambiguate." + ) + return layers[0] + + def read_metadata(gdb: Path, layer: str, as_string: bool = False) -> Metadata | None: with gdal.ExceptionMgr(): layer_info = gdal.alg.vector.info(