You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
from dataclasses import dataclass
|
|
import datetime
|
|
from dataclasses_json import dataclass_json
|
|
import requests
|
|
import json
|
|
import pprint
|
|
|
|
|
|
@dataclass_json
|
|
@dataclass
|
|
class Basiswetterdaten:
|
|
ts: datetime.datetime
|
|
stationsname: str
|
|
outtemp: float = None
|
|
outluftfeuchte: float = None
|
|
heatindex: float = None
|
|
taupunkt: float = None
|
|
luftdruck: float = None
|
|
regenaktuell: float = None
|
|
regenrate: float = None
|
|
wind: float = None
|
|
windrichtung: str = None
|
|
windboe: float = None
|
|
windboe_richtung: str = None
|
|
windrichtung_grad: float = None
|
|
out_abs_luftfeuchte: float = None
|
|
windchill: float = None
|
|
|
|
|
|
@dataclass_json
|
|
@dataclass
|
|
class Zusatzwetterdaten:
|
|
ts: datetime.datetime
|
|
stationsname: str
|
|
wertname: str
|
|
wert: float
|
|
public: bool
|
|
|
|
|
|
def status_auswerten(r, logger, daten):
|
|
if not (r.status_code == 200 or r.status_code == 201):
|
|
logger.error(f"Statuscode: {r.status_code}\n Message: {r.text}")
|
|
logger.error(pprint.pprint(daten))
|
|
|
|
|
|
def sende_daten(url, table, headers, daten, logger):
|
|
url = f"{url}{table}"
|
|
for data in daten:
|
|
data.ts = data.ts.strftime("%Y-%m-%d %H:%M:%S")
|
|
logger.debug(f"Folgende Daten werden gesendet an {table}:\n {daten}")
|
|
r = requests.post(url, headers=headers, json=[data.to_dict() for data in daten])
|
|
status_auswerten(r, logger, daten)
|
|
if r.status_code == 409:
|
|
logger.error("Primary Key Verletzung. Beginne Datensätze einzeln zu übertragen")
|
|
for data in daten:
|
|
r = requests.post(url, headers=headers, json=data.to_dict())
|
|
status_auswerten(r, logger, data)
|
|
|
|
|
|
def hole_letzten_ts(url, table, headers, username):
|
|
query = f"select=ts&stationsname=eq.{username}&limit=1&order=ts.desc"
|
|
url = f"{url}{table}"
|
|
r = requests.get(url, headers=headers, params=query)
|
|
if r.status_code == 200:
|
|
inhalt = json.loads(r.text)
|
|
try:
|
|
ts_str = inhalt[0]["ts"]
|
|
ts = datetime.datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S%z")
|
|
except IndexError:
|
|
ts = datetime.datetime(1970, 1, 1)
|
|
return ts
|
|
else:
|
|
raise TypeError(r.text)
|