Glue code for fgdb metadata

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)
+
+

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

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],
 }