AWS lambda wrapper for WeatherKit API + OpenAI Summarization

app.py

# generates a REST lambda for WeatherKit data & OpenAI summarization

import requests, json, os, boto3, configparser
from flask import Flask, Response, request, json, jsonify
import openai

app = Flask(__name__)

# set up API keys from external config secrets.txt configparser file
SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__))
config = configparser.ConfigParser()
config.read(SCRIPT_PATH +'/secrets.txt')
openai.api_key = config.get('secrets', 'openai_apikey')
TOKEN = config.get('secrets', 'weatherkit_token')
DEFAULT_LAT = config.get('secrets', 'default_lat')
DEFAULT_LNG = config.get('secrets', 'default_lng')

# globals
TIMEZONE = 'GMT-5'
COUNTRY_CODE = 'US'
LANGUAGE = 'en'
DATASETS = 'currentWeather,forecastDaily,forecastHourly'
DEFAULT_HOURS = 4

# for testing
def index():
    return 'hello', 200

# main route
@app.route('/weather') 
def get_weather_data():
    if request.args.get('lat'):
        lat = request.args.get('lat')
    else:
        lat = DEFAULT_LAT
    if request.args.get('lng'):
        lng = request.args.get('lng')
    else:
        lng = DEFAULT_LNG
    if request.args.get('summary_hours'):
        summary_hours = int(request.args.get('summary_hours'))
    else:
        summary_hours = DEFAULT_HOURS
    url = f'https://weatherkit.apple.com/api/v1/weather/{LANGUAGE}/{lat}/{lng}'
    params = {'countryCode': COUNTRY_CODE, 'dataSets': DATASETS, 'timezone': TIMEZONE}
    headers = {'Authorization': TOKEN}
    r = requests.get(url, params=params, headers=headers)
    data = r.json()
    #print(json.dumps(data, indent=4))
    if summary_hours:
        data['summary'] = summarize(data, summary_hours)
    return data

# for testing available datasets
@app.route('/availability')
def get_availability_data():
    lat = request.args.get('lat')
    lng = request.args.get('lng')
    url = f'https://weatherkit.apple.com/api/v1/availability/{lat}/{lng}'
    params = {'countryCode': COUNTRY_CODE}
    headers = {'Authorization': TOKEN}
    r = requests.get(url, params=params, headers=headers)
    return r.text

# call OpenAI for condition summarization
def summarize(data, hours):
    prompt = f'Given that for the next {hours} hours, it will be "'
    for idx, hour in enumerate(data['forecastHourly']['hours']):
        if idx == hours:
            break
        prompt = prompt + humanize_conditionCode(hour['conditionCode']) + ', '
        
    prompt = prompt[:-2] + '", provide a concise, one-sentence summary of what that weather will be.'
    print("Q: " + prompt)
    try:
        completion = openai.ChatCompletion.create(model="gpt-4", messages=[{"role": "user", "content": prompt }])
        response = completion.choices[0].message.content
    except:
        response = ''
    print("A: " + response)
    return(response)

# humanize WeatherKit conditionCode strings
def humanize_conditionCode(conditionCode):
    if conditionCode == "Clear":
        condition = 'clear'
    elif conditionCode == "MostlyClear":
        condition = 'mostly clear'
    elif conditionCode == "PartlyCloudy":
        condition = 'partly cloudy'
    elif conditionCode == "MostlyCloudy":
        condition = 'mostly cloudy'
    elif conditionCode == "Cloudy":
        condition = 'cloudy'
    elif conditionCode == "Hazy":
        condition = 'hazy'
    elif conditionCode == "ScatteredThunderstorms":
        condition = 'scattered thunderstorms'
    elif conditionCode == "Drizzle":
        condition = 'drizzle'
    elif conditionCode == "Rain":
        condition = 'rain'
    elif conditionCode == "HeavyRain":
        condition = 'heavy rain'
    else:
        condition = conditionCode
    return conditionCode

# main function
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)