PapaTutuWawa/mira-modules
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix: Format using black and fit to mira's new SM API

Browse Source
This commit is contained in:
PapaTutuWawa 2021-06-15 20:33:24 +02:00
parent 89cb06e29d
commit 3cfda79aa8
2 changed files with 94 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

185
papatutuwawa/mira/modules/pollen.py
View File

@ -1,4 +1,4 @@
''' """
Copyright (C) 2021 Alexander "PapaTutuWawa" Copyright (C) 2021 Alexander "PapaTutuWawa"
This program is free software: you can redistribute it and/or modify This program is free software: you can redistribute it and/or modify
@ -13,7 +13,7 @@ GNU General Public License for more details.
You should have received a copy of the GNU General Public License You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>. along with this program. If not, see <https://www.gnu.org/licenses/>.
''' """
# TODO: remove Alle sollte jede subscription entfernen # TODO: remove Alle sollte jede subscription entfernen
# TODO: Deal with weird casing and typos # TODO: Deal with weird casing and typos
# TODO: pollen hilfe <CMD> should print help per command # TODO: pollen hilfe <CMD> should print help per command
@ -25,113 +25,114 @@ from mira.module import BaseModule
import requests import requests
import aioxmpp import aioxmpp
NAME = 'pollen' NAME = "pollen"
API_ENDPOINT = "https://allergie.hexal.de/pollenflug/vorhersage/load_pollendaten.php"
API_ENDPOINT = 'https://allergie.hexal.de/pollenflug/vorhersage/load_pollendaten.php'
def api_query(plz, date): def api_query(plz, date):
return API_ENDPOINT + '?datum={}&plz={}'.format(date, plz) return API_ENDPOINT + "?datum={}&plz={}".format(date, plz)
def intensity_str(intensity): def intensity_str(intensity):
''' """
Map the intensity value to a human readable string Map the intensity value to a human readable string
''' """
return ('Kein Pollenflug', return ("Kein Pollenflug", "Schwach", "Mittelschwer", "Stark")[intensity]
'Schwach',
'Mittelschwer',
'Stark')[intensity]
class PollenModule(BaseModule): class PollenModule(BaseModule):
__instance = None __instance = None
_subcommand_table = {} _subcommand_table = {}
_pollen_data = {} # PLZ -> Date -> Type -> Severity _pollen_data = {} # PLZ -> Date -> Type -> Severity
_dates_notified = {} # JID -> [Dates already sent out] _dates_notified = {} # JID -> [Dates already sent out]
_sleep_duration = 100 _sleep_duration = 100
@staticmethod @staticmethod
def get_instance(base, **kwargs): def get_instance(base, **kwargs):
if PollenModule.__instance == None: if PollenModule.__instance == None:
PollenModule(base, **kwargs) PollenModule(base, **kwargs)
return PollenModule.__instance return PollenModule.__instance
def __init__(self, base, **kwargs): def __init__(self, base, **kwargs):
if PollenModule.__instance != None: if PollenModule.__instance != None:
raise Exception('Trying to init singleton twice') raise Exception("Trying to init singleton twice")
super().__init__(base, **kwargs) super().__init__(base, **kwargs)
PollenModule.__instance = self PollenModule.__instance = self
self._subcommand_table = { self._subcommand_table = {
'subscribe': self._subscribe, "subscribe": self._subscribe,
'unsubscribe': self._unsubscribe, "unsubscribe": self._unsubscribe,
'hilfe': self._help, "hilfe": self._help,
'*': self._any "*": self._any,
} }
self._sleep_duration = self.get_option('sleep_duration', 12 * 3600) self._sleep_duration = self.get_option("sleep_duration", 12 * 3600)
# Load data # Load data
# NOTE: Since we request the pollen_data at start anyway, we won't # NOTE: Since we request the pollen_data at start anyway, we won't
# need to save it. We only save dates_notified to prevent # need to save it. We only save dates_notified to prevent
# sending one notification too many. # sending one notification too many.
self._dates_notified = self._stm.get_data('dates_notified') self._dates_notified = self._stm.get_data("dates_notified")
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
periodic = loop.create_task(self._request_pollen_data(True)) periodic = loop.create_task(self._request_pollen_data(True))
def notification_body(self, plz, date, allergies): def notification_body(self, plz, date, allergies):
''' """
Format a message that will be sent to the user Format a message that will be sent to the user
''' """
msg = '{} (*{}*)\n'.format(date, plz) msg = "{} (*{}*)\n".format(date, plz)
if 'Alle' in [x['type'] for x in allergies]: if "Alle" in [x["type"] for x in allergies]:
allergies = list(self._pollen_data[plz][date].keys) allergies = list(self._pollen_data[plz][date].keys)
count = 0 count = 0
for allergy in allergies: for allergy in allergies:
intensity = self._pollen_data[plz][date][allergy['type']] intensity = self._pollen_data[plz][date][allergy["type"]]
if intensity >= allergy['level']: if intensity >= allergy["level"]:
msg += '{}: {}'.format(allergy['type'], intensity_str(intensity)) msg += "{}: {}".format(allergy["type"], intensity_str(intensity))
count += 1 count += 1
if count == 0: if count == 0:
return '' return ""
return msg return msg
def _is_notified(self, jid, date): def _is_notified(self, jid, date):
''' """
Returns True, when the pollen data for the date of date has Returns True, when the pollen data for the date of date has
already been sent to jid. already been sent to jid.
''' """
if not jid in self._dates_notified: if not jid in self._dates_notified:
return False return False
return date in self._dates_notified[jid] return date in self._dates_notified[jid]
def _set_dates_notified(self, jid, new_dates): def _set_dates_notified(self, jid, new_dates):
''' """
Marks a date as being sent to a user. Also ensures that Marks a date as being sent to a user. Also ensures that
there are at maximum 7 days in this list there are at maximum 7 days in this list
''' """
if not jid in self._dates_notified: if not jid in self._dates_notified:
self._dates_notified[jid] = new_dates self._dates_notified[jid] = new_dates
self._stm.set_data('dates_notified', self._dates_notified) self._stm.set_data("dates_notified", self._dates_notified)
return return
self._dates_notified[jid] += new_dates self._dates_notified[jid] += new_dates
self._dates_notified[jid] = self._dates_notified[jid][-7:] self._dates_notified[jid] = self._dates_notified[jid][-7:]
self._stm.set_data('dates_notified', self._dates_notified) self._stm.set_data("dates_notified", self._dates_notified)
def _broadcast_jid(self, jid, filter_allergies=[]): def _broadcast_jid(self, jid, filter_allergies=[]):
''' """
Sends the pollen data to jid. If filter_allergies is set, then Sends the pollen data to jid. If filter_allergies is set, then
only the data for the pollen types in filter_allergies will be sent. only the data for the pollen types in filter_allergies will be sent.
''' """
for plz, data in self._sum.get_subscriptions_for(jid).items(): for plz, data in self._sum.get_subscriptions_for_jid(jid).items():
if filter_allergies: if filter_allergies:
allergies = filter_allergies allergies = filter_allergies
else: else:
allergies = data['data'] allergies = data["data"]
notified = [] notified = []
for date in self._pollen_data[plz]: for date in self._pollen_data[plz]:
@ -139,21 +140,16 @@ class PollenModule(BaseModule):
continue continue
notified.append(date) notified.append(date)
msg = self.notification_body(plz, msg = self.notification_body(plz, date, allergies)
date,
allergies)
if msg: if msg:
self.send_message(aioxmpp.JID.fromstr(jid), self.send_message(aioxmpp.JID.fromstr(jid), msg)
msg)
if notified: if notified:
self._set_dates_notified(jid, notified) self._set_dates_notified(jid, notified)
def _broadcast_all(self): def _broadcast_all(self):
for plz in self._pollen_data: for plz in self._pollen_data:
for subscription in self._sum.get_subscriptions_for_keyword(plz): for jid, data in self._sum.get_subscriptions_for_keyword(plz).items():
jid = subscription[0]
data = subscription[1]
notified = [] notified = []
for date in self._pollen_data[plz]: for date in self._pollen_data[plz]:
@ -161,22 +157,22 @@ class PollenModule(BaseModule):
continue continue
notified.append(date) notified.append(date)
self.send_message(aioxmpp.JID.fromstr(jid), self.send_message(
self.notification_body(plz, aioxmpp.JID.fromstr(jid),
date, self.notification_body(plz, date, data),
data)) )
if notified: if notified:
self._set_dates_notified(jid, notified) self._set_dates_notified(jid, notified)
async def _request_pollen_data(self, loop): async def _request_pollen_data(self, loop):
while True: while True:
today = datetime.date.today().strftime('%Y-%m-%d') today = datetime.date.today().strftime("%Y-%m-%d")
for plz in self._sum.get_subscription_keywords(): for plz in self._sum.get_subscription_keywords():
req = requests.get(api_query(plz, today)) req = requests.get(api_query(plz, today))
data = req.json()['content'] data = req.json()["content"]
pollen = data['pollen'] pollen = data["pollen"]
for date in data['values']: for date in data["values"]:
if not plz in self._pollen_data: if not plz in self._pollen_data:
self._pollen_data[plz] = {} self._pollen_data[plz] = {}
if not date in self._pollen_data[plz]: if not date in self._pollen_data[plz]:
@ -184,7 +180,9 @@ class PollenModule(BaseModule):
for i in range(len(pollen)): for i in range(len(pollen)):
type_ = pollen[i] type_ = pollen[i]
self._pollen_data[plz][date][type_] = int(data['values'][date][i]) self._pollen_data[plz][date][type_] = int(
data["values"][date][i]
)
# For when we want to just refresh data as someone just subscribed # For when we want to just refresh data as someone just subscribed
if not loop: if not loop:
@ -192,10 +190,12 @@ class PollenModule(BaseModule):
self._broadcast_all() self._broadcast_all()
await asyncio.sleep(self._sleep_duration) await asyncio.sleep(self._sleep_duration)
async def _subscribe(self, cmd, msg): async def _subscribe(self, cmd, msg):
if len(cmd) < 3 or len(cmd) > 4: if len(cmd) < 3 or len(cmd) > 4:
self.send_message(msg.from_, 'Verwendung: pollen add <PLZ> <Allergen> [<Level>]') self.send_message(
msg.from_, "Verwendung: pollen add <PLZ> <Allergen> [<Level>]"
)
return return
plz = cmd[1] plz = cmd[1]
@ -203,69 +203,70 @@ class PollenModule(BaseModule):
jid = str(msg.from_.bare()) jid = str(msg.from_.bare())
level = int(cmd[3]) if len(cmd) == 4 else 0 level = int(cmd[3]) if len(cmd) == 4 else 0
if self._sum.is_subscribed_to_data(jid, plz, 'Alle'): if self._sum.is_subscribed_to_data(jid, plz, "Alle"):
self.send_message(msg.from_, 'Du hast schon alle Allergene abonniert') self.send_message(msg.from_, "Du hast schon alle Allergene abonniert")
return return
if self._sum.is_subscribed_to_data(jid, plz, allergy): if self._sum.is_subscribed_to_data(jid, plz, allergy):
self.send_message(msg.from_, 'Du hast das schon abonniert') self.send_message(msg.from_, "Du hast das schon abonniert")
return return
if level < 0 or level > 3: if level < 0 or level > 3:
self.send_message(msg.from_, 'Das Level muss zwischen 0 und 3 liegen') self.send_message(msg.from_, "Das Level muss zwischen 0 und 3 liegen")
return return
self._sum.append_data_for_subscription(jid, plz, { self._sum.append_subscription_data(
'type': allergy, jid, plz, {"type": allergy, "level": level}
'level': level )
}) self.send_message(msg.from_, "Du erhälst nun Pollenmeldungen zu %s" % allergy)
self.send_message(msg.from_, 'Du erhälst nun Pollenmeldungen zu %s' % allergy)
# Just some bandwidth saving measure # Just some bandwidth saving measure
if plz not in self._pollen_data: if plz not in self._pollen_data:
await self._request_pollen_data(False) await self._request_pollen_data(False)
self._broadcast_jid(jid, [{ self._broadcast_jid(jid, [{"type": allergy, "level": level}])
'type': allergy,
'level': level
}])
async def _unsubscribe(self, cmd, msg): async def _unsubscribe(self, cmd, msg):
if len(cmd) != 3: if len(cmd) != 3:
self.send_message(msg.from_, 'Verwendung: pollen remove <PLZ> <Allergen>') self.send_message(msg.from_, "Verwendung: pollen remove <PLZ> <Allergen>")
return return
plz = cmd[1] plz = cmd[1]
allergy = cmd[2] allergy = cmd[2]
jid = str(msg.from_.bare()) jid = str(msg.from_.bare())
if allergy == 'Alle': if allergy == "Alle":
self._sum.filter_items_for_subscriptions(jid, plz, lambda x: False) self._sum.filter_subscription_data_items(jid, plz, lambda x: False)
self.send_message(msg.from_, 'Du erhälst keine Pollenmeldungen mehr') self.send_message(msg.from_, "Du erhälst keine Pollenmeldungen mehr")
return
if not self._sum.is_subscribed_to_data_one(jid, plz, lambda x: x['type'] == allergy):
self.send_message(msg.from_, 'Du hast %s nicht abonniert' % (allergy))
return return
self._sum.filter_items_for_subscription(str(msg.from_.bare()), if not self._sum.is_subscribed_to_data_one(
plz, jid, plz, lambda x: x["type"] == allergy
lambda x: x['type'] != allergy) ):
self.send_message(msg.from_, 'Du erhälst nun keine Pollenmeldungen zu %s mehr' % (allergy)) self.send_message(msg.from_, "Du hast %s nicht abonniert" % (allergy))
return
self._sum.filter_subscription_data_items(
str(msg.from_.bare()), plz, lambda x: x["type"] != allergy
)
self.send_message(
msg.from_, "Du erhälst nun keine Pollenmeldungen zu %s mehr" % (allergy)
)
async def _help(self, cmd, msg): async def _help(self, cmd, msg):
body = '''Verfügbare Befehle: body = """Verfügbare Befehle:
pollen subscribe <PLZ> <Typ> [<Min. Level>] pollen subscribe <PLZ> <Typ> [<Min. Level>]
pollen unsubscribe <PLZ> <Typ> pollen unsubscribe <PLZ> <Typ>
pollen hilfe pollen hilfe
Pollentypen: Ambrosia, Ampfer, Beifuß, Birke, Buche, Eiche, Erle, Esche, Gräser, Hasel, Pappel, Roggen, Ulme, Wegerich, Weide, Alle (Alias für alle Pollentypen) Pollentypen: Ambrosia, Ampfer, Beifuß, Birke, Buche, Eiche, Erle, Esche, Gräser, Hasel, Pappel, Roggen, Ulme, Wegerich, Weide, Alle (Alias für alle Pollentypen)
Level: 0 (Kein Pollenflug) - 3 (Starker Pollenflug) Level: 0 (Kein Pollenflug) - 3 (Starker Pollenflug)
''' """
self.send_message(msg.from_, body) self.send_message(msg.from_, body)
async def _any(self, cmd, msg): async def _any(self, cmd, msg):
self.send_message(msg.from_, 'Unbekannter Befehl') self.send_message(msg.from_, "Unbekannter Befehl")
await self._help(cmd, msg) await self._help(cmd, msg)
def get_instance(base, **kwargs): def get_instance(base, **kwargs):
return PollenModule.get_instance(base, **kwargs) return PollenModule.get_instance(base, **kwargs)

2
setup.py
View File

@ -10,7 +10,7 @@ setup(
license = 'GPLv3', license = 'GPLv3',
packages = find_packages(), packages = find_packages(),
install_requires = [ install_requires = [
'mira>=0.2.0', 'git+https://git.polynom.me/PapaTutuWawa/mira.git@0.3.0',
'requests>=2.23.0' 'requests>=2.23.0'
], ],
zip_safe=True zip_safe=True