add support for import/transform schema autodetection (#198) (#262)

* add support for import/transform schema autodetection (#198)

* fix PR comments

Author: tomkralidis

README.md

@@ -64,8 +64,14 @@ pygeometa metadata validate path/to/file.yml
 # import a metadata document to MCF
 pygeometa metadata import path/to/file.xml --schema=iso19139
 
+# import a metadata document to MCF, autodetecting the metadata file format
+pygeometa metadata import path/to/file.xml --schema=autodetect  # --schema=autodetect is default
+
 # transform from one metadata representation to another
 pygeometa metadata transform path/to/file.xml --input-schema=iso19139 --output-schema=oarec-record
+
+# transform from one metadata representation to another, autodetecting the metadata file format
+pygeometa metadata transform path/to/file.xml --input-schema=autodetect --output-schema=oarec-record  # --input-schema=autodetect is default
 ```
 
 ### Supported schemas

docs/content/tutorial.md

@@ -62,8 +62,14 @@ pygeometa validate path/to/file.yml
 # import a metadata document to MCF
 pygeometa metadata import path/to/file.xml --schema=iso19139
 
+# import a metadata document to MCF, autodetecting the metadata file format
+pygeometa metadata import path/to/file.xml --schema=autodetect  # --schema=autodetect is default
+
 # transform from one metadata representation to another
 pygeometa metadata transform path/to/file.xml --input-schema=iso19139 --output-schema=oarec-record
+
+# transform from one metadata representation to another, autodetecting the metadata file format
+pygeometa metadata transform path/to/file.xml --input-schema=autodetect --output-schema=oarec-record  # --input-schema=autodetect is default
 ```
 
 ## For Developers

pygeometa/core.py

@@ -324,6 +324,58 @@ def __parse_mcf_dict_recursive(dict2):
     return mcf_dict
 
 
+def import_metadata(schema: str, metadata: str) -> dict:
+    """
+    Import metadata
+
+    :param schema: schema / format
+    :metadata: metadata string
+
+    :returns: MCF object
+    """
+
+    if schema == 'autodetect':
+        schemas = get_supported_schemas()
+    else:
+        schemas = [schema]
+
+    for s in schemas:
+        LOGGER.debug(f'Attempting to import into {s}')
+        schema_object = load_schema(s)
+
+        try:
+            return schema_object.import_(metadata)
+        except NotImplementedError:
+            raise RuntimeError(f'Import not supported for {s}')
+        except Exception as err:
+            raise RuntimeError(f'Import failed: {err}')
+
+
+def transform_metadata(input_schema: str, output_schema: str,
+                       metadata: str) -> str:
+    """
+    Transform metadata
+
+    :param input_schema: input schema / format
+    :param output_schema: output schema / format
+    :metadata: metadata string
+
+    :returns: transformed metadata or `None`
+    """
+
+    try:
+        content = import_metadata(input_schema, metadata)
+
+        LOGGER.info(f'Processing into {output_schema}')
+        schema_object_output = load_schema(output_schema)
+        content = schema_object_output.write(content)
+    except Exception as err:
+        LOGGER.debug(err)
+        return None
+
+    return content
+
+
 def pretty_print(xml: str) -> str:
     """
     clean up indentation and spacing
@@ -479,23 +531,20 @@ class MCFValidationError(Exception):
 @cli_options.OPTION_OUTPUT
 @cli_options.OPTION_VERBOSITY
 @click.option('--schema', required=True,
-              type=click.Choice(get_supported_schemas()),
+              type=click.Choice(get_supported_schemas(include_autodetect=True)),  # noqa
+              default='autodetect',
               help='Metadata schema')
 def import_(ctx, metadata_file, schema, output, verbosity):
     """import metadata"""
 
-    LOGGER.info(f'Importing {metadata_file} into {schema}')
-    schema_object = load_schema(schema)
-
     try:
-        content = schema_object.import_(metadata_file.read())
-    except NotImplementedError:
-        raise click.ClickException(f'Import not supported for {schema}')
-
-    if output is None:
-        click.echo(yaml.dump(content))
-    else:
-        output.write(yaml.dump(content, indent=4))
+        content = import_metadata(schema, metadata_file.read())
+        if output is None:
+            click.echo(yaml.dump(content))
+        else:
+            output.write(yaml.dump(content, indent=4))
+    except Exception as err:
+        raise click.ClickException(f'No supported schema detecte/found: {err}')
 
 
 @click.command()
@@ -585,7 +634,8 @@ def validate(ctx, mcf, verbosity):
 @cli_options.OPTION_OUTPUT
 @cli_options.OPTION_VERBOSITY
 @click.option('--input-schema', required=True,
-              type=click.Choice(get_supported_schemas()),
+              type=click.Choice(get_supported_schemas(include_autodetect=True)),  # noqa
+              default='autodetect',
               help='Metadata schema of input file')
 @click.option('--output-schema', required=True,
               type=click.Choice(get_supported_schemas()),
@@ -594,18 +644,11 @@ def transform(ctx, metadata_file, input_schema, output_schema, output,
               verbosity):
     """transform metadata"""
 
-    LOGGER.info(f'Importing {metadata_file} into {input_schema}')
-    schema_object_input = load_schema(input_schema)
-    content = None
-
-    try:
-        content = schema_object_input.import_(metadata_file.read())
-    except NotImplementedError:
-        raise click.ClickException(f'Import not supported for {input_schema}')
+    content = transform_metadata(input_schema, output_schema,
+                                 metadata_file.read())
 
-    LOGGER.info(f'Processing into {output_schema}')
-    schema_object_output = load_schema(output_schema)
-    content = schema_object_output.write(content)
+    if content is None:
+        raise click.ClickException('No supported input schema detected/found')
 
     if output is None:
         click.echo(content)

pygeometa/schemas/__init__.py

@@ -65,11 +65,13 @@
 }
 
 
-def get_supported_schemas(details: bool = False) -> list:
+def get_supported_schemas(details: bool = False,
+                          include_autodetect: bool = False) -> list:
     """
     Get supported schemas
 
     :param details: provide read/write details
+    :param include_autodetect: include magic auto detection mode
 
     :returns: list of supported schemas
     """
@@ -91,7 +93,12 @@ def has_mode(plugin: BaseOutputSchema, mode: str) -> bool:
     LOGGER.debug('Generating list of supported schemas')
 
     if not details:
-        return SCHEMAS.keys()
+        if include_autodetect:
+            schemas_keys = list(SCHEMAS.keys())
+            schemas_keys.append('autodetect')
+            return schemas_keys
+        else:
+            return SCHEMAS.keys()
 
     for key in SCHEMAS.keys():
         schema = load_schema(key)
@@ -105,6 +112,14 @@ def has_mode(plugin: BaseOutputSchema, mode: str) -> bool:
             'write': can_write
         })
 
+    if include_autodetect:
+        schema_matrix.append({
+            'id': 'autodetect',
+            'description': 'Auto schema detection',
+            'read': True,
+            'write': False
+        })
+
     return schema_matrix
 
 

tests/run_tests.py

@@ -20,7 +20,7 @@
 #
 # Copyright (c) 2015 Government of Canada
 # Copyright (c) 2016 ERT Inc.
-# Copyright (c) 2024 Tom Kralidis
+# Copyright (c) 2025 Tom Kralidis
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -54,10 +54,11 @@
 import yaml
 
 from pygeometa.core import (read_mcf, pretty_print, render_j2_template,
-                            get_charstring, normalize_datestring,
-                            prune_distribution_formats,
+                            get_charstring, import_metadata,
+                            normalize_datestring, prune_distribution_formats,
                             prune_transfer_option, MCFReadError,
-                            MCFValidationError, SCHEMAS, validate_mcf)
+                            MCFValidationError, SCHEMAS, transform_metadata,
+                            validate_mcf)
 from pygeometa.helpers import json_dumps
 from pygeometa.schemas import (get_supported_schemas, InvalidSchemaError,
                                load_schema)
@@ -229,10 +230,16 @@ def test_get_supported_schemas(self):
                          'Expected specific number of supported schemas')
         self.assertEqual(sorted(schemas),
                          sorted(['dcat', 'iso19139', 'iso19139-2',
-                                 'iso19139-hnap', 'oarec-record', 'stac-item',
-                                 'wmo-cmp', 'wmo-wcmp2', 'wmo-wigos']),
+                                 'iso19139-hnap', 'oarec-record',
+                                 'stac-item', 'wmo-cmp', 'wmo-wcmp2',
+                                 'wmo-wigos']),
                          'Expected exact list of supported schemas')
 
+        schemas = get_supported_schemas(include_autodetect=True)
+        self.assertEqual(len(schemas), 10,
+                         'Expected specific number of supported schemas')
+        self.assertIn('autodetect', schemas, 'Expected autodetect in list')
+
     def test_render_j2_template(self):
         """test template rendering"""
 
@@ -397,8 +404,8 @@ def test_validate_mcf(self):
         with self.assertRaises(MCFValidationError):
             is_valid = validate_mcf({'foo': 'bar'})
 
-    def test_import(self):
-        """test metadata import"""
+    def test_schema_import(self):
+        """test direct metadata schema import"""
 
         schema = ISO19139OutputSchema()
 
@@ -434,6 +441,46 @@ def test_import(self):
             self.assertEqual(expected_bbox, result_bbox,
                              'Expected specific BBOX')
 
+    def test_import_metadata(self):
+        """test metadata import"""
+
+        with open(get_abspath('md-SMJP01RJTD-gmd.xml')) as fh:
+            mcf = import_metadata('iso19139', fh.read())
+
+            self.assertEqual(
+                mcf['identification']['title'],
+                'WIS/GTS bulletin SMJP01 RJTD in FM12 SYNOP',
+                'Expected specific title')
+
+        with open(get_abspath('md-SMJP01RJTD-gmd.xml')) as fh:
+            mcf = import_metadata('autodetect', fh.read())
+
+            self.assertEqual(
+                mcf['identification']['title'],
+                'WIS/GTS bulletin SMJP01 RJTD in FM12 SYNOP',
+                'Expected specific title')
+
+    def test_transform_metadata(self):
+        """test metadata transform"""
+
+        with open(get_abspath('md-SMJP01RJTD-gmd.xml')) as fh:
+            m = transform_metadata('iso19139', 'oarec-record', fh.read())
+
+            m = json.loads(m)
+            self.assertEqual(
+                m['properties']['title'],
+                'WIS/GTS bulletin SMJP01 RJTD in FM12 SYNOP',
+                'Expected specific title')
+
+        with open(get_abspath('md-SMJP01RJTD-gmd.xml')) as fh:
+            m = transform_metadata('autodetect', 'oarec-record', fh.read())
+
+            m = json.loads(m)
+            self.assertEqual(
+                m['properties']['title'],
+                'WIS/GTS bulletin SMJP01 RJTD in FM12 SYNOP',
+                'Expected specific title')
+
 
 def get_abspath(filepath):
     """helper function absolute file access"""