99 lines
2.2 KiB
Python
99 lines
2.2 KiB
Python
|
|
|
|
from requests.structures import CaseInsensitiveDict
|
|
import json
|
|
import requests
|
|
from dateutil import parser
|
|
from datetime import datetime
|
|
from lora import lora_controller
|
|
from math import floor
|
|
|
|
|
|
class Departure:
|
|
|
|
id_pool = []
|
|
storage = dict()
|
|
|
|
@staticmethod
|
|
def get_id():
|
|
i = 0
|
|
while True:
|
|
i += 1
|
|
if i not in Departure.id_pool:
|
|
Departure.id_pool.append(i)
|
|
return i
|
|
|
|
@staticmethod
|
|
def clear():
|
|
remove = []
|
|
for did in Departure.storage:
|
|
d = Departure.storage[did]
|
|
if d.get_departure() < -35:
|
|
Departure.id_pool.remove(d.id)
|
|
remove.append(did)
|
|
for did in remove:
|
|
Departure.storage.pop(did)
|
|
|
|
@staticmethod
|
|
def fetch(stop_id, limit):
|
|
|
|
url = "https://jizdnirady.pmdp.cz/odjezdy/vyhledat"
|
|
|
|
headers = CaseInsensitiveDict()
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
data = {
|
|
"Stop": {
|
|
"StopId": stop_id
|
|
},
|
|
"DateAndTime": None,
|
|
"MaxResults": limit,
|
|
"FullResults": False
|
|
}
|
|
|
|
resp = requests.post(url, headers=headers, data=json.dumps(data), timeout=100).json()
|
|
|
|
for c in resp:
|
|
did = f"{stop_id}|{abs(c['ConnectionId']['ConnectionId'])}"
|
|
if did not in Departure.storage:
|
|
Departure(
|
|
did,
|
|
c["Line"]["TractionType"],
|
|
c["Line"]["Name"],
|
|
c["LastStopName"],
|
|
c["DepartureTime"],
|
|
0 if not c["DelayMin"] else c["DelayMin"]
|
|
)
|
|
else:
|
|
Departure.storage[did].update(
|
|
0 if not c["DelayMin"] else c["DelayMin"]
|
|
)
|
|
|
|
def __init__(self, did, type, line, last_stop, departure, delay):
|
|
departure = (parser.parse(departure)).timestamp()
|
|
if -(datetime.now().timestamp() - (departure + delay*60))/60 <= -1:
|
|
return
|
|
Departure.storage[did] = self
|
|
if len(last_stop) >= 21:
|
|
last_stop = last_stop[:20].strip() + "..."
|
|
self.did = did
|
|
self.id = Departure.get_id()
|
|
self.line = line
|
|
self.type = type
|
|
self.last_stop = last_stop
|
|
self.departure = departure
|
|
self.delay = delay
|
|
lora_controller.data(self)
|
|
|
|
def update(self, delay):
|
|
if delay != self.delay:
|
|
lora_controller.data(self)
|
|
self.delay = delay
|
|
|
|
def get_departure(self):
|
|
departure = -round(((datetime.now().timestamp() - (self.departure + self.delay*60))/60)*10)
|
|
return departure
|
|
|
|
def __repr__(self):
|
|
return f"{self.id}|{self.type}|{self.line}|{self.last_stop}|{self.get_departure()}"
|