"""Center Weather Advisories (CWA)"""
import math
import re
from typing import Tuple
from shapely.geometry import LineString, Point, Polygon
from pyiem.models.cwa import CWAModel
from pyiem.nws.product import TextProduct
from pyiem.nws.ugc import str2time
from pyiem.util import LOG
LINE3 = re.compile(
r"(?P<loc>[A-Z1-9]{4}) CWA (?P<ddhhmi>[0-9]{6})\s?(?P<cor>COR)?"
)
LINE4 = re.compile(
r"(?P<loc>[A-Z1-9]{3,4}) CWA (?P<num>\d+) VALID UNTIL (?P<ddhhmi>[0-9]{6})"
)
LALO_RE = re.compile(
r"^(?P<d1>[NEWS])\s?(?P<v1>[\d]{4,5})\s*(?P<d2>[NEWS])\s?(?P<v2>[\d]{4,5})"
r"(?P<leftover>.*)$"
)
FROM_RE = re.compile(
r"""
^(?P<offset>[0-9]+)?\s?
(?P<drct>N|NE|NNE|ENE|E|ESE|SE|SSE|S|SSW|SW|WSW|W|WNW|NW|NNW)?\s?
(?P<loc>[A-Z0-9]{3})\s?(?P<leftover>.*)$
""",
re.VERBOSE,
)
NM_WIDE = re.compile(r"(\s|\.)(?P<width>\d+)\s?NM WIDE")
DIAMETER = re.compile(r"DIAM (?P<diameter>\d+)\s?NM")
CANCEL_LINE = re.compile("(CANCEL|ERROR|CNCL)")
dirs = {
"NNE": 22.5,
"ENE": 67.5,
"NE": 45.0,
"E": 90.0,
"ESE": 112.5,
"SSE": 157.5,
"SE": 135.0,
"S": 180.0,
"SSW": 202.5,
"WSW": 247.5,
"SW": 225.0,
"W": 270.0,
"WNW": 292.5,
"NW": 315.0,
"NNW": 337.5,
"N": 0,
"": 0,
}
KM_NM = 1.852
[docs]
def go2lonlat(lon0, lat0, direction, displacement):
"""http://stackoverflow.com/questions/7222382"""
# Radius of the Earth
R = 6378.1
# Bearing is 90 degrees converted to radians.
if isinstance(direction, str):
direction = dirs.get(direction, 0)
brng = math.radians(direction)
# Distance in km
d = displacement * KM_NM
# Current lat point converted to radians
lat1 = math.radians(lat0)
# Current long point converted to radians
lon1 = math.radians(lon0)
lat2 = math.asin(
math.sin(lat1) * math.cos(d / R)
+ math.cos(lat1) * math.sin(d / R) * math.cos(brng)
)
lon2 = lon1 + math.atan2(
math.sin(brng) * math.sin(d / R) * math.cos(lat1),
math.cos(d / R) - math.sin(lat1) * math.sin(lat2),
)
lat2 = math.degrees(lat2)
lon2 = math.degrees(lon2)
return lon2, lat2
[docs]
def parse_polygon(prod: TextProduct, line: str) -> Tuple[Polygon, str]:
"""Figure out what the polygon is!"""
# condition, and yes, le sigh
line = (
line.replace("FFROM ", "")
.replace("FROM ", "")
.replace("=", "")
.strip()
)
# Account for quasi common FROM typo
if line.startswith("ROM "):
line = line[4:]
# Condense multiple spaces
tokens = (" ".join(line.split())).split("-")
pts = []
narrative = None
workdone = []
for i, token in enumerate(tokens):
s = LALO_RE.match(token.strip())
if s:
d = s.groupdict()
v1 = float(d["v1"]) / 100.0
v1 = v1 if d["d1"] not in ["S", "W"] else v1 * -1
v2 = float(d["v2"]) / 100.0
v2 = v2 if d["d2"] not in ["S", "W"] else v2 * -1
pts.append(
[
v2 if d["d2"] in ["E", "W"] else v1,
v1 if d["d1"] in ["S", "N"] else v2,
]
)
if d["leftover"]:
narrative = d["leftover"].strip()
break
continue
s = FROM_RE.match(token.strip())
if s:
d = s.groupdict()
if d["offset"] is not None:
(lon1, lat1) = go2lonlat(
prod.nwsli_provider[d["loc"]]["lon"],
prod.nwsli_provider[d["loc"]]["lat"],
d["drct"],
float(d["offset"]),
)
else:
(lon1, lat1) = (
prod.nwsli_provider[d["loc"]]["lon"],
prod.nwsli_provider[d["loc"]]["lat"],
)
workdone.append(f"{token} -> {lon1:.2f}, {lat1:.2f}")
pts.append((lon1, lat1))
if d["leftover"]:
# Could have a stray dash in here, so need to do some tricks
lookfor = d["loc"]
if d["offset"] is not None:
lookfor = f"{d['offset']}{d['drct']} {lookfor}"
narrative = "-".join(tokens[i:]).replace(lookfor, "").strip()
break
else:
narrative = "-".join(tokens[i:])
break
m = NM_WIDE.search(prod.unixtext)
if not pts:
return None, narrative
poly = None
if len(pts) >= 2 and m is not None:
res = m.groupdict()
# approx
width_deg = float(res["width"]) * KM_NM / 111.0
line = LineString(pts)
right = line.parallel_offset(width_deg / 2, "right", join_style=2)
left = line.parallel_offset(width_deg / 2, "left", join_style=2)
# NB This may be brittle to GEOS library version
poly = Polygon(list(left.coords) + list(right.coords[::-1]))
elif len(pts) == 1:
# We have a point
if m := DIAMETER.search(prod.unixtext):
res = m.groupdict()
# approx
diameter_deg = float(res["diameter"]) * KM_NM / 111.0
poly = Point(*pts[0]).buffer(diameter_deg / 2)
else:
poly = Polygon(pts)
if poly and not poly.is_valid:
poly = poly.buffer(0)
msg = "\n".join(workdone)
if any(
[not isinstance(poly, Polygon), not poly.is_valid, poly.is_empty]
):
prod.warnings.append(f"Polygon is not valid\n{msg}")
return None, narrative
msg = f"Polygon is not valid, but buffer(0) fixed it...\n{msg}"
LOG.warning(msg)
prod.warnings.append(msg)
return poly, narrative
[docs]
def parse_product(prod: TextProduct) -> CWAModel:
"""Do the parsing we need for the data model."""
lines = prod.unixtext.split("\n")
# This is not tenable at the moment
for ln in [4, 5]:
m = CANCEL_LINE.findall(lines[ln])
if m:
return None
m = LINE3.match(lines[2])
if m is None:
prod.warnings.append(f"Line 3 `{lines[2]}` not {LINE3.pattern}")
return None
res3 = m.groupdict()
issue = str2time(res3["ddhhmi"], prod.valid)
# Could fail, but this is a requirement anyway
res4 = LINE4.match(lines[3]).groupdict()
expire = str2time(res4["ddhhmi"], prod.valid)
# line work is not straight foward and could span multiple lines, sigh
poly, narrative = parse_polygon(prod, " ".join(lines[4:]))
if poly is None:
prod.warnings.append("CWA: No points found in polygon")
return None
return CWAModel(
center=res4["loc"],
issue=issue,
expire=expire,
geom=poly,
is_corrected=(res3["cor"] is not None),
narrative=narrative,
num=int(res4["num"]),
)
[docs]
class CWAProduct(TextProduct):
"""
Represents a Center Weather Advsiory (CWA) product.
"""
def __init__(
self, text, utcnow=None, ugc_provider=None, nwsli_provider=None
):
"""constructor"""
super().__init__(text, utcnow, ugc_provider, nwsli_provider)
self.data = parse_product(self)
# Need to do our faked AFOS, so other things work below
self.afos = f"CWA{self.source[1:]}"
[docs]
def sql(self, txn):
"""Do SQL related stuff that is required"""
data = self.data
if data is None:
return
if data.is_corrected:
txn.execute(
"DELETE from cwas where issue = %s and num = %s and "
"center = %s",
(data.issue, data.num, data.center),
)
if txn.rowcount == 0:
self.warnings.append("Corrected CWA updated no rows")
txn.execute(
"INSERT into cwas (issue, expire, center, num, narrative, "
"product_id, geom) VALUES (%s, %s, %s, %s, %s, %s, "
"ST_GeomFromText(%s, 4326))",
(
data.issue,
data.expire,
data.center,
data.num,
data.narrative,
self.get_product_id(),
data.geom.wkt,
),
)
[docs]
def get_jabbers(self, _uri, _uri2=None):
"""Return the Jabber for this sigmet"""
data = self.data
if data is None:
return []
apurl = (
"https://mesonet.agron.iastate.edu/plotting/auto/plot/226/"
f"network:CWSU::cwsu:{data.center}::num:{data.num}::"
f"issue:{data.issue:%Y-%m-%d%%20%H%M}::_r:86.png"
)
texturl = (
"https://mesonet.agron.iastate.edu/"
f"p.php?pid={self.get_product_id()}"
)
till = f"{data.expire:%-d %b %H%M}Z"
text = (
f"{data.center} issues CWA {data.num} till {till} ... "
f"{data.narrative} {texturl}"
)
html = (
f'<p>{data.center} issues <a href="{texturl}">CWA {data.num}</a>'
f" till {till}<br/>{data.narrative}</p>"
)
channels = ["CWA...", f"CWA{data.center}"]
xtra = {
"channels": ",".join(channels),
"twitter": text,
"twitter_media": apurl,
}
return [(text, html, xtra)]
[docs]
def parser(text, utcnow=None, ugc_provider=None, nwsli_provider=None):
"""Helper function"""
# Prevent an unnecessary ugc database load
if ugc_provider is None:
ugc_provider = {}
return CWAProduct(text, utcnow, ugc_provider, nwsli_provider)