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