mirror of
				https://gitlab.com/ceda_ei/sky
				synced 2025-10-31 10:30:04 +01:00 
			
		
		
		
	[Refactor] Better path parsing. Clean up codebase.
This commit is contained in:
		
							
								
								
									
										97
									
								
								server.py
									
									
									
									
									
								
							
							
						
						
									
										97
									
								
								server.py
									
									
									
									
									
								
							| @@ -1,23 +1,27 @@ | |||||||
| #!/usr/bin/env python3 | #!/usr/bin/env python3 | ||||||
|  | "Sky - A simple weather monitor" | ||||||
|  |  | ||||||
|  | import re | ||||||
|  | import sqlite3 | ||||||
|  | from datetime import datetime | ||||||
|  | from textwrap import wrap | ||||||
|  |  | ||||||
|  | import darksky | ||||||
| from flask import Flask | from flask import Flask | ||||||
| from mapbox import Geocoder | from mapbox import Geocoder | ||||||
| import darksky | from pytz import timezone | ||||||
|  |  | ||||||
|  | import config | ||||||
| from columnizer import Columnizer | from columnizer import Columnizer | ||||||
| from ascii import arts | from ascii import arts | ||||||
| import sqlite3 |  | ||||||
| import config |  | ||||||
| from datetime import datetime |  | ||||||
| from pytz import timezone |  | ||||||
| from textwrap import wrap |  | ||||||
| import re |  | ||||||
|  |  | ||||||
| geocoder = Geocoder(access_token=config.mapbox_key) | GEOCODER = Geocoder(access_token=config.mapbox_key) | ||||||
| app = Flask(__name__) | APP = Flask(__name__) | ||||||
| cache_db = sqlite3.connect("cache.sqlite", check_same_thread=False) | CACHE_DB = sqlite3.connect("cache.sqlite", check_same_thread=False) | ||||||
|  |  | ||||||
|  |  | ||||||
| def get_coordinates(location, geocoder, cache_db): | def get_coordinates(location, geocoder, cache_db): | ||||||
|  |     "Returns the coordinates for a given location" | ||||||
|     row = cache_db.execute("select lat,lon,full_name from location where " |     row = cache_db.execute("select lat,lon,full_name from location where " | ||||||
|                            "name = ?", (location,)).fetchone() |                            "name = ?", (location,)).fetchone() | ||||||
|  |  | ||||||
| @@ -33,11 +37,14 @@ def get_coordinates(location, geocoder, cache_db): | |||||||
|  |  | ||||||
|  |  | ||||||
| def get_weather(coordinates): | def get_weather(coordinates): | ||||||
|  |     "Returns the forecast from the coordinates" | ||||||
|     return darksky.forecast(config.dark_sky_key, coordinates[0], |     return darksky.forecast(config.dark_sky_key, coordinates[0], | ||||||
|                             coordinates[1]) |                             coordinates[1]) | ||||||
|  |  | ||||||
|  |  | ||||||
| def summary_to_c(summary): | def summary_to_c(summary): | ||||||
|  |     "Converts Fahrenheits to Celsius in a string" | ||||||
|  |     # pylint: disable=invalid-name | ||||||
|     for i in re.findall(r"[\d.]+°F", summary): |     for i in re.findall(r"[\d.]+°F", summary): | ||||||
|         f = float(i[:-2]) |         f = float(i[:-2]) | ||||||
|         c = 5 * (f - 32)/9 |         c = 5 * (f - 32)/9 | ||||||
| @@ -46,6 +53,7 @@ def summary_to_c(summary): | |||||||
|  |  | ||||||
|  |  | ||||||
| def weather_to_text(forecast, kind, unit, name): | def weather_to_text(forecast, kind, unit, name): | ||||||
|  |     "Converts the forecast to text" | ||||||
|     text = name + "\n" |     text = name + "\n" | ||||||
|     if unit == "c": |     if unit == "c": | ||||||
|         text += summary_to_c(forecast[kind]["summary"]) + "\n\n" |         text += summary_to_c(forecast[kind]["summary"]) + "\n\n" | ||||||
| @@ -138,8 +146,9 @@ def weather_to_text(forecast, kind, unit, name): | |||||||
|     return text |     return text | ||||||
|  |  | ||||||
|  |  | ||||||
| @app.route('/') | @APP.route('/') | ||||||
| def index(): | def index(): | ||||||
|  |     "/ - Gives the default index page" | ||||||
|     text = ("\nSky - A simple weather monitor\n\n" |     text = ("\nSky - A simple weather monitor\n\n" | ||||||
|             "Usage\n\n" |             "Usage\n\n" | ||||||
|             "To check weather for your location simply run the following " |             "To check weather for your location simply run the following " | ||||||
| @@ -157,6 +166,7 @@ def index(): | |||||||
|  |  | ||||||
|  |  | ||||||
| def main(location, geocoder, cache_db, kind, unit): | def main(location, geocoder, cache_db, kind, unit): | ||||||
|  |     "Main function that returns the weather based on parameters" | ||||||
|     try: |     try: | ||||||
|         coordinates = get_coordinates(location, geocoder, cache_db) |         coordinates = get_coordinates(location, geocoder, cache_db) | ||||||
|     except KeyError: |     except KeyError: | ||||||
| @@ -168,53 +178,34 @@ def main(location, geocoder, cache_db, kind, unit): | |||||||
|  |  | ||||||
|  |  | ||||||
| def strip_colors(text): | def strip_colors(text): | ||||||
|  |     "Strips ANSI colors from a string" | ||||||
|     return re.sub('\033\\[[0-9;]+?m', '', text) |     return re.sub('\033\\[[0-9;]+?m', '', text) | ||||||
|  |  | ||||||
|  |  | ||||||
| app.add_url_rule('/<location>', 'location', lambda location: | @APP.route('/<path:path>') | ||||||
|                  main(location, geocoder, cache_db, "daily", "c")) | def url_parser(path): | ||||||
|  |     "Parses the URL" | ||||||
| app.add_url_rule('/<location>/', 'location/', lambda location: |     location = '' | ||||||
|                  main(location, geocoder, cache_db, "daily", "c")) |     unit = 'c' | ||||||
|  |     plain = False | ||||||
| app.add_url_rule('/<location>/t', 'today_location', lambda location: |     frequency = 'daily' | ||||||
|                  main(location, geocoder, cache_db, "hourly", "c")) |     for part in path.strip('/').split('/'): | ||||||
|  |         if part == 'f': | ||||||
| app.add_url_rule('/f/<location>', 'location_f', lambda location: |             unit = 'f' | ||||||
|                  main(location, geocoder, cache_db, "daily", "f")) |         elif part == 'p': | ||||||
|  |             plain = True | ||||||
| app.add_url_rule('/f/<location>/', 'location_f/', lambda location: |         elif part == 't': | ||||||
|                  main(location, geocoder, cache_db, "daily", "f")) |             frequency = 'hourly' | ||||||
|  |         else: | ||||||
| app.add_url_rule('/f/<location>/t', 'today_location_f', lambda location: |             location = part | ||||||
|                  main(location, geocoder, cache_db, "hourly", "f")) |     if plain: | ||||||
|  |         return strip_colors(main(location, GEOCODER, CACHE_DB, frequency, | ||||||
| app.add_url_rule('/p/<location>', 'plain_location', lambda location: |                                  unit)) | ||||||
|                  strip_colors(main(location, geocoder, cache_db, "daily", |     return main(location, GEOCODER, CACHE_DB, frequency, unit) | ||||||
|                               "c"))) |  | ||||||
|  |  | ||||||
| app.add_url_rule('/p/<location>/', 'plain_location/', lambda location: |  | ||||||
|                  strip_colors(main(location, geocoder, cache_db, "daily", |  | ||||||
|                               "c"))) |  | ||||||
|  |  | ||||||
| app.add_url_rule('/p/<location>/t', 'plain_today_location', lambda |  | ||||||
|                  location: strip_colors(main(location, geocoder, cache_db, |  | ||||||
|                                         "hourly", "c"))) |  | ||||||
|  |  | ||||||
| app.add_url_rule('/p/f/<location>', 'plain_location_f', lambda location: |  | ||||||
|                  strip_colors(main(location, geocoder, cache_db, "daily", |  | ||||||
|                               "f"))) |  | ||||||
|  |  | ||||||
| app.add_url_rule('/p/f/<location>/', 'plain_location_f/', lambda location: |  | ||||||
|                  strip_colors(main(location, geocoder, cache_db, "daily", |  | ||||||
|                               "f"))) |  | ||||||
|  |  | ||||||
| app.add_url_rule('/p/f/<location>/t', 'plain_today_location_f', lambda |  | ||||||
|                  location: strip_colors(main(location, geocoder, cache_db, |  | ||||||
|                                         "hourly", "f"))) |  | ||||||
|  |  | ||||||
|  |  | ||||||
| @app.after_request | @APP.after_request | ||||||
| def after(response): | def after(response): | ||||||
|  |     "Sets content type to text/plain" | ||||||
|     response.headers['Content-Type'] = "text/plain; charset=utf-8" |     response.headers['Content-Type'] = "text/plain; charset=utf-8" | ||||||
|     return response |     return response | ||||||
|   | |||||||
		Reference in New Issue
	
	Block a user