Last active
July 8, 2023 16:27
-
-
Save jeffehobbs/28206bfe2ca1750766bed8c3e06f2d59 to your computer and use it in GitHub Desktop.
AWS lambda wrapper for WeatherKit API + OpenAI Summarization
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment