Hugging Face's logo

Spaces:
sahandv
/
SoilWaterBalanceAnalyser
Sleeping

App Files Community
SoilWaterBalanceAnalyser / scripts /display.py
Sahand
initial commit
6b83428
raw
history blame
21 kB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
import datetime
import logging
import os
import traceback
from typing import Dict, List
import numpy as np
import plotly.express as px
import rasterio
from dash import Dash, Input, Output, State, dcc, html
from dash.exceptions import PreventUpdate
from libs.utils import setup_logging
from libs.utils import verbose as vprint
from scripts.analyse import analyse
setup_logging()
log = logging.getLogger(__name__)
CONFIG = {}
V = 1
V_IGNORE = [] # Debug, Warning, Error
# ===============================================================================
# Soil Moisture Comparison Tool App Layout
# ===============================================================================
colorscales = px.colors.named_colorscales()
# external JavaScript files
external_scripts = [
"https://www.google-analytics.com/analytics.js",
{"src": "https://cdn.polyfill.io/v2/polyfill.min.js"},
{
"src": "https://cdnjs.cloudflare.com/ajax/libs/lodash.js/4.17.10/lodash.core.js",
"integrity": "sha256-Qqd/EfdABZUcAxjOkMi8eGEivtdTkh3b65xCZL4qAQA=",
"crossorigin": "anonymous",
},
]
# external CSS stylesheets
external_stylesheets = [
"https://codepen.io/chriddyp/pen/bWLwgP.css",
{
"href": "https://stackpath.bootstrapcdn.com/bootstrap/4.1.3/css/bootstrap.min.css",
"rel": "stylesheet",
"integrity": "sha384-MCw98/SFnGE8fJT3GXwEOngsV7Zt27NXFoaoApmYm81iuXoPkFOJwJ8ERdknLPMO",
"crossorigin": "anonymous",
},
]
app = Dash(
__name__,
external_scripts=external_scripts,
external_stylesheets=external_stylesheets,
title="Soil Moisture Comparison Tool",
update_title="Loading the tool...",
)
# farm_name = "Arawa"
# layer = "SM2"
today = datetime.datetime.today()
time_delta = datetime.timedelta(days=20)
FAIL_IMAGE = app.get_asset_url("icons/fail.png")
SUCCESS_IMAGE = app.get_asset_url("icons/success.png")
WAIT_IMAGE = app.get_asset_url("icons/wait.png")
current_working_directory = os.getcwd()
app.index_template = os.path.join(current_working_directory, "templates", "index.html")
# app.index_string = """
# <!DOCTYPE html>
# <html>
# <head>
# {%metas%}
# <title>{%title%}</title>
# {%favicon%}
# {%css%}
# </head>
# <body>
# <div class="col-12">
# <br>
# <h2>Soil Moisture Comparison Tool</h2>
# <br>
# <hr>
# </div>
# {%app_entry%}
# <footer>
# {%config%}
# {%scripts%}
# {%renderer%}
# </footer>
# <div class="col-12">
# <hr>
# <br>
# Copyright @ 2023 Sydney Informatics Hub (SIH)
# <br>
# </div>
# </body>
# </html>
# """
app.layout = html.Div(
[
# html.Div(
# className="app-header",
# children=[
# html.Div('Soil Moisture Comparison Tool', className="app-header--title")
# ]
# ),
dcc.Store(id="farm-name-session", storage_type="session"),
html.Div(
[
html.P(
"""This tool will use the produced datacubes to compare the soil moisture of a farm against historic data.
Please select the desired comaprison method and dates to make the comparison as in section A.
Then choose the visualisation in section B to see the results.""",
style={"font-size": "larger"},
),
html.Hr(),
html.H3("A"),
],
className="col-lg-12",
style={"padding-top": "1%", "padding-left": "1%"},
),
html.Div(
[
html.Div(
[
# html.P("Write farm name/ID:"),
dcc.Input(
id="farm-name",
type="text",
placeholder="Farm name",
style={"width": "80%"},
),
html.Img(
id="farm-image",
src=WAIT_IMAGE,
style={"width": "30px", "margin-left": "15px"},
),
],
className="col-lg-5",
# style = {'padding-top':'1%', 'padding-left':'1%'}
),
html.Div(
[
html.P(),
],
className="col-lg-7",
# style = {'padding-top':'1%', 'padding-left':'1%'}
),
],
className="row",
style={"padding-top": "1%", "padding-left": "1%"},
),
html.Div(
[
html.Div(
[
html.P("Select soil layer:"),
dcc.Dropdown(
id="layer-dropdown",
options=[
{"label": "SM1", "value": "SM1"},
{"label": "SM2", "value": "SM2"},
{"label": "SM3", "value": "SM3"},
{"label": "SM4", "value": "SM4"},
{"label": "SM5", "value": "SM5"},
{"label": "DD", "value": "DD"},
],
value="SM2",
),
],
className="col-lg-4",
style={"padding": "1%"},
),
html.Div(
[
html.P("Select the historic years to compare against:"),
dcc.Dropdown(
id="historic-dropdown",
options=[
{"label": year, "value": year} for year in range(1, 20)
],
value=2,
),
],
className="col-lg-4",
style={"padding": "1%"},
),
html.Div(
[
html.P("Select the most recent window of dates to analyse:"),
dcc.DatePickerRange(
id="window-select",
min_date_allowed=datetime.date(2000, 1, 1),
max_date_allowed=today.strftime("%Y-%m-%d"),
initial_visible_month=datetime.date(2023, 1, 1),
clearable=False,
display_format="YYYY-MM-DD",
start_date_placeholder_text="Start date",
end_date_placeholder_text="End date",
style={"width": "100%"},
),
],
className="col-lg-4",
style={"padding": "1%"},
),
],
className="row",
style={"padding-top": "1%"},
),
html.Div(
[
html.Div(
[
html.P("Select window aggregation method:"),
dcc.Dropdown(
id="w-aggregation-dropdown",
options=[
{"label": "Mean", "value": "mean"},
{"label": "Median", "value": "median"},
{"label": "Max", "value": "max"},
{"label": "Min", "value": "min"},
{"label": "Sum", "value": "sum"},
{"label": "std", "value": "std"},
{"label": "var", "value": "var"},
],
value="mean",
),
],
className="col-lg-6",
style={"padding": "1%"},
),
html.Div(
[
html.P("Select historic aggregation method:"),
dcc.Dropdown(
id="h-aggregation-dropdown",
options=[
{"label": "Mean", "value": "mean"},
{"label": "Median", "value": "median"},
{"label": "Max", "value": "max"},
{"label": "Min", "value": "min"},
{"label": "Sum", "value": "sum"},
{"label": "std", "value": "std"},
{"label": "var", "value": "var"},
],
value="mean",
),
],
className="col-lg-6",
style={"padding": "1%"},
),
],
className="row",
# style = {'padding-top':'1%'}
),
html.Div(
[
html.Button("Generate Images", id="generate-button"),
html.Br(),
html.Hr(),
],
className="col-lg-12",
style={"margin-bottom": "1%"},
),
html.Div(
[
html.H3("B"),
],
className="col-lg-12",
style={"padding-top": "1%", "padding-left": "1%"},
),
html.Div(
[
html.Div(
[
html.P("Select visualisation name:"),
dcc.Dropdown(id="visualisation-select"),
],
className="col-lg-6",
style={"padding": "1%"},
),
html.Div(
[
html.P("Select your palette:"),
dcc.Dropdown(
id="platter-dropdown", options=colorscales, value="viridis"
),
],
className="col-lg-6",
style={"padding": "1%"},
),
],
className="row",
# style = {'padding-top':'1%'}
),
html.Div(
[
html.Hr(),
html.H3("Results"),
dcc.Graph(id="graph"),
],
className="col-lg-12",
style={"padding-top": "1%"},
),
# html.Div(
# className="app-footer",
# children=[
# html.Div(f"Copyright @ {today.strftime('%Y')} Sydney Informatics Hub (SIH)", className="app-footer--copyright")
# ]
# ),
],
className="container-fluid",
)
# ==================================================================================================
# Functions
# ==================================================================================================
def find_analyses(path):
"""Find all the analysis files in a directory.
Parameters
----------
path: str
Path to the directory containing the analysis files
Returns
-------
files: list
List of analysis files
"""
files = [f for f in os.listdir(path) if f.endswith(".tif")]
return files
def open_image(path):
"""Open a raster image and return the data and coordinates.
Parameters
----------
path: str
path to the raster image
Returns
-------
band1: np.array
The raster data
lons: np.array
The longitude coordinates
lats: np.array
The latitude coordinates
"""
with rasterio.open(path) as src:
band1 = src.read(1)
print("Band1 has shape", band1.shape)
height = band1.shape[0]
width = band1.shape[1]
cols, rows = np.meshgrid(np.arange(width), np.arange(height))
xs, ys = rasterio.transform.xy(src.transform, rows, cols)
lons = np.array(xs)
lats = np.array(ys)
return band1, lons, lats
def perform_analysis(
input,
window_start,
window_end,
historic_years: int,
layer: str,
match_raster: str = None,
output: str = None,
agg_history: str = "mean",
agg_window: str = "mean",
comparison: str = "diff",
**args,
) -> Dict[str, str]:
"""Perform the analysis.
This is a wrapper function for the analysis module. It takes the input parameters and passes them to the analysis module.
Parameters
----------
input : str
path to the input data
window_start : str
start date of the window
window_end : str
end date of the window
historic_years : int
number of years to use for the historic data
layer : str
layer to use for the analysis
match_raster : str, optional
path to the raster to match the output to, by default None
output : str, optional
path to the output file, by default None
agg_history : str, optional
aggregation method for the historic data, by default "mean"
agg_window : str, optional
aggregation method for the window data, by default "mean"
comparison : str, optional
comparison method for the window and historic data, by default "diff"
Returns
-------
files: dict
Dict of analysis files
"""
files = analyse(
input=input,
window_start=window_start,
window_end=window_end,
historic_years=historic_years,
agg_window=agg_window,
agg_history=agg_history,
comparison=comparison,
layer=layer,
output=output,
match_raster=match_raster,
)
return files
# ====================================================================================================
# Callbacks
# ====================================================================================================
@app.callback(
[
Output("farm-name-session", "data"),
Output("farm-image", "src"),
],
[Input("farm-name", "value"), State("farm-name-session", "data")],
)
def update_session(farm_name, session):
session = farm_name
if farm_name is None or farm_name == "":
session = ""
image = WAIT_IMAGE
else:
print(f"Getting some data about farm: {farm_name}")
# if the path does not exist, do not update the session
real_path = INPUT.format(farm_name)
print(f"Checking {real_path}")
if os.path.exists(real_path):
session = farm_name
image = SUCCESS_IMAGE
else:
session = ""
image = FAIL_IMAGE
print(f"\n\nSession updated to {session}")
print(f"Image updated to {image}\n\n")
return session, image
@app.callback(
Output("farm-name", "value"),
Input("farm-name-session", "modified_timestamp"),
State("farm-name-session", "data"),
)
def display_name_from_session(timestamp, name):
print(f"Updating the farm name from the session: {name}")
if timestamp is not None:
return name
else:
return ""
@app.callback(
Output("visualisation-select", "options"),
# Input("farm-name", "value"),
Input("layer-dropdown", "value"),
Input("window-select", "start_date"),
Input("window-select", "end_date"),
Input("historic-dropdown", "value"),
Input("w-aggregation-dropdown", "value"),
Input("h-aggregation-dropdown", "value"),
Input("generate-button", "n_clicks"),
State("farm-name-session", "data"),
)
def get_analysis(
layer, window_start, window_end, historic_years, w_agg, h_agg, n_clicks, farm_name
) -> List[Dict[str, str]]:
"""Get the analysis files and return them as a list of dicts.
Parameters
----------
layer : str
layer to use for the analysis
window_start : str
start date of the window
window_end : str
end date of the window
historic_years : int
number of years to use for the historic data
w_agg : str
aggregation method for the window data
h_agg : str
aggregation method for the historic data
n_clicks : int
number of times the generate button has been clicked
Returns
-------
files : list
list of dicts of analysis files
"""
print("\nAnalysis callback triggered")
if n_clicks == 0 or n_clicks is None:
raise PreventUpdate
path = f"/home/sahand/Data/results_default/{farm_name}/soilwatermodel"
# window_start = datetime.datetime.strptime(window_start, '%Y-%m-%d')
# window_end = datetime.datetime.strptime(window_end, '%Y-%m-%d')
print(f"\nPath: {path}\n")
files = perform_analysis(
input=path,
window_start=window_start,
window_end=window_end,
historic_years=historic_years,
layer=layer,
agg_window=w_agg,
agg_history=h_agg,
comparison="diff",
output=None,
match_raster=None,
)
print(path)
print(
f"n_clicks: {n_clicks}\n"
+ f"window_start: {window_start}\n"
+ f"window_end: {window_end}\n"
+ f"historic_years: {historic_years}\n"
+ f"layer: {layer}\n"
+ f"agg_window: {w_agg}\n"
+ f"agg_history: {h_agg}\n"
+ "comparison: 'diff'\n"
+ f"output: {None}\n"
+ f"match_raster: {None}\n"
)
print(files)
files = {
i: [
" ".join(files[i].split("/")[-1].split(".")[0].split("-")).capitalize(),
files[i],
]
for i in files
}
print(files)
options = [{"label": files[i][0], "value": files[i][1]} for i in files]
return options
@app.callback(
Output("graph", "figure"),
Input("visualisation-select", "value"),
Input("platter-dropdown", "value"),
)
def change_colorscale(file, palette):
"""Display the selected visualisation and change the colorscale of the
visualisation.
Parameters
----------
file : str
path to the visualisation file
palette : str
name of the colorscale to use
Returns
-------
fig : plotly.graph_objects.Figure
plotly figure object
"""
band1, lons_a, lats_a = open_image(file)
# Get the second dimension of the lons
lats = lats_a[:, 0]
lons = lons_a[0, :]
print(lons.shape, lons)
print(lats.shape, lats)
print(band1.shape, band1)
fig = px.imshow(band1, x=lons, y=lats, color_continuous_scale=palette)
fig.update(
data=[
{
"customdata": np.stack((band1, lats_a, lons_a), axis=-1),
"hovertemplate": "<b>SM</b>: %{customdata[0]}<br>"
+ "<b>Lat</b>: %{customdata[1]}<br>"
+ "<b>Lon</b>: %{customdata[2]}<br>"
+ "<extra></extra>",
}
]
)
print("Render successful")
return fig
# ==============================================================================
# Main
# ==============================================================================
if __name__ == "__main__":
# Load Configs
parser = argparse.ArgumentParser(
description="Download rainfall data from Google Earth Engine for a range of dates.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
"-i",
"--input",
help="Absolute or relative path to the netcdf data directory for each farm. Should be in this format: '/path/to/farm/{}/soilwatermodel'",
default=os.path.join(
os.path.expanduser("~"), "Data/results_default/{}/soilwatermodel"
),
)
args = parser.parse_args()
INPUT = args.input
try:
app.run_server(debug=True)
except Exception as e:
vprint(
0,
V,
V_IGNORE,
Error="Failed to execute the main function:",
ErrorMessage=e,
)
traceback.print_exc()
raise e