[schema-org] support for the polygon concept in GeoShape

pygeometa/schemas/schema_org/__init__.py

@@ -47,6 +47,7 @@
 import logging
 import os
 from typing import Union
+from shapely.geometry import Polygon
 
 from pygeometa.core import get_charstring
 from pygeometa.helpers import generate_datetime, json_dumps
@@ -136,13 +137,14 @@ def import_(self, metadata: str) -> dict:
                 sc = _get_list_or_dict(md['spatial'])
 
             crs = 4326
-
-            if isinstance(sc, str):
+            if sc in [None, '']:
+                None
+            elif isinstance(sc, str):
                 # simple location name
                 mcf['identification']['extents']['spatial'].append({
                     'description': sc
                 })
-            elif 'geo' in sc:
+            elif isinstance(sc, dict) and 'geo' in sc:
                 geo = _get_list_or_dict(sc['geo'])
 
                 if geo['@type'] == 'GeoCoordinates':
@@ -153,8 +155,7 @@ def import_(self, metadata: str) -> dict:
                 elif geo['@type'] == 'GeoShape':
                     mcf['spatial']['datatype'] = 'vector'
                     mcf['spatial']['geomtype'] = 'polygon'
-                    bt = geo['box'].split()
-                    bbox = bt[1], bt[0], bt[3], bt[2]
+                    bbox = _get_box_from_coords(geo)
                 else:
                     bbox = [-180, -90, 180, 90]
 
@@ -575,6 +576,37 @@ def generate_link(self, distribution: dict) -> dict:
         return link
 
 
+def _get_box_from_coords(geo: dict) -> list:
+    """
+    Helper function to retrieve box from GeoCoordinates
+
+    :param geo: a schema-org geometry object
+
+    :returns: `list` bbox or None
+    """
+    if 'bbox' in geo:
+        bt = geo['bbox'].split()
+        if bt and len(bt) == 4:
+            return [float(bt[1]), float(bt[0]), float(bt[3]), float(bt[2])]
+    elif 'polygon' in geo:
+        # polygon string is space or comma separated
+        if ',' in geo['polygon']:
+            coords = geo['polygon'].split(",")
+            latlon_points = [
+                tuple(map(float, p.strip().split()))
+                for p in coords
+            ]
+        else:
+            coords = list(map(float, geo['polygon'].split()))
+            latlon_points = list(zip(coords[0::2], coords[1::2]))
+        # flip (lat, lon) → (lon, lat) for Shapely
+        lonlat_points = [(lon, lat) for lat, lon in latlon_points]
+        poly = Polygon(lonlat_points)
+        return list(poly.bounds)
+    else:
+        return None
+
+
 def _get_list_or_dict(value:
                       Union[None, list, dict, str]) -> Union[None, str, dict]:
     """

tests/run_tests.py

@@ -64,6 +64,7 @@
                                load_schema)
 from pygeometa.schemas.iso19139 import ISO19139OutputSchema
 from pygeometa.schemas.ogcapi_records import OGCAPIRecordOutputSchema
+from pygeometa.schemas.schema_org import _get_box_from_coords
 
 from sample_schema import SampleOutputSchema
 
@@ -514,6 +515,24 @@ def test_transform_metadata(self):
                 'WIS/GTS bulletin SMJP01 RJTD in FM12 SYNOP',
                 'Expected specific title')
 
+    def test_schema_org_coords(self):
+        """Test helper method schema-org parse geometry"""
+        geo1 = {
+            "@type": "GeoShape",
+            "bbox": "2 1 4 3"
+        }
+        self.assertEqual(_get_box_from_coords(geo1), [1, 2, 3, 4])
+        geo2 = {
+            "@type": "GeoShape",
+            "polygon": "2 1,4 1,2 3,2 1"
+        }
+        self.assertEqual(_get_box_from_coords(geo2), [1, 2, 3, 4])
+        geo3 = {
+            "@type": "GeoShape",
+            "polygon": "2 1 4 1 2 3 4 3 2 1"
+        }
+        self.assertEqual(_get_box_from_coords(geo3), [1, 2, 3, 4])
+
 
 def get_abspath(filepath):
     """helper function absolute file access"""