UK Flood Alert / Flood Data (Python)

Python code for the Flood Alert API

Today I received a phone call and SMS text message that a flood alert had become active for the area I live in. After making preparations incase the alert severity increased I got to thinking; is there a better way to be warned about potential floods?

One (quick) Google search later and I found that the Environment Agency very kindly provides API access to their data at https://environment.data.gov.uk/flood-monitoring/doc/reference#introduction. With this knowledge in hand it was time to create some Python to access the APIs, grab the data (in handy JSON format) and see what could be returned.

I’ve created several functions so far and my ultimate plan will be to use the automatic functions running regularly via a cron job on my Raspberry Pi and then outputting an alert. I may also look to regularly taking readings from my local station sensor and charting them.

—————————
# geektechstuff
# UK Flood Details
import requests
import json
def flood_user_input():
# checks flood details near by
atlantis =””
input_lat = input(‘Enter Latitude: ‘)
input_long = input(‘Enter Longitude: ‘)
distance = input(‘How close to the latitude / longitude should be seached?: ‘)
flood_parameters = ‘?lat=’+input_lat+’&long=’+input_long+’&dist=’+distance
flood_query = flood_url+flood_parameters
get_data = requests.get(flood_query)
get_data.raise_for_status()
analysis = get_data.json()
items = analysis[“items”]
river = items[0][“floodArea”][“riverOrSea”]
severity = items[0][“severity”]
# severity level 1 is severe, 2 is flood warning, 3 is flood alert and 4 is warning no longer in force
s_level = items[0][“severityLevel”]
message = items[0][“message”]
time_severity_changed = items[0][“timeSeverityChanged”]
if s_level >=3:
severity=str(severity)
s_level=str(s_level)
message=str(message)
time_severity_changed=str(time_severity_changed)
atlantis = “\nWARNING \n \n”+river+”\n \n”+”Severity level is at “+s_level+”\n \n”+severity+”\n \n”+message+”\n \n”+”Last update: “+time_severity_changed+”\n”
else:
atlantis = “All OKAY”
return(atlantis)
def auto_flood_check():
# checks flood details near by
# Replace XXXXXXX with latitude and longitude
atlantis = “”
get_data = requests.get(flood_query)
get_data.raise_for_status()
analysis = get_data.json()
items = analysis[“items”]
river = items[0][“floodArea”][“riverOrSea”]
severity = items[0][“severity”]
# severity level 1 is severe, 2 is flood warning, 3 is flood alert and 4 is warning no longer in force
s_level = items[0][“severityLevel”]
message = items[0][“message”]
time_severity_changed = items[0][“timeSeverityChanged”]
if s_level >=3:
severity=str(severity)
s_level=str(s_level)
message=str(message)
time_severity_changed=str(time_severity_changed)
atlantis = “\nWARNING \n \n”+river+”\n \n”+”Severity level is at “+s_level+”\n \n”+severity+”\n \n”+message+”\n \n”+”Last update: “+time_severity_changed+”\n”
else:
atlantis = “All OKAY”
return(atlantis)
def stations_near_user_input():
#returns details of the water stations near by
input_lat = input(‘Enter Latitude: ‘)
input_long = input(‘Enter Longitude: ‘)
distance = input(‘How close to the latitude / longitude should be seached?: ‘)
user_parameters = ‘?lat=’+input_lat+’&long=’+input_long+’&dist=’+distance
station_url = base_url+user_parameters
get_data = requests.get(station_url)
get_data.raise_for_status()
analysis = get_data.json()
items = analysis[“items”]
url_for_station = items[0][“@id”]
RLOIid = items[0][“RLOIid”]
date_opened = items[0][“dateOpened”]
whats_measured = items[0][“measures”][0][“parameterName”]
str_url_for_station = str(url_for_station)
str_station_name = str(station_name)
str_RLOIid = str(RLOIid)
str_dateOpened = str(date_opened)
str_whats_measured = str(whats_measured)
output_msg = “\n Station Name: “+str_station_name+”\n Station URL: “+str_url_for_station+”\n Station RLOIid: “+str_RLOIid+”\n Date Station Opened: “+str_dateOpened+”\n Station Measures: “+str_whats_measured+”\n”
return(output_msg)
def stations_near_auto():
# Returns details of the water stations near by
# Replace XXXXXXX with latitude and longitude
get_data = requests.get(station_url)
get_data.raise_for_status()
analysis = get_data.json()
items = analysis[“items”]
url_for_station = items[0][“@id”]
RLOIid = items[0][“RLOIid”]
date_opened = items[0][“dateOpened”]
station_name = items[0][“label”]
whats_measured = items[0][“measures”][0][“parameterName”]
str_url_for_station = str(url_for_station)
str_station_name = str(station_name)
str_RLOIid = str(RLOIid)
str_dateOpened = str(date_opened)
str_whats_measured = str(whats_measured)
output_msg = “\n Station Name: “+str_station_name+”\n Station URL: “+str_url_for_station+”\n Station RLOIid: “+str_RLOIid+”\n Date Station Opened: “+str_dateOpened+”\n Station Measures: “+str_whats_measured+”\n”
return(output_msg)
def readings_from_station_auto():
# Gives station details, current reading and historic high / low readings from station
# Replace XXXXXXX with the station @id (see url_for_station from stations_near_auto())
get_data = requests.get(station_url)
get_data.raise_for_status()
analysis = get_data.json()
analysis = get_data.json()
items = analysis[“items”]
url_for_station = items[“@id”]
str_url_for_station = str(url_for_station)
RLOIid = items[“RLOIid”]
str_RLOIid = str(RLOIid)
date_opened = items[“dateOpened”]
str_date_opened = str(date_opened)
station_name = items[“label”]
str_station_name = str(station_name)
# current measurements
current_measurement_date_time = items[“measures”][“latestReading”][“dateTime”]
str_current_measurement_date_time = str(current_measurement_date_time)
current_measurement_value = items[“measures”][“latestReading”][“value”]
str_current_measurement_value = str(current_measurement_value)
# historic measurements
max_high = items[“stageScale”][“maxOnRecord”][“value”]
str_max_high = str(max_high)
max_high_date = items[“stageScale”][“maxOnRecord”][“dateTime”]
str_max_high_date = str(max_high_date)
min_low = items[“stageScale”][“minOnRecord”][“value”]
str_min_low = str(min_low)
min_low_date = items[“stageScale”][“minOnRecord”][“dateTime”]
str_min_low_date = str(min_low_date)
output_msg = “\n Station Name: “+str_station_name+”\n RLOIid: “+str_RLOIid+”\n Date Opened: “+str_date_opened+”\n Current Measurement: “+str_current_measurement_value+”\n Measurement Taken at: “+str_current_measurement_date_time+”\n Historic Hight: “+str_max_high+”\n Max High Date: “+str_max_high_date+”\n Min Height: “+str_min_low+”\n Minimum Height Date: “+str_min_low_date+”\n”
return(output_msg)

——————————

Boxing Day (26-Dec-2015) saw flooding in my home city of Salford
Boxing Day (26-Dec-2015) saw flooding in my home city of Salford

I have posted my complete Python coding on my GitHub at https://github.com/geektechdude/Python_UK_Flood_Alert

geektechstuff_river_flood_1
Python showing flood alert

If you are in the Manchester (UK) area and want to keep up to date on the March 2019 flooding has caused then see the Manchester Evening News page: https://www.manchestereveningnews.co.uk/news/greater-manchester-news/manchester-weather-rain-flooding-live-15983239

Leave a Reply

Please log in using one of these methods to post your comment:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

This site uses Akismet to reduce spam. Learn how your comment data is processed.