# app.py
from flask import Flask, render_template, request
import requests
import re
import csv
from datetime import datetime
app = Flask(__name__)
class METARDecoder:
def __init__(self):
self.wind_directions = {
'N': 'north', 'NNE': 'north-northeast', 'NE': 'northeast',
'ENE': 'east-northeast', 'E': 'east', 'ESE': 'east-southeast',
'SE': 'southeast', 'SSE': 'south-southeast', 'S': 'south',
'SSW': 'south-southwest', 'SW': 'southwest', 'WSW': 'west-southwest',
'W': 'west', 'WNW': 'west-northwest', 'NW': 'northwest',
'NNW': 'north-northwest'
}
def decode_wind(self, wind_token: str):
# Examples: 16008KT, 00000KT, VRB03KT, 18012G20KT
if not wind_token:
return "Wind: not reported"
if wind_token.startswith("VRB"):
m = re.match(r'VRB(\d{2,3})KT', wind_token)
if m:
return f"Variable wind at {int(m.group(1))} knots"
if wind_token == '00000KT':
return "Calm (0 knots)"
m = re.match(r'(\d{3})(\d{2,3})(G(\d{2,3}))?KT', wind_token)
if m:
deg = int(m.group(1))
speed = int(m.group(2))
gust = m.group(4)
# convert degrees to compass abbreviation (16 sectors)
dirs = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"]
idx = int((deg + 11.25) // 22.5) % 16
dir_text = self.wind_directions.get(dirs[idx], dirs[idx])
gust_text = f", gusts to {gust} knots" if gust else ""
return f"Wind: {speed} knots from the {dir_text} ({deg}°){gust_text}"
return f"Wind: {wind_token}"
def decode_temp_dew(self, tempdew: str):
# Example: 27/11 (Celsius)
m = re.match(r'([M]?\d{1,2})/([M]?\d{1,2})', tempdew)
if not m:
return "Temperature/dewpoint not reported"
def parse_value(s):
if s.startswith('M'):
return -int(s[1:])
return int(s)
temp_c = parse_value(m.group(1))
dew_c = parse_value(m.group(2))
temp_f = round((temp_c * 9/5) + 32)
dew_f = round((dew_c * 9/5) + 32)
return f"Temperature: {temp_f}°F ({temp_c}°C), Dew point: {dew_f}°F ({dew_c}°C)"
def decode_visibility(self, vis_token: str):
# Example: 10SM (10 statute miles)
if not vis_token:
return "Visibility: not reported"
m = re.match(r'(\d+)SM', vis_token)
if m:
return f"Visibility: {m.group(1)} statute miles"
return f"Visibility: {vis_token}"
def decode_clouds(self, clouds_section: str):
# Simplified handling: look for CLR, FEW, SCT, BKN, OVC and heights (in hundreds of feet)
if not clouds_section or clouds_section.strip() == '':
return "Sky: not reported"
if 'CLR' in clouds_section or 'SKC' in clouds_section:
return "Clear skies"
clouds = []
tokens = clouds_section.split()
for tok in tokens:
m = re.match(r'(FEW|SCT|BKN|OVC)(\d{3})', tok)
if m:
amount = m.group(1)
height_hundreds = int(m.group(2))
height_ft = height_hundreds * 100
desc = {
'FEW': 'a few clouds',
'SCT': 'scattered clouds',
'BKN': 'broken clouds',
'OVC': 'overcast'
}.get(amount, amount)
clouds.append(f"{desc} at {height_ft} ft")
return "Sky: " + (", ".join(clouds) if clouds else clouds_section)
def decode_altimeter(self, alt_token: str):
# Example: A3005 -> 30.05 inHg
m = re.match(r'A(\d{4})', alt_token)
if m:
inches = float(m.group(1)) / 100.0
return f"Altimeter: {inches:.2f} inHg"
return "Altimeter: not reported"
def decode_metar(self, metar_string: str):
# Very simple parser, intended for common, simple METARs.
# Split tokens and attempt to find standard fields.
tokens = metar_string.split()
decoded = {}
# Station
if tokens:
decoded['station'] = tokens[0]
# Time token (e.g., 052253Z)
for tok in tokens[1:4]:
if tok.endswith('Z') and re.match(r'\d{6}Z', tok):
decoded['time'] = tok
break
# Find wind, visibility, temp/dew, altimeter
for tok in tokens:
if re.match(r'(\d{3}|VRB)\d{2,3}(G\d{2,3})?KT', tok) or tok == '00000KT':
decoded['wind'] = self.decode_wind(tok)
elif tok.endswith('SM'):
decoded['visibility'] = self.decode_visibility(tok)
elif re.match(r'[M]?\d{1,2}/[M]?\d{1,2}', tok):
decoded['temp_dew'] = self.decode_temp_dew(tok)
elif tok.startswith('A') and re.match(r'A\d{4}', tok):
decoded['altimeter'] = self.decode_altimeter(tok)
elif tok in ['CLR', 'SKC'] or re.match(r'(FEW|SCT|BKN|OVC)\d{3}', tok):
# Gather the rest of the cloud tokens starting from here
idx = tokens.index(tok)
cloud_section = ' '.join(tokens[idx:idx+4]) # small window
decoded['clouds'] = self.decode_clouds(cloud_section)
# Raw METAR fallback fields
decoded['raw'] = metar_string
return decoded
decoder = METARDecoder()
def fetch_metar(airport_code: str):
"""
Fetch raw METAR data for the given airport code.
This example uses the [AviationWeather ADDS (Aviation Digital Data Service)](https://aviationweather.gov/dataserver)
"dataserver_current" endpoint to request METARs in CSV format and extract
the raw_text field. This is more reliable than attempting to fetch a single
raw line from a generic HTML endpoint.
Note: the ADDS server supports multiple parameters; here we request the most
recent METAR for the station within the past couple of hours. For production
use, review the ADDS API documentation and consider caching, rate limits,
and error handling improvements.
"""
base_url = "https://aviationweather.gov/adds/dataserver_current/httpparam"
params = {
'dataSource': 'metars',
'requestType': 'retrieve',
'format': 'csv',
'stationString': airport_code.upper(),
'hoursBeforeNow': '2',
'mostRecent': 'true'
}
try:
resp = requests.get(base_url, params=params, timeout=10)
resp.raise_for_status()
text = resp.text.strip()
if not text:
return None
# Parse CSV and look for raw_text column
try:
reader = csv.DictReader(text.splitlines())
for row in reader:
raw = row.get('raw_text') or row.get('rawMETAR') or None
if raw:
return raw.strip()
except Exception:
# Fallback: attempt to find a line that starts with the station ID
for line in text.splitlines():
if line.strip().upper().startswith(airport_code.upper()):
return line.strip()
# Final fallback: first non-empty line
for line in text.splitlines():
if line.strip():
return line.strip()
return None
except requests.RequestException:
return None
@app.route('/', methods=['GET'])
def index():
return render_template('index.html')
@app.route('/result', methods=['POST'])
def result():
airport = request.form.get('airport', '').strip().upper()
if not airport:
return render_template('result.html', error="No airport code provided.")
raw = fetch_metar(airport)
if not raw:
return render_template('result.html', error=f"Could not fetch METAR for {airport}.")
decoded = decoder.decode_metar(raw)
return render_template('result.html', airport_code=airport, raw_metar=raw, decoded=decoded)
if __name__ == '__main__':
app.run(debug=True)
