[미니 프로젝트] OpenWeather API를 활용한 날씨 DashBoard 개발

목적

원하는 도시의 기온 데이터를 1시간 동안 10분 간격으로 받아와 시각화하는 기능을 개발하고자 합니다.

[ 전반적인 코드 순서 ]

API를 활용해 원하는 도시의 데이터를 10분 간격으로 6번 가져옴.
- API로 데이터를 받아오는 코드를 6번 반복
- time.sleep()을 활용하여 코드 실행을 지연
- json 형태로 받아온 데이터를 리스트에 추가
1번 과정에서 받아온 리스트에서 데이터를 하나씩 꺼내어 원하는 데이터를 추출
- 기온 , 도시이름 데이터 추출
- 분(minute) 데이터는 idx * 10 으로 10분 간격임을 나타냄
추출된 데이터 저장 및 시각화
- txt 형태로 저장
- plotly를 활용하여 시각화( x 축 → 분(mintue) , y축 → temperature(기온)

1. API 를 활용해 원하는 도시의 데이터 받아오기

API를 활용해 원하는 도시의 데이터를 받아와 리스트에 추가합니다.
1시간 동안 총 10분 간격으로 데이터를 받아오므로 위 코드를 총 6번 반복했습니다.
10분 동안 코드 실행을 지연시키는 방법을 활용했습니다.→ time.sleep(600)

import requests
import schedule
import time 

class DataFetcher:
    def __init__(self, api_key , city):
        self.api_key = api_key
        self.city = city
        self.data_list = []

    def fetch_weather(self, city):

        url = f"http://api.openweathermap.org/data/2.5/weather?q={self.city}&appid={self.api_key}&units=metric"
        response = requests.get(url)
            
        if response.status_code == 200:
            data = response.json()
            self.data_list.append(data)
            print('데이터가 추가되었습니다.')
        
        else:
            print("error")
            return None
        
    
    def update_weather_periodically(self) :
        
        # data list에 data를 추가하는 코드를 6번 실행
        
        for i in range(6) :
            
            self.fetch_weather(self.city) 
            
            if i < 5 :
                time.sleep(600) #10분동안 코드 실행 지연 
            else : 
                break #마지막일 경우 바로 종료 
        
        return self.data_list

이때 API는 OpenWeather API를 활용했습니다.

https://openweathermap.org/api

Weather API - OpenWeatherMap

openweathermap.org

2. 원하는 데이터를 추출하기 위해 정제하기

원하는 데이터 : 도시, 기온 , 분(시간)
1번 과정에서 받아온 데이터 리스트에서 원하는 데이터를 추출했습니다.

class DataProcessor:
    def __init__(self):
        self.time_list = []
        self.temp_list = []

    def process_data(self, data_list) : 
        
        for idx, data in enumerate(data_list) : 
            
            temp = data['main']['temp']
            self.temp_list.append(temp)
            self.time_list.append(f'{idx*10} min')
        
        city_name = data_list[0]['name']    
        
        #그래프 x축 y축 
        x = self.time_list
        y = self.temp_list 
        
        return city_name, x , y

3. 데이터 저장 (txt 활용)

2번 과정에서 return 되는 데이터를 txt로 저장했습니다.

class Logger:
    
    def __init__(self, filename = 'log.txt'):
       self.filename = filename
    
    def save_data(self, data) : 
        with open(self.filename , 'w') as f :
            city = data[0]
            for time , temp in zip(data[1] , data[2]) : 
                f.write(f'City : {city} , Time : {time} , Temperature : {temp}\n')

4. plotly를 활용하여 시각화

2번 과정에서 받아온 분(시간)리스트와 기온리스트를 활용하여 시각화 하였습니다.
x 축은 분(minute) , y축은 기온(temperature)로 나타내었습니다.

import plotly.graph_objects as go

class Visualizer:
    def plot_data(self, data, output_file='output.png'):

        city = data[0] #도시
        time = data[1] #시간 
        temperature = data[2] #기온 
        
        fig = go.Figure()
        fig.add_trace(go.Scatter(x = time , y=temperature, mode='lines+markers', name='Temperature'))
        
        fig.update_layout(title=f'{city}: Temperature Over Time(1h)',
                          xaxis_title='Time',
                          yaxis_title='Temperature (°C)',
                          template='plotly_white',
                          font_family = 'Times New Roman')

        fig.write_image(output_file)
        print(f"Graph saved to {output_file}")

5. main.py를 통해서 나머지 모듈들을 불러와서 한번에 실행하는 기능

from data_fetcher import DataFetcher
from data_processor import DataProcessor
from visualizer import Visualizer
from logger import Logger

def main():
    api_key = # openweathermap API 키 입력
    city = input("Which City? ")
    
    fetcher = DataFetcher(api_key,city)
    processor = DataProcessor()
    visualizer = Visualizer()
    logger = Logger()
    
    #10분 간격으로 저장된 데이터 리스트 리턴  
    raw_data = fetcher.update_weather_periodically()
    #원하는 데이터 추출 (기온 , 도시이름 )  
    processed_data = processor.process_data(raw_data)
    #추출한 데이터 txt 파일로 저장 
    logger.save_data(processed_data)
    #시각화 
    visualizer.plot_data(processed_data)

if __name__ == "__main__":
    main()