Skip to content

Commit

Permalink
feat: handle both <cap:x> and <x> XML tags by introducing post-proces…
Browse files Browse the repository at this point in the history
…sing step
  • Loading branch information
RoryPTB committed Sep 3, 2024
1 parent 4fc1122 commit c9a8a8d
Show file tree
Hide file tree
Showing 2 changed files with 72 additions and 37 deletions.
16 changes: 14 additions & 2 deletions src/cap2geojson/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,18 @@
#
###############################################################################

from .convert import Converter
import logging

__version__ = '0.1.0-dev1'
from .convert import to_geojson

__version__ = "0.1.0-dev1"

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)


def convert(xml: str) -> dict:
return to_geojson(xml)
93 changes: 58 additions & 35 deletions src/cap2geojson/convert.py
Original file line number Diff line number Diff line change
@@ -1,40 +1,45 @@
import logging
import math
from typing import Union
import re
import xmltodict

logger = logging.getLogger(__name__)


def get_properties(alert: dict) -> dict:
"""Creates the properties object for the GeoJSON Feature object from the CAP alert.
"""Creates the properties object for the GeoJSON Feature object
from the CAP alert.
Args:
alert (dict): The extracted CAP alert object.
Returns:
dict: The formatted properties object.
"""
info = alert["cap:info"]
info = alert["info"]
return {
"identifier": alert["cap:identifier"],
"sender": alert["cap:sender"],
"sent": alert["cap:sent"],
"status": alert["cap:status"],
"msgType": alert["cap:msgType"],
"scope": alert["cap:scope"],
"category": info["cap:category"],
"event": info["cap:event"],
"urgency": info["cap:urgency"],
"severity": info["cap:severity"],
"certainty": info["cap:certainty"],
"effective": info["cap:effective"],
"onset": info["cap:onset"],
"expires": info["cap:expires"],
"senderName": info["cap:senderName"],
"headline": info["cap:headline"],
"description": info["cap:description"],
"instruction": info["cap:instruction"],
"web": info["cap:web"],
"contact": info["cap:contact"],
"areaDesc": get_area_desc(info["cap:area"]),
"identifier": alert["identifier"],
"sender": alert["sender"],
"sent": alert["sent"],
"status": alert["status"],
"msgType": alert["msgType"],
"scope": alert["scope"],
"category": info["category"],
"event": info["event"],
"urgency": info["urgency"],
"severity": info["severity"],
"certainty": info["certainty"],
"effective": info["effective"],
"onset": info["onset"],
"expires": info["expires"],
"senderName": info["senderName"],
"headline": info["headline"],
"description": info["description"],
"instruction": info["instruction"],
"web": info["web"],
"contact": info["contact"],
"areaDesc": get_area_desc(info["area"]),
}


Expand All @@ -49,8 +54,8 @@ def get_area_desc(area: Union[dict, list]) -> str:
str: The formatted area description.
"""
if isinstance(area, dict):
return area["cap:areaDesc"]
return ", ".join([a["cap:areaDesc"] for a in area])
return area["areaDesc"]
return ", ".join([a["areaDesc"] for a in area])


def get_all_circle_coords(
Expand All @@ -70,8 +75,9 @@ def get_all_circle_coords(
list: The n estimated coordinates of the circle.
"""

def get_circle_coord(theta: float, x_centre: float,
y_centre: float, radius: float) -> list:
def get_circle_coord(
theta: float, x_centre: float, y_centre: float, radius: float
) -> list:
x = radius * math.cos(theta) + x_centre
y = radius * math.sin(theta) + y_centre
# Round to 5 decimal places to prevent excessive precision
Expand All @@ -94,16 +100,16 @@ def get_polygon_coordinates(single_area: dict) -> list:
Returns:
list: The list of polygon coordinate pairs.
"""
if "cap:circle" in single_area:
if "circle" in single_area:
# Takes form "x,y r"
centre, radius = map(float, single_area["cap:circle"].split(" "))
centre, radius = map(float, single_area["circle"].split(" "))
x_centre, y_centre = centre.split(",")
# Estimate the circle coordinates with n=100 points
return get_all_circle_coords(x_centre, y_centre, radius, 100)

if "cap:polygon" in single_area:
if "polygon" in single_area:
# Takes form "x,y x,y x,y" but with newlines that need to be removed
polygon_str = single_area["cap:polygon"].replace("\n", "").split()
polygon_str = single_area["polygon"].replace("\n", "").split()
return [list(map(float, coord.split(","))) for coord in polygon_str]

return []
Expand Down Expand Up @@ -131,20 +137,37 @@ def get_geometry(area: Union[dict, list]) -> dict:
}


def to_geojson(xml: bytes) -> dict:
def handle_namespace(xml: str) -> str:
"""Removes the 'cap:' prefix from the XML string tags,
so for example '<cap:info>' becomes '<info>' and '</cap:info>'
becomes '</info>'.
Args:
xml (str): The CAP XML string.
Returns:
str: The XML string with the 'cap:' prefix removed from the tags.
"""
xml = re.sub(r"<cap:(\w+)>", r"<\1>", xml)
xml = re.sub(r"</cap:(\w+)>", r"</\1>", xml)
return xml


def to_geojson(xml: str) -> dict:
"""Takes the CAP alert XML and converts it to a GeoJSON.
Args:
xml (bytes): The CAP XML byte string.
xml (str): The CAP XML string.
Returns:
dict: The final GeoJSON object.
"""
xml = handle_namespace(xml)
data = xmltodict.parse(xml)
alert = data["cap:alert"]

alert = data["alert"]
alert_properties = get_properties(alert)
alert_geometry = get_geometry(alert["cap:info"]["cap:area"])
alert_geometry = get_geometry(alert["info"]["area"])

return {
"type": "FeatureCollection",
Expand Down

0 comments on commit c9a8a8d

Please sign in to comment.