Use navitia IDFM API directly
This commit is contained in:
parent
b77f1cd0d0
commit
a7f428ace1
155
modules/ratp.py
155
modules/ratp.py
@ -9,6 +9,28 @@ import urllib.request
|
||||
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
def whenStr(date, now):
|
||||
if date < now + timedelta(days=1):
|
||||
return date.hours() + "h" + date.minutes()
|
||||
elif date < now + timedelta(days=7):
|
||||
weekday = date.weekday()
|
||||
if weekday == 0:
|
||||
return "lundi"
|
||||
elif weekday == 1:
|
||||
return "mardi"
|
||||
elif weekday == 2:
|
||||
return "mercredi"
|
||||
elif weekday == 3:
|
||||
return "jeudi"
|
||||
elif weekday == 4:
|
||||
return "vendredi"
|
||||
elif weekday == 5:
|
||||
return "samedi"
|
||||
else:
|
||||
return "dimanche"
|
||||
else:
|
||||
return date.strftime("%d %b")
|
||||
|
||||
class IDFMAPI:
|
||||
|
||||
fnt_R_path = "./fonts/Parisine-Regular.ttf"
|
||||
@ -65,24 +87,13 @@ class IDFMAPI:
|
||||
self.cache_timeout = config.cache_timeout
|
||||
self.max_cache_timeout = config.max_cache_timeout
|
||||
|
||||
def fromIVtoPRIM(src, line):
|
||||
def fromHTMLDisruption(src):
|
||||
cleanr = re.compile('<.*?>')
|
||||
cleanrA = re.compile('<a.*?>.*?</a>')
|
||||
return {
|
||||
"InfoChannelRef": {
|
||||
"value": "Perturbation" if src["severity"] >= 2 else ("Travaux" if src["severity"] == 1 else "Message"),
|
||||
},
|
||||
"Content": {
|
||||
"Message": [{
|
||||
"MessageType": "SHORT_MESSAGE",
|
||||
"MessageText": {
|
||||
"id": src["id"],
|
||||
"title": src["title"].replace("Métro " + line + " : ", '').replace("Ligne " + line + " : ", '').replace("Tramway " + line + " : ", ''),
|
||||
"value": re.sub(cleanr, '', re.sub(cleanrA, '', src["message"].replace(' ', ' ').replace('’', "'").replace('à', 'à').replace('é', 'é').replace('è', 'è').replace('<br>', ' ').replace('</p>', ' ').replace('Information Ile de France Mobilités :', ''))).strip(),
|
||||
}
|
||||
}],
|
||||
}
|
||||
}
|
||||
period = re.compile('Période :[^.]+. ')
|
||||
period2 = re.compile('Dates? :[^.]+. ')
|
||||
more = re.compile(" Plus d'informations sur [^.]+.")
|
||||
return re.sub(cleanr, '', re.sub(cleanrA, '', re.sub(period, '', re.sub(period2, '', re.sub(more, '', src.replace(' ', ' ').replace(' ', ' ').replace('’', "'").replace('à', 'à').replace('é', 'é').replace('è', 'è').replace('ê', 'ê').replace('û', 'û').replace('Î', 'Î').replace('<br>', ' ').replace('</p>', ' ').replace('Information Ile de France Mobilités :', '').replace("Les horaires du calculateur d'itinéraire tiennent compte des travaux. ", '').replace("à la demande de la Préfecture de Police et d'Île-de-France Mobilités, et ", '')))))).strip()
|
||||
|
||||
def get_schedules(self, mode, line, station, way="A+R"):
|
||||
if mode == "M":
|
||||
@ -128,16 +139,6 @@ class IDFMAPI:
|
||||
return [m for m in res["result"]["schedules"] if "message" in m and m["message"] != "Train sans arrêt"]
|
||||
|
||||
def get_weather(self):
|
||||
ret = {}
|
||||
|
||||
for mode in IDFMAPI.lines:
|
||||
ret[mode] = {}
|
||||
for line in IDFMAPI.lines[mode]:
|
||||
ret[mode][line] = self.get_line_weather(mode, line)
|
||||
|
||||
return ret
|
||||
|
||||
def get_line_weather(self, mode, line):
|
||||
cache_file = self._cached_file % ("ratp-disruptions")
|
||||
# Read the mod time
|
||||
statinfo = None
|
||||
@ -148,7 +149,8 @@ class IDFMAPI:
|
||||
|
||||
if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(minutes=self.cache_timeout) < datetime.now(tz=timezone.utc):
|
||||
# Do the request and save it
|
||||
req = urllib.request.Request("https://api-iv.iledefrance-mobilites.fr/disruptions")
|
||||
req = urllib.request.Request(self.baseurl + "/navitia/line_reports/coverage/fr-idf/line_reports?count=2000")
|
||||
req.headers["apikey"] = self.apikey
|
||||
try:
|
||||
with urllib.request.urlopen(req) as f:
|
||||
with open(cache_file, 'wb') as fd:
|
||||
@ -169,11 +171,29 @@ class IDFMAPI:
|
||||
with open(cache_file) as f:
|
||||
res = json.load(f)
|
||||
|
||||
for l in res["currentIT"]:
|
||||
if "id" in l:
|
||||
if l["id"] == "line:IDFM:" + IDFMAPI.lines[mode][line]:
|
||||
for d in l["disruptions"]:
|
||||
yield IDFMAPI.fromIVtoPRIM(d, line)
|
||||
disruptions = {}
|
||||
for d in res["disruptions"]:
|
||||
disruptions[d["id"]] = d
|
||||
|
||||
ret = {}
|
||||
|
||||
for mode in IDFMAPI.lines:
|
||||
ret[mode] = {}
|
||||
for line in IDFMAPI.lines[mode]:
|
||||
ret[mode][line] = self.get_line_weather(res, disruptions, mode, line)
|
||||
|
||||
return ret
|
||||
|
||||
def get_line_weather(self, res, disruptions, mode, line):
|
||||
for l in res["line_reports"]:
|
||||
if "line" not in l:
|
||||
continue
|
||||
|
||||
if "id" in l["line"]:
|
||||
if l["line"]["id"] == "line:IDFM:" + IDFMAPI.lines[mode][line]:
|
||||
for link in l["line"]["links"]:
|
||||
if disruptions[link["id"]]["status"] != "past":
|
||||
yield disruptions[link["id"]]
|
||||
|
||||
return None
|
||||
|
||||
@ -230,24 +250,54 @@ class RATPWeatherModule:
|
||||
return image
|
||||
return icon
|
||||
|
||||
for info in weather[mode][line]:
|
||||
if "InfoChannelRef" not in info or (info["InfoChannelRef"]["value"] != "Perturbation" and info["InfoChannelRef"]["value"] != "Travaux"):
|
||||
for disruption in weather[mode][line]:
|
||||
if "messages" not in disruption:
|
||||
continue
|
||||
|
||||
if "Message" not in info["Content"]:
|
||||
if disruption["status"] != "active" and "application_periods" not in disruption:
|
||||
continue
|
||||
|
||||
for msg in info["Content"]["Message"]:
|
||||
if "MessageType" not in msg or msg["MessageType"] != "SHORT_MESSAGE":
|
||||
continue
|
||||
title = ""
|
||||
subtitle = ""
|
||||
content = ""
|
||||
|
||||
if "id" in msg["MessageText"]:
|
||||
if msg["MessageText"]["id"] in id_seens:
|
||||
if "application_periods" in disruption:
|
||||
now = datetime.now()
|
||||
nextweek = now + timedelta(days=1)
|
||||
|
||||
application_periods = []
|
||||
for ap in disruption["application_periods"]:
|
||||
ap["begin"] = datetime.fromisoformat(ap["begin"])
|
||||
ap["end"] = datetime.fromisoformat(ap["end"])
|
||||
|
||||
if ap["end"] < now:
|
||||
continue
|
||||
id_seens.append(msg["MessageText"]["id"])
|
||||
elif ap["begin"] > nextweek:
|
||||
continue
|
||||
elif len(application_periods) > 0 and application_periods[0]["begin"] + timedelta(hours=16) < ap["begin"]:
|
||||
continue
|
||||
else:
|
||||
application_periods.append(ap)
|
||||
|
||||
if len(application_periods) == 0:
|
||||
continue
|
||||
elif len(application_periods) == 1:
|
||||
if application_periods[0]["begin"] > now:
|
||||
subtitle = "De " + whenStr(application_periods[0]["begin"], now) + " à " + whenStr(application_periods[0]["end"], now)
|
||||
elif application_periods[0]["end"] > now:
|
||||
subtitle = "Fin " + whenStr(application_periods[0]["end"], now)
|
||||
|
||||
for msg in disruption["messages"]:
|
||||
if msg["channel"]["name"] == "titre":
|
||||
title = msg["text"]
|
||||
elif msg["channel"]["name"] == "moteur":
|
||||
content = IDFMAPI.fromHTMLDisruption(msg["text"])
|
||||
elif len(content) == "":
|
||||
content = IDFMAPI.fromHTMLDisruption(msg["text"])
|
||||
yield {
|
||||
"title": msg["MessageText"]["title"] if "title" in msg["MessageText"] else "",
|
||||
"description": msg["MessageText"]["value"].replace(" ", " "),
|
||||
"title": title,
|
||||
"subtitle": subtitle,
|
||||
"description": content,
|
||||
"icon": alert_icon(mode, line),
|
||||
}
|
||||
|
||||
@ -275,11 +325,22 @@ class RATPWeatherModule:
|
||||
align_y += line_height + 6
|
||||
|
||||
states = []
|
||||
for info in weather[mode][line]:
|
||||
if "InfoChannelRef" in info:
|
||||
states.append(info["InfoChannelRef"]["value"])
|
||||
for disruption in weather[mode][line]:
|
||||
if disruption["status"] != "active":
|
||||
continue
|
||||
if "severity" in disruption:
|
||||
if disruption["cause"] == "travaux" and "line" not in [x["pt_object"]["embedded_type"] for x in disruption["impacted_objects"]]:
|
||||
states.append(disruption["severity"]["effect"] + " " + disruption["cause"])
|
||||
else:
|
||||
states.append(disruption["severity"]["effect"])
|
||||
|
||||
fill = "black" if "Perturbation" in states else ("gray" if "Travaux" in states else "darkgray")
|
||||
fill = "darkgray"
|
||||
if "NO_SERVICE" in states:
|
||||
fill = "black"
|
||||
elif "REDUCED_SERVICE" in states or "UNKNOWN_EFFECT" in states or "OTHER_EFFECT" in states:
|
||||
fill = "teal"
|
||||
elif len(states) > 0:
|
||||
fill = "gray"
|
||||
|
||||
icon = IDFMAPI.get_line_icon(mode, line, line_height, fill=fill)
|
||||
image.paste(icon, (align_x, align_y), icon)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user