summaryrefslogtreecommitdiff
path: root/waterspout_radar/_radar.py
blob: 380c70bb9976975661e67a7f17a83e4ef2f768e1 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import dataclasses
import datetime
import typing

import numpy as np
import pint
import szilagyi
from geopy import geocoders
from metpy import calc
from metpy import units as metunits
from windy import point_forecast, Windy

_L = point_forecast.Level


@dataclasses.dataclass
class Prediction:
	time: datetime.datetime
	latitude: float
	longitude: float
	temperature_difference: pint.Quantity
	convective_cloud_depth: pint.Quantity
	wind: pint.Quantity
	low_clouds: float
	swi: float

	def json(self):
		return {
			"time": self.time.isoformat(),
			"latitude": self.latitude,
			"longitude": self.longitude,
			"dt": self.temperature_difference.m_as("kelvin"),
			"ccd": self.convective_cloud_depth.m_as("foot"),
			"wind": self.wind.m_as("knot"),
			"clouds": self.low_clouds,
			"swi": self.swi,
		}

	@classmethod
	def from_json(cls, json):
		return cls(
			datetime.datetime.fromisoformat(json["time"]),
			json["latitude"],
			json["longitude"],
			json["dt"] * metunits.units.kelvin,
			json["ccd"] * metunits.units.foot,
			json["wind"] * metunits.units.knot,
			json["clouds"],
			json["swi"])



def calculate(config) -> typing.List[Prediction]:
	units = metunits.units
	windy = Windy(units)

	def _calculate(latitude, longitude):
		forecasts = windy.point_forecast(
			config.key,
			latitude, longitude,
			point_forecast.Model.ICONEU,
			("temp", "dewpoint", "wind", "pressure", "lclouds"),
			tuple(_L))
		for cast in forecasts:
			dt = abs(cast.at("temp", _L.H850) - cast.at("temp", _L.SURFACE))
			pressure, _ = calc.lcl(
				cast.at("pressure", _L.SURFACE),
				cast.at("temp", _L.SURFACE),
				cast.at("dewpoint", _L.SURFACE))
			lcl = calc.pressure_to_height_std(pressure)
			pressure, _ = calc.el(cast["pressure"], cast["temp"], cast["dewpoint"])
			el = calc.pressure_to_height_std(pressure)
			ccd = (el - lcl).to(units.ft)
			clouds = cast["lclouds"][0].magnitude / 100
			try:
				swi = szilagyi.calculate_swi(dt, ccd)
			except ValueError:
				swi = -10
			wind = [cast.at("wind_u", _L.H850), cast.at("wind_v", _L.H850)]
			wind = np.array([x.m_as("kts") for x in wind])
			wind = np.linalg.norm(wind) * units.kts
			yield Prediction(cast.timestamp, latitude, longitude, dt, ccd, wind, clouds, swi)

	predictions = []
	locator = geocoders.Nominatim(user_agent="waterspout-radar")
	for location in config.locations:
		found = locator.geocode(location)
		predictions.extend(_calculate(found.latitude, found.longitude))
	return predictions