nemunaire/epaper
1
0
Fork 0
Code Issues Pull Requests Releases Activity

ratp: Use disruptions_bulk route
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
nemunaire 2024-04-02 11:29:22 +02:00
parent 8be48bb062
commit adbd46f715
1 changed files with 56 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
modules/ratp.py
View File

@ -87,7 +87,7 @@ class IDFMAPI:
self.fnt_R_path = os.path.join(config.fonts_dir, IDFMAPI.fnt_R_path)
self.fnt_RB_path = os.path.join(config.fonts_dir, IDFMAPI.fnt_RB_path)
self.baseurl = "https://prim.iledefrance-mobilites.fr/marketplace/v2"
self.baseurl = "https://prim.iledefrance-mobilites.fr/marketplace/disruptions_bulk"
self.apikey = apikey or os.environ["TOKEN_IDFM"]
self._cached_file = ".ratp-%s.cache"
@ -156,7 +156,7 @@ 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(self.baseurl + "/navitia/line_reports/line_reports?count=2000")
req = urllib.request.Request(self.baseurl + "/disruptions/v2")
req.headers["apikey"] = self.apikey
try:
with urllib.request.urlopen(req) as f:
@ -192,15 +192,25 @@ class IDFMAPI:
return ret
def get_line_weather(self, res, disruptions, mode, line):
for l in res["line_reports"]:
if "line" not in l:
for l in res["lines"]:
if l["id"] == "line:IDFM:" + IDFMAPI.lines[mode][line]:
for impactedObject in l["impactedObjects"]:
for disruptionId in impactedObject["disruptionIds"]:
status = "past"
for ap in disruptions[disruptionId]["applicationPeriods"]:
end = datetime.strptime(ap["end"], "%Y%m%dT%H%M%S")
if end < datetime.now():
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"]]
begin = datetime.strptime(ap["begin"], "%Y%m%dT%H%M%S")
if begin > datetime.now():
status = "future"
continue
status = "active"
break
disruptions[disruptionId]["status"] = status
yield disruptions[disruptionId]
return None
@ -258,40 +268,41 @@ class RATPWeatherModule:
return icon
for disruption in weather[mode][line]:
if "messages" not in disruption:
if "message" not in disruption:
continue
if disruption["status"] != "active" and "application_periods" not in disruption:
if disruption["id"] in id_seens:
continue
if disruption["status"] != "active" and "applicationPeriods" not in disruption:
continue
title = ""
subtitle = ""
content = ""
oneline = False
if "application_periods" in disruption:
if "applicationPeriods" in disruption:
now = datetime.now()
yesterday = now + timedelta(days=-1)
nextweek = now + timedelta(days=1)
application_periods = []
for ap in disruption["application_periods"]:
ap["begin"] = datetime.strptime(ap["begin"], "%Y%m%dT%H%M%S")
ap["end"] = datetime.strptime(ap["end"], "%Y%m%dT%H%M%S")
for ap in disruption["applicationPeriods"]:
begin = datetime.strptime(ap["begin"], "%Y%m%dT%H%M%S")
end = datetime.strptime(ap["end"], "%Y%m%dT%H%M%S")
if ap["end"] < now:
if end < now:
continue
elif ap["begin"] > nextweek:
elif begin > nextweek:
continue
elif len(application_periods) > 0 and application_periods[0]["begin"] + timedelta(hours=16) < ap["begin"]:
elif len(application_periods) > 0 and application_periods[0]["begin"] + timedelta(hours=16) < begin:
continue
else:
application_periods.append(ap)
application_periods.append({"begin": begin, "end": end})
if ap["begin"] < yesterday:
if begin < yesterday:
oneline = True
if "cause" in disruption and disruption["cause"] == "travaux" and ap["begin"] > now:
if "cause" in disruption and disruption["cause"].lower() == "travaux" and begin > now:
oneline = True
if len(application_periods) == 0:
@ -302,17 +313,11 @@ class RATPWeatherModule:
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"])
id_seens.append(disruption["id"])
yield {
"title": title,
"title": disruption["title"],
"subtitle": subtitle,
"description": content,
"description": IDFMAPI.fromHTMLDisruption(disruption["message"]),
"icon": alert_icon(mode, line),
"oneline": oneline,
}
@ -342,13 +347,27 @@ class RATPWeatherModule:
states = []
for disruption in weather[mode][line]:
if disruption["status"] != "active":
status = "past"
for ap in disruption["applicationPeriods"]:
end = datetime.strptime(ap["end"], "%Y%m%dT%H%M%S")
if end < datetime.now():
continue
begin = datetime.strptime(ap["begin"], "%Y%m%dT%H%M%S")
if begin > datetime.now():
status = "future"
continue
status = "active"
break
if 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"])
if disruption["cause"] == "TRAVAUX":
states.append(disruption["severity"] + " " + disruption["cause"])
else:
states.append(disruption["severity"]["effect"])
states.append(disruption["severity"])
fill = "darkgray"
if "NO_SERVICE" in states: