Add CovJSON CSV Formatter

Author: webb-ben

pygeoapi/formatter/csv_.py

@@ -71,6 +71,8 @@ def write(self, options: dict = {}, data: dict = None) -> str:
 
         if 'Feature' in type or 'features' in data:
             return self._write_from_geojson(options, data)
+        elif 'Coverage' in type or 'coverages' in data:
+            return self._write_from_covjson(options, data)
 
     def _write_from_geojson(
         self, options: dict = {}, data: dict = None, is_point=False
@@ -135,7 +137,79 @@ def _add_feature(
             LOGGER.error(err)
             raise FormatterSerializationError('Error writing CSV output')
 
+    def _write_from_covjson(
+        self, options: dict = {}, data: dict = None
+    ) -> str:
+        """
+        Generate CovJSON data in CSV format
+
+        :param options: CSV formatting options
+        :param data: dict of CovJSON data
+
+        :returns: string representation of format
+        """
+        LOGGER.debug('Processing CovJSON data for CSV output')
+        units = {}
+        for p, v in data['parameters'].items():
+            unit = v['unit']['symbol']
+            if isinstance(unit, dict):
+                unit = unit.get('value')
+
+            units[p] = unit
+
+        fields = ['parameter', 'datetime', 'value', 'unit', 'x', 'y']
+        LOGGER.debug(f'CSV fields: {fields}')
+        output = io.StringIO()
+        writer = csv.DictWriter(output, fields)
+        writer.writeheader()
+
+        if data['type'] == 'Coverage':
+            is_point = 'point' in data['domain']['domainType'].lower()
+            self._add_coverage(writer, units, data, is_point)
+        else:
+            [
+                self._add_coverage(writer, units, coverage, True)
+                for coverage in data['coverages']
+                if 'point' in coverage['domain']['domainType'].lower()
+            ]
         return output.getvalue().encode('utf-8')
 
+    @staticmethod
+    def _add_coverage(
+        writer: csv.DictWriter, units: dict, data: dict, is_point: bool = False
+    ) -> None:
+        """
+        Add coverage data to CSV writer
+
+        :param writer: CSV DictWriter
+        :param units: dict of parameter units
+        :param data: dict of CovJSON coverage data
+        :param is_point: whether the coverage is a point coverage
+        """
+
+        if is_point is False:
+            LOGGER.warning('Non-point coverages not supported for CSV output')
+            return
+
+        axes = data['domain']['axes']
+        time_range = range(len(axes['t']['values']))
+
+        try:
+            [
+                writer.writerow({
+                    'parameter': parameter,
+                    'datetime': axes['t']['values'][time_value],
+                    'value': data['ranges'][parameter]['values'][time_value],
+                    'unit': units[parameter],
+                    'x': axes['x']['values'][-1],
+                    'y': axes['y']['values'][-1]
+                })
+                for parameter in data['ranges']
+                for time_value in time_range
+            ]
+        except ValueError as err:
+            LOGGER.error(err)
+            raise FormatterSerializationError('Error writing CSV output')
+
     def __repr__(self):
         return f'<CSVFormatter> {self.name}'

tests/formatter/test_csv__formatter.py

@@ -138,3 +138,61 @@ def test_invalid_geometry_data(invalid_geometry_data):
     formatter = CSVFormatter({'geom': True})
     with pytest.raises(FormatterSerializationError):
         formatter.write(data=invalid_geometry_data)
+
+
+@pytest.fixture
+def point_coverage_data():
+    return {
+        'type': 'Coverage',
+        'domain': {
+            'type': 'Domain',
+            'domainType': 'PointSeries',
+            'axes': {
+                'x': {'values': [-10.1]},
+                'y': {'values': [-40.2]},
+                't': {'values': [
+                    '2013-01-01', '2013-01-02', '2013-01-03',
+                    '2013-01-04', '2013-01-05', '2013-01-06']}
+            }
+        },
+        'parameters': {
+            'PSAL': {
+                'type': 'Parameter',
+                'description': {'en': 'The measured salinity'},
+                'unit': {'symbol': 'psu'},
+                'observedProperty': {
+                    'id': 'http://vocab.nerc.ac.uk/standard_name/sea_water_salinity/',  # noqa
+                    'label': {'en': 'Sea Water Salinity'}
+                }
+            }
+        },
+        'ranges': {
+            'PSAL': {
+                'axisNames': ['t'],
+                'shape': [6],
+                'values': [
+                    43.9599, 43.9599, 43.9640, 43.9640, 43.9679, 43.987
+                ]
+            }
+        }
+    }
+
+
+def test_point_coverage_csv(point_coverage_data):
+    """Test CSV output of point coverage data"""
+    formatter = CSVFormatter({'geom': True})
+    output = formatter.write(data=point_coverage_data)
+    csv_reader = DictReader(StringIO(output.decode('utf-8')))
+    rows = list(csv_reader)
+
+    # Verify number of rows
+    assert len(rows) == 6
+
+    # Verify data
+    first_row = rows[0]
+    assert first_row['parameter'] == 'PSAL'
+    assert first_row['datetime'] == '2013-01-01'
+    assert first_row['value'] == '43.9599'
+    assert first_row['unit'] == 'psu'
+    assert first_row['x'] == '-10.1'
+    assert first_row['y'] == '-40.2'