diff --git a/setup.py b/setup.py index 8423bb5..626d78c 100644 --- a/setup.py +++ b/setup.py @@ -4,6 +4,7 @@ requirements = [ 'begins', + 'geojson2kml', 'jsonschema', 'lxml', 'pyquery', diff --git a/src/formpack/constants.py b/src/formpack/constants.py index 608b013..a7dccd9 100644 --- a/src/formpack/constants.py +++ b/src/formpack/constants.py @@ -104,11 +104,13 @@ EXPORT_TYPE_CSV = 'csv' EXPORT_TYPE_GEOJSON = 'geojson' +EXPORT_TYPE_KML = 'kml' EXPORT_TYPE_SPSS = 'spss_labels' EXPORT_TYPE_XLS = 'xls' VALID_EXPORT_TYPES = [ EXPORT_TYPE_CSV, EXPORT_TYPE_GEOJSON, + EXPORT_TYPE_KML, EXPORT_TYPE_SPSS, EXPORT_TYPE_XLS, ] diff --git a/src/formpack/reporting/export.py b/src/formpack/reporting/export.py index bb815fe..c84331f 100644 --- a/src/formpack/reporting/export.py +++ b/src/formpack/reporting/export.py @@ -1,18 +1,22 @@ # coding: utf-8 import json import re +import tempfile import warnings import zipfile from collections import defaultdict, OrderedDict from inspect import isclass +from pathlib import Path from typing import ( Dict, Generator, Iterator, Optional, ) +from xml.etree import ElementTree as ET import xlsxwriter +from geojson2kml import convert_file as convert_geojson_file from ..constants import ( TAG_COLUMNS_AND_SEPARATORS, @@ -29,6 +33,12 @@ from ..utils.string import unique_name_for_xls from ..utils.text import get_valid_filename +KML_NAMESPACE = 'http://earth.google.com/kml/2.2' +GX_NAMESPACE = 'http://www.google.com/kml/ext/2.2' + +ET.register_namespace('', KML_NAMESPACE) +ET.register_namespace('gx', GX_NAMESPACE) + class Export: def __init__( @@ -780,6 +790,37 @@ def to_geojson( else: yield array_epilogue + def to_kml(self, submissions, flatten=True): + submissions = list(submissions) + geojson = "".join(self.to_geojson(submissions, flatten=flatten)) + geojson = json.loads(geojson) + + survey_instances = self._collect_survey_instances(submissions) + normalized_geojson = self._normalize_feature_collection(geojson) + + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_dir_path = Path(tmp_dir) + geojson_path = tmp_dir_path / 'export.geojson' + geojson_path.write_text( + json.dumps(normalized_geojson), encoding='utf-8' + ) + + kml_path = convert_geojson_file(str(geojson_path), tmp_dir) + with open(kml_path, 'r', encoding='utf-8') as handle: + kml_text = handle.read() + kml_text = re.sub( + r'http://www\.opengis\.net/kml/2\.2', + KML_NAMESPACE, + kml_text, + ) + + # Add ExtendedData elements to map properties correctly + kml_text = self._add_extended_data_to_kml( + kml_text, normalized_geojson, survey_instances + ) + + yield kml_text + def to_table(self, submissions): table = OrderedDict(((s, [list(l)]) for s, l in self.labels.items())) @@ -1021,3 +1062,228 @@ def to_spss_labels(self, output_file): filename, spss_label_commands.encode('utf-8-sig'), ) + + def _add_extended_data_to_kml( + self, kml_text, geojson_data, survey_instances + ): + """ + Add ExtendedData elements to KML placemarks for all GeoJSON properties. + + Parses the KML XML and inserts elements containing + children for each property in the corresponding GeoJSON feature. + """ + try: + root = ET.fromstring(kml_text) + except ET.ParseError: + return kml_text + + ns_uri = KML_NAMESPACE + ns = {'kml': ns_uri} + + document = root.find('kml:Document', ns) + if document is not None and document.find('kml:name', ns) is None: + document.insert(0, ET.Element(f'{{{ns_uri}}}name')) + + for style in root.findall('.//kml:Style', ns): + self._style_kml_icon(ET, style, ns_uri) + + features = geojson_data.get('features', []) + placemarks = root.findall('.//kml:Placemark', ns) + if not (len(placemarks) == len(features) == len(survey_instances)): + raise RuntimeError( + 'KML placemark count does not match GeoJSON feature count ' + 'or survey instance count' + ) + + for placemark, feature, survey_instance in zip( + placemarks, features, survey_instances + ): + properties = feature.get('properties', {}) + self._update_placemark_metadata( + ET, placemark, properties, survey_instance, ns + ) + + kml_output = ET.tostring(root, encoding='unicode', method='xml') + if not kml_output.startswith('\n' + kml_output + return kml_output + + def _build_extended_data(self, ET, properties): + extended_data = ET.Element(f'{{{KML_NAMESPACE}}}ExtendedData') + for key, value in properties.items(): + data = ET.SubElement(extended_data, f'{{{KML_NAMESPACE}}}Data') + data.set('name', str(key)) + value_elem = ET.SubElement(data, f'{{{KML_NAMESPACE}}}value') + value_elem.text = str(value) if value is not None else '' + return extended_data + + def _collect_survey_instances(self, submissions): + survey_instances = [] + first_section_name = get_first_occurrence(self.sections.keys()) + self.reset() + + for submission in submissions: + version = self.get_version_for_submission(submission) + formatted_chunks = self.parse_one_submission(submission, version) + if not formatted_chunks: + continue + + all_fields = version.sections[first_section_name].fields.values() + all_geo_fields = [ + field for field in all_fields if field.data_type in GEO_TYPES + ] + rows = formatted_chunks[first_section_name] + + survey_instance = submission.get('_uuid') + if not survey_instance: + survey_instance = submission.get('meta/rootUuid') + if not survey_instance: + survey_instance = submission.get('meta/instanceID') + if isinstance(survey_instance, str): + survey_instance = survey_instance.replace('uuid:', '', 1) + + for geo_field in all_geo_fields: + for _row in rows: + try: + geo_response = submission[geo_field.path] + field_and_response_to_geometry(geo_field, geo_response) + except (KeyError, FormPackGeoJsonError, RuntimeError): + continue + survey_instances.append(survey_instance) + + return survey_instances + + def _geometry_element(self, placemark, ns): + geometry = placemark.find('kml:Point', ns) + if geometry is not None: + return geometry + geometry = placemark.find('kml:LineString', ns) + if geometry is not None: + return geometry + return placemark.find('kml:Polygon', ns) + + def _normalize_feature_collection(self, payload): + if isinstance(payload, list): + return self._normalize_feature_list(payload) + if isinstance(payload, dict): + return self._normalize_feature_dict(payload) + return payload + + def _normalize_feature_dict(self, payload): + normalized = dict(payload) + features = payload.get('features', []) + if not isinstance(features, list): + return normalized + + normalized_features = [] + for feature in features: + if not isinstance(feature, dict): + normalized_features.append(feature) + continue + + normalized_feature = dict(feature) + properties = feature.get('properties', {}) + if isinstance(properties, dict): + normalized_feature['properties'] = OrderedDict( + properties.items() + ) + normalized_features.append(normalized_feature) + + normalized['features'] = normalized_features + return normalized + + def _normalize_feature_list(self, payload): + features = [] + name = None + for item in payload: + if not isinstance(item, dict): + continue + item_features = item.get('features') + if isinstance(item_features, list): + features.extend(item_features) + if not name and item.get('name'): + name = item['name'] + continue + if isinstance(item.get('geometry'), dict): + features.append(item) + + if not features: + return {'type': 'FeatureCollection', 'features': []} + + return { + 'type': 'FeatureCollection', + 'name': name, + 'features': [ + self._normalize_feature_collection(feature) + for feature in features + ], + } + + def _style_kml_icon(self, ET, style, ns_uri): + ns = {'kml': ns_uri} + icon_style = style.find('kml:IconStyle', ns) + if icon_style is None: + return + + self._upsert_kml_child(ET, icon_style, ns_uri, 'scale', '1.3') + + icon = self._upsert_kml_child(ET, icon_style, ns_uri, 'Icon') + self._upsert_kml_child( + ET, + icon, + ns_uri, + 'href', + 'https://maps.google.com/mapfiles/kml/paddle/red-circle.png', + ) + + hotspot = self._upsert_kml_child(ET, icon_style, ns_uri, 'hotSpot') + hotspot.set('x', '32') + hotspot.set('y', '1') + hotspot.set('xunits', 'pixels') + hotspot.set('yunits', 'pixels') + + list_style = self._upsert_kml_child(ET, style, ns_uri, 'ListStyle') + item_icon = self._upsert_kml_child(ET, list_style, ns_uri, 'ItemIcon') + self._upsert_kml_child( + ET, + item_icon, + ns_uri, + 'href', + 'https://maps.google.com/mapfiles/kml/paddle/red-circle-lv.png', + ) + + def _update_placemark_metadata( + self, ET, placemark, properties, survey_instance, ns + ): + name = placemark.find('kml:name', ns) + if name is not None: + placemark.remove(name) + + description = placemark.find('kml:description', ns) + if description is not None and survey_instance: + description.text = ( + f'\n Survey Instance: {survey_instance}\n ' + ) + + if not properties: + return + + existing_extended_data = placemark.find('kml:ExtendedData', ns) + if existing_extended_data is not None: + placemark.remove(existing_extended_data) + + extended_data = self._build_extended_data(ET, properties) + geometry = self._geometry_element(placemark, ns) + if geometry is not None: + geometry_index = list(placemark).index(geometry) + placemark.insert(geometry_index, extended_data) + return + placemark.append(extended_data) + + def _upsert_kml_child(self, ET, parent, ns_uri, tag, text=None): + child = parent.find(f'kml:{tag}', {'kml': ns_uri}) + if child is None: + child = ET.SubElement(parent, f'{{{ns_uri}}}{tag}') + if text is not None: + child.text = text + return child diff --git a/tests/test_exports.py b/tests/test_exports.py index 60fe1b9..f7ccbf4 100644 --- a/tests/test_exports.py +++ b/tests/test_exports.py @@ -6,6 +6,7 @@ import pathlib import tempfile import unittest +import xml.etree.ElementTree as ET from collections import OrderedDict from dateutil import parser from io import BytesIO, TextIOWrapper @@ -18,6 +19,7 @@ from formpack import FormPack from formpack.constants import UNTRANSLATED from formpack.errors import TranslationError +from formpack.reporting.export import Export from formpack.schema.fields import ( ValidationStatusCopyField, IdCopyField, @@ -2948,6 +2950,434 @@ def test_geojson_shape(self): ], } + def test_kml_export_contains_all_geo_types(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + self.assertEqual(len(fp.versions), 2) + + export = fp.export(versions=fp.versions.keys()) + kml_str = ''.join(export.to_kml(submissions, flatten=True)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + placemarks = root.findall('.//kml:Placemark', namespace) + self.assertGreaterEqual(len(placemarks), 3) + + geometry_tags = { + 'Point': any( + placemark.find('kml:Point', namespace) is not None + for placemark in placemarks + ), + 'LineString': any( + placemark.find('kml:LineString', namespace) is not None + for placemark in placemarks + ), + 'Polygon': any( + placemark.find('kml:Polygon', namespace) is not None + for placemark in placemarks + ), + } + self.assertDictEqual( + geometry_tags, + { + 'Point': True, + 'LineString': True, + 'Polygon': True, + }, + ) + + def test_kml_export_always_flattens(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + # Call with flatten=False + kml_str = ''.join(export.to_kml(submissions, flatten=False)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + placemarks = root.findall('.//kml:Placemark', namespace) + + # If flattening failed, you'd see nested FeatureCollections or missing geometry. + self.assertGreaterEqual(len(placemarks), 3) + + def test_kml_export_accepts_submission_iterator(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + submission_iterator = iter(submissions) + kml_str = ''.join(export.to_kml(submission_iterator, flatten=True)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + placemarks = root.findall('.//kml:Placemark', namespace) + + self.assertGreaterEqual(len(placemarks), 3) + + def test_kml_export_schema_data_types_map_correctly(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + kml_str = ''.join(export.to_kml(submissions, flatten=True)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + placemarks = root.findall('.//kml:Placemark', namespace) + + # Map schema field names to expected KML geometry tags + expected = { + 'geopoint': 'Point', + 'geotrace': 'LineString', + 'geoshape': 'Polygon', + } + + for field_name, geometry_tag in expected.items(): + found = any( + placemark.find(f'kml:{geometry_tag}', namespace) is not None + for placemark in placemarks + ) + self.assertTrue( + found, f"{field_name} should produce {geometry_tag}" + ) + + def test_kml_export_maps_geojson_properties_to_extended_data(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + kml_str = ''.join(export.to_kml(submissions[:1], flatten=True)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + + extended_data_elements = root.findall('.//kml:ExtendedData', namespace) + + # Should have at least one ExtendedData element per placemark + _ = root.findall('.//kml:Placemark', namespace) + self.assertGreater( + len(extended_data_elements), + 0, + 'KML should contain ExtendedData elements for properties', + ) + + # Each ExtendedData should contain Data elements with name and value + for extended_data in extended_data_elements: + data_elements = extended_data.findall('kml:Data', namespace) + self.assertGreater( + len(data_elements), + 0, + 'ExtendedData should contain Data elements', + ) + + # Each Data element should have name attribute and value child + for data_elem in data_elements: + self.assertIsNotNone( + data_elem.get('name'), + 'Data element should have name attribute', + ) + value_elem = data_elem.find('kml:value', namespace) + self.assertIsNotNone( + value_elem, 'Data element should contain value element' + ) + + # Verify specific expected properties appear in ExtendedData + # (based on all_geo_types fixture fields) + expected_properties = { + 'start', + 'end', + 'Just_a_regular_text_question', + 'Multiple_choice_question', + 'Multiple_select_question', + } + found_properties = set() + for extended_data in extended_data_elements: + for data_elem in extended_data.findall('kml:Data', namespace): + prop_name = data_elem.get('name') + if prop_name in expected_properties: + found_properties.add(prop_name) + + self.assertGreater( + len(found_properties), + 0, + 'ExtendedData should contain expected properties from fixture', + ) + + def test_kml_extended_data_preserves_all_geojson_properties(self): + export = Export.__new__(Export) + kml_str = ( + '\n' + '\n' + ' \n' + ' \n' + ' None\n' + ' placeholder\n' + ' 1,2,0\n' + ' \n' + ' \n' + '' + ) + geojson_data = { + 'type': 'FeatureCollection', + 'features': [ + { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [1, 2, 0]}, + 'properties': OrderedDict( + [ + ('start', '2026-07-01T13:08:06.524-04:00'), + ('_uuid', '716ddcf9-e83a-41ce-855a-120c712df3e0'), + ('__version__', 'abc123'), + ( + 'meta/rootUuid', + 'uuid:716ddcf9-e83a-41ce-855a-120c712df3e0', + ), + ('nullable', None), + ] + ), + } + ], + } + + updated_kml = export._add_extended_data_to_kml( + kml_str, + geojson_data, + ['716ddcf9-e83a-41ce-855a-120c712df3e0'], + ) + root = ET.fromstring(updated_kml) + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + + data_names = [ + data_elem.get('name') + for data_elem in root.findall( + './/kml:ExtendedData/kml:Data', namespace + ) + ] + self.assertEqual( + data_names, + ['start', '_uuid', '__version__', 'meta/rootUuid', 'nullable'], + ) + + nullable_value = root.find( + ".//kml:ExtendedData/kml:Data[@name='nullable']/kml:value", + namespace, + ) + self.assertIsNotNone(nullable_value) + self.assertIsNone(nullable_value.text) + + def test_kml_extended_data_raises_on_count_mismatch(self): + export = Export.__new__(Export) + kml_str = ( + '\n' + '\n' + ' \n' + ' one\n' + ' two\n' + ' \n' + '' + ) + geojson_data = { + 'type': 'FeatureCollection', + 'features': [ + { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [1, 2, 0]}, + 'properties': OrderedDict([('start', '2026-07-01')]), + } + ], + } + + with self.assertRaises(RuntimeError): + export._add_extended_data_to_kml( + kml_str, + geojson_data, + ['716ddcf9-e83a-41ce-855a-120c712df3e0'], + ) + + def test_kml_extended_data_returns_original_on_parse_error(self): + export = Export.__new__(Export) + broken_kml = '' + + result = export._add_extended_data_to_kml( + broken_kml, + {'type': 'FeatureCollection', 'features': []}, + [], + ) + + self.assertEqual(result, broken_kml) + + def test_collect_survey_instances_falls_back_to_meta_root_uuid(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + submission = dict(submissions[0]) + submission.pop('_uuid', None) + submission['meta/rootUuid'] = ( + 'uuid:716ddcf9-e83a-41ce-855a-120c712df3e0' + ) + + survey_instances = export._collect_survey_instances([submission]) + + self.assertEqual( + survey_instances, + ['716ddcf9-e83a-41ce-855a-120c712df3e0'] * 3, + ) + + def test_collect_survey_instances_skips_invalid_and_uses_instance_id(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + invalid_submission = {'invalid': 'data'} + submission = dict(submissions[0]) + submission.pop('_uuid', None) + submission.pop('meta/rootUuid', None) + submission['meta/instanceID'] = 'instance-123' + submission['Trace'] = '1 2 3 4;broken' + + survey_instances = export._collect_survey_instances( + [invalid_submission, submission] + ) + + self.assertEqual(survey_instances, ['instance-123', 'instance-123']) + + def test_normalize_feature_collection_handles_list_and_passthrough(self): + export = Export.__new__(Export) + payload = [ + 'ignore-me', + { + 'type': 'FeatureCollection', + 'name': 'nested', + 'features': [ + { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [1, 2, 0]}, + 'properties': {'a': 1}, + } + ], + }, + { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [3, 4, 0]}, + 'properties': {'b': 2}, + }, + ] + + normalized = export._normalize_feature_collection(payload) + self.assertEqual(normalized['type'], 'FeatureCollection') + self.assertEqual(normalized['name'], 'nested') + self.assertEqual(len(normalized['features']), 2) + self.assertEqual( + export._normalize_feature_collection('passthrough'), + 'passthrough', + ) + + def test_normalize_feature_collection_handles_empty_and_mixed_dict_features( + self, + ): + export = Export.__new__(Export) + + self.assertEqual( + export._normalize_feature_collection([]), + {'type': 'FeatureCollection', 'features': []}, + ) + + normalized = export._normalize_feature_collection( + { + 'type': 'FeatureCollection', + 'features': [ + 'raw-value', + { + 'type': 'Feature', + 'geometry': {'type': 'Point', 'coordinates': [1, 2, 0]}, + 'properties': 'not-a-dict', + }, + ], + } + ) + self.assertEqual(normalized['features'][0], 'raw-value') + self.assertEqual( + normalized['features'][1]['properties'], + 'not-a-dict', + ) + + self.assertEqual( + export._normalize_feature_collection({'features': 'not-a-list'}), + {'features': 'not-a-list'}, + ) + + def test_style_kml_icon_ignores_styles_without_iconstyle(self): + export = Export.__new__(Export) + style = ET.fromstring( + '' + ) + + export._style_kml_icon(ET, style, 'http://earth.google.com/kml/2.2') + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + self.assertIsNone(style.find('kml:ListStyle', namespace)) + + def test_update_placemark_metadata_handles_missing_name_geometry_and_properties( + self, + ): + export = Export.__new__(Export) + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + + placemark = ET.fromstring( + '' + 'placeholder' + '' + ) + export._update_placemark_metadata( + ET, + placemark, + OrderedDict([('a', 1)]), + None, + namespace, + ) + self.assertIsNotNone(placemark.find('kml:ExtendedData', namespace)) + + placemark_no_props = ET.fromstring( + '' + 'placeholder' + '' + ) + export._update_placemark_metadata( + ET, + placemark_no_props, + OrderedDict(), + None, + namespace, + ) + self.assertIsNone( + placemark_no_props.find('kml:ExtendedData', namespace) + ) + + def test_kml_export_description_uses_survey_instance(self): + title, schemas, submissions = build_fixture('all_geo_types') + fp = FormPack(schemas, title) + export = fp.export(versions=fp.versions.keys()) + + submission = dict(submissions[0]) + submission['_uuid'] = '716ddcf9-e83a-41ce-855a-120c712df3e0' + + kml_str = ''.join(export.to_kml([submission], flatten=True)) + root = ET.fromstring(kml_str) + + namespace = {'kml': 'http://earth.google.com/kml/2.2'} + descriptions = root.findall( + './/kml:Placemark/kml:description', namespace + ) + + self.assertGreater(len(descriptions), 0) + for description in descriptions: + self.assertEqual( + description.text.strip(), + 'Survey Instance: 716ddcf9-e83a-41ce-855a-120c712df3e0', + ) + def test_geojson_invalid(self): title, schemas, _ = build_fixture('all_geo_types') fp = FormPack(schemas, title)