Spaces:
Sleeping
Sleeping
import requests | |
import json | |
import os | |
import xmltodict | |
import gzip | |
import shutil | |
import matplotlib.pyplot as plt | |
import matplotlib.font_manager as fm | |
def crawl(url, filename): | |
try: | |
try: | |
response = requests.get(url) | |
except Exception as e: | |
print("Error",e) | |
with open(f'{filename}.xml.gz', 'wb') as f: | |
f.write(response.content) | |
# 解壓縮 | |
with gzip.open(f'{filename}.xml.gz', 'rb') as f_in: | |
with open(f'{filename}.xml', 'wb') as f_out: | |
shutil.copyfileobj(f_in, f_out) | |
# xml to json | |
with open(f'{filename}.xml', 'r') as f: | |
data_dict = xmltodict.parse(f.read()) | |
data_json = json.dumps(data_dict, ensure_ascii=False) | |
# dump to Etag.json with json | |
json.dump(data_dict, open(f'{filename}.json', 'w'), ensure_ascii=False, indent=2) | |
except Exception as e: | |
print(e) | |
print(f'{url} failed') | |
print(f'{filename} done') | |
def timeSeries(startDate,endDate): | |
print("startDate: ", startDate, type(startDate)) | |
print("endDate: ", endDate, type(endDate)) | |
EtagPairId = '01F0339S-01F0376S' | |
EtagStart = '01F0339S' | |
EtagEnd = '01F0376S' | |
EtagStartRoad = ['五股','高公局','林口(文化一路)'] | |
VehicleTypeDict = { | |
31: "Car", | |
32: "Small Truck", | |
41: "Bus", # (含計程車) | |
42: "Heavy Truck", | |
5: "Trailer" | |
} | |
time_series_31data = {} | |
time_series_32data = {} | |
time_series_41data = {} | |
time_series_42data = {} | |
time_series_5data = {} | |
day_list = ['202405%02d'%(x) for x in range(startDate,endDate+1)] | |
hour_list = ['%02d'%(i) for i in range(24)] | |
five_minute_list = ['%02d'%(i) for i in range(0,60,5)] | |
print("Lens = ", len(day_list) * len(hour_list) * len(five_minute_list)) | |
for day in day_list: | |
for hour in hour_list: | |
for minute in five_minute_list: | |
i = hour + minute | |
filename = f'./timeSeriesData/ETagPairLive_{day}_{str(i)}' | |
# Check if the file exists | |
if not os.path.exists(f'{filename}.json'): | |
url = f'https://tisvcloud.freeway.gov.tw/history/motc20/ETag/{day}/ETagPairLive_{str(i)}.xml.gz' | |
crawl(url, filename) | |
try: | |
with open(f'{filename}.json', 'r') as f: | |
data = json.load(f) | |
EtagPairLive = data['ETagPairLiveList']['ETagPairLives']['ETagPairLive'] | |
for ETagPair in EtagPairLive: | |
if ETagPair['ETagPairID'] == EtagPairId: | |
# print(ETagPair['ETagPairID']) | |
Flow = ETagPair['Flows']['Flow'] | |
for vehicle in Flow: | |
# print(f"VehicleType: {VehicleTypeDict[int(vehicle['VehicleType'])]}, VehicleNum: {vehicle['VehicleCount']}, Speed: {vehicle['SpaceMeanSpeed']}, Time: {vehicle['TravelTime']}") | |
if int(vehicle['VehicleType']) == 31: | |
time_series_31data[f'{day}_{i}'] = int(vehicle['VehicleCount']) | |
elif int(vehicle['VehicleType']) == 32: | |
time_series_32data[f'{day}_{i}'] = int(vehicle['VehicleCount']) | |
elif int(vehicle['VehicleType']) == 41: | |
time_series_41data[f'{day}_{i}'] = int(vehicle['VehicleCount']) | |
elif int(vehicle['VehicleType']) == 42: | |
time_series_42data[f'{day}_{i}'] = int(vehicle['VehicleCount']) | |
elif int(vehicle['VehicleType']) == 5: | |
time_series_5data[f'{day}_{i}'] = int(vehicle['VehicleCount']) | |
except: | |
pass | |
step = 6 | |
plt.figure(figsize=(30,10)) | |
plt.plot(list(time_series_31data.keys())[::step], list(time_series_31data.values())[::step], label=VehicleTypeDict[31]) | |
plt.plot(list(time_series_32data.keys())[::step], list(time_series_32data.values())[::step], label=VehicleTypeDict[32]) | |
plt.plot(list(time_series_41data.keys())[::step], list(time_series_41data.values())[::step], label=VehicleTypeDict[41]) | |
plt.plot(list(time_series_42data.keys())[::step], list(time_series_42data.values())[::step], label=VehicleTypeDict[42]) | |
plt.plot(list(time_series_5data.keys())[::step], list(time_series_5data.values())[::step], label=VehicleTypeDict[5]) | |
plt.xlabel('Time') | |
plt.xticks(fontsize=14) | |
plt.xticks(rotation=60) | |
plt.ylabel('VehicleCount') | |
plt.title(f'{EtagStart} to {EtagEnd}') | |
plt.legend() | |
# plt.show() | |
return plt | |