### import moduleimportosimportrequestsimportraise_error_type### function# from ASRAD get Taiwan's water vapor color satellite image# the result is an .jpg image defget_taiwan_water_vapor_color_satellite_image(year:str='2024',month:str='09',day:str='05',hour:str='00',minute:str='00'):filename=year+month+day+'_'+hour+minute+'.PCCU.WVP.jpg'url=f'https://asrad.pccu.edu.tw/catalog/pccu/{year}/{month}/{day}/{filename}'response=requests.get(url)save_dir='water_vapor_color_satellite_image'ifnotos.path.exists(save_dir):os.mkdir(save_dir)ifresponse.status_code==200:file_path=f'{save_dir}\\{filename}'withopen(file=file_path,mode='wb')asjpg_file:jpg_file.write(response.content)print(f'Successful download \"{file_path}\".')else:raiseraise_error_type.ResponseError(f'request failed with status code {response.status_code}')
颱風資料
由颱風資料庫網站,可以看到其颱風清單的資料獲取 API ,由於網站資料是以 JSON 格式回傳,因此下載完成後可直接轉換成 Python 物件,方便後續分析與處理。
程式共實作兩個函式。第一個 get_taiwan_warning_typhon_data() 用來取得臺灣地區歷年曾發布海上或陸上颱風警報的颱風列表;第二個 get_northwest_pacific_typhon_data() 則取得西北太平洋所有颱風資料。兩者皆透過 HTTP POST 請求向中央氣象署取得資料,並將 JSON 回傳內容解析後提供程式使用。
### import moduleimportrequestsimportjsonimportraise_error_type### function# from CWA get Taiwan's warning typhoon list# the result is a json defget_taiwan_warning_typhon_data():url='https://rdc28.cwa.gov.tw/TDB/public/warning_typhoon_list/get_warning_typhoon'headers={'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Mobile Safari/537.36','Content-Type':'application/x-www-form-urlencoded; charset=UTF-8','X-Requested-With':'XMLHttpRequest'}response=requests.post(url,headers=headers,data={'year':'all'})ifresponse.status_code==200:decoded_content=response.content.decode('utf-8-sig')# decode by 'utf-8-sig'data=json.loads(decoded_content)return(data)else:raiseraise_error_type.ResponseError(f'request failed with status code {response.status_code}')# from CWA get northwest Pacific's typhoon list# the result is a json defget_northwest_pacific_typhon_data():url='https://rdc28.cwa.gov.tw/TDB/public/typhoon_list/get_typhoon'headers={'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Mobile Safari/537.36','Content-Type':'application/x-www-form-urlencoded; charset=UTF-8','X-Requested-With':'XMLHttpRequest'}response=requests.post(url,headers=headers,data={'year':'all'})ifresponse.status_code==200:decoded_content=response.content.decode('utf-8-sig')# decode by 'utf-8-sig'data=json.loads(decoded_content)return(data)else:raiseraise_error_type.ResponseError(f'request failed with status code {response.status_code}')
下載與分類
完成前述兩項資料下載函式後,即可開始進行資料的批次下載與分類。
get_newest_taiwan_warning_typhon() 會先檢查本機是否已存在當日下載的颱風警報資料,若存在則直接讀取;若不存在,便呼叫前述 API 下載最新資料,並轉存為 CSV 檔,以減少重複下載相同資料所造成的時間與網路資源浪費。
### import moduleimportosimportdatetimeimportpandasaspdimportraise_error_typefromtqdmimporttqdmimportshutil# move file### functiondefget_newest_taiwan_warning_typhon():# it will return a csv file pathfilename='warning_typhon_list_'+f'{datetime.datetime.now().strftime("%Y-%m-%d")}'+'.csv'#filename = f'{datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")}' + '.csv'# check whether the newest warning typhoon list existsifos.path.exists(filename):print(f'The newest Taiwan\'s warning typhoon list has exist at \"{filename}\".')returnfilenameelse:print(f'Did not find the newest Taiwan\'s warning typhoon list, using crawler method and downloading.')# download data form CWA by using crawler methodwarning_typhon=get_taiwan_warning_typhon_data()# it is a json filewarning_typhon=pd.DataFrame(warning_typhon)# save the list to filewarning_typhon.to_csv(filename,index=False,encoding='utf-8')print(f'The newest Taiwan\'s warning typhoon list has save at \"{filename}\".')returnfilenamedefdownload_satellite_image():# generate date rangestart_date='2026-07-09'end_date=datetime.datetime.today()date_range=pd.date_range(start=start_date,end=end_date,freq='min')# download imagefordateindate_range:year=str(date.year)month=format(date.month)day=format(date.day)hour=format(date.hour)minute=format(date.minute)try:get_taiwan_water_vapor_color_satellite_image(year=year,month=month,day=day,hour=hour,minute=minute)exceptraise_error_type.ResponseErroraserror:print(f'Skipped as raise_error_type.ResponseError:{error}')defformat(i):returnstr(i)ifi>9elsestr(f'0{i}')### main program## warning tyhoon listpath=get_newest_taiwan_warning_typhon()warning_typhoon=pd.read_csv(path,encoding='utf-8')warning_typhoon.dropna(subset=['max_intensity','sea_start_datetime','sea_end_datetime'],inplace=True)## satellite image downloaddownload_satellite_image()# make classify foldersfordirin['強烈颱風','中度颱風','輕度颱風','無颱風']:ifnotos.path.exists(dir):os.makedirs(dir)# define a mapping of intensity to folder namesintensity_folders={'s':'強烈颱風','m':'中度颱風','w':'輕度颱風'}# classify imagesforimgintqdm(os.listdir('water_vapor_color_satellite_image')):# classify from each imageimg_path=os.path.join('water_vapor_color_satellite_image',img)ifos.path.isfile(img_path):# check if it is a fileimg_time=datetime.datetime.strptime(img[0:13],'%Y%m%d_%H%M')# change str to timeforrowinwarning_typhoon.itertuples():sea_start=datetime.datetime.strptime(row.sea_start_datetime,'%Y-%m-%d %H:%M:%S')sea_end=datetime.datetime.strptime(row.sea_end_datetime,'%Y-%m-%d %H:%M:%S')intensity=row.max_intensityifsea_start<=img_time<=sea_end:# determine the folder based on intensityfolder_name=intensity_folders.get(intensity)new_img_path=os.path.join(folder_name,img)shutil.move(img_path,new_img_path)# move the image to the new folderbreakelse:new_img_path=os.path.join('無颱風',img)shutil.move(img_path,new_img_path)breakprint('Successful download and classify all images.')input('Press any key to continue...')