app.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
import datetime
import logging
import os
import traceback
from typing import Dict, List

# import dash_html_components as html
# import flask
import numpy as np
import plotly.express as px
from dash import Dash, Input, Output, State  # , dcc  # ,html
from dash.exceptions import PreventUpdate
from flask import Flask, render_template

from libs.dashapp import layout, open_image, perform_analysis
from libs.utils import setup_logging
from libs.utils import verbose as vprint

setup_logging()
log = logging.getLogger(__name__)
CONFIG = {}
V = 1
V_IGNORE = []  # Debug, Warning, Error
MODEL_PATH = ""
FARM_NAME = ""

# ===============================================================================
# The Main App
# ===============================================================================
app = Flask(__name__)


@app.route("/")
def home():
    return render_template(
        "index.html"
    )  # "Welcome to Nowcasting App.<br><br>The dashapp is still under development..."


@app.route("/register")
def register():
    return "Welcome to Nowcasting signup page.<br><br>The dashapp is still under development..."


@app.route("/addjob")
def addjob():
    return "Welcome to Nowcasting signup page.<br><br>The dashapp is still under development..."


@app.route("/dashboard")
def dashboard():
    return "Welcome to Nowcasting dashboard page.<br><br>This page will contain dashboard for the prediction pipelines, and allow you to manage and add new predictions.<br>The dashapp is still under development..."
    # dashapp


# ===============================================================================
# Soil Moisture Comparison Tool App Layout
# ===============================================================================
# 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",
    },
]


dashapp = Dash(
    __name__,
    server=app,
    routes_pathname_prefix="/dashapp/",
    external_scripts=external_scripts,
    external_stylesheets=external_stylesheets,
    title="Soil Moisture Comparison Tool",
    update_title="Loading the tool...",
)

# farm_name = "Arawa"
# layer = "SM2"

time_delta = datetime.timedelta(days=20)
FAIL_IMAGE = dashapp.get_asset_url("icons/fail.png")
SUCCESS_IMAGE = dashapp.get_asset_url("icons/success.png")
WAIT_IMAGE = dashapp.get_asset_url("icons/wait.png")

current_working_directory = os.getcwd()
dashapp.index_template = os.path.join(
    current_working_directory, "templates", "dashapp.html"
)
dashapp.layout = layout(WAIT_IMAGE)


def find_model_path(real_path):
    for root, dirs, _ in os.walk(real_path):
        # print(root,dirs,"\n\n")
        if "soilwatermodel" in dirs:
            real_path = os.path.join(root, "soilwatermodel")
            print("real_path is", real_path)
            return real_path
    return real_path


# ====================================================================================================
# Callbacks
# ====================================================================================================
@dashapp.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):
    global MODEL_PATH
    global FARM_NAME
    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}")
        try:
            path = find_model_path(real_path)
            real_path = path
        except Exception:
            print("Exception raised while searching for the root")

        MODEL_PATH = path

        if os.path.exists(real_path):
            session = farm_name
            FARM_NAME = 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


@dashapp.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 ""


@dashapp.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
    """
    global MODEL_PATH
    global FARM_NAME
    print("\nAnalysis callback triggered")

    if n_clicks == 0 or n_clicks is None:
        raise PreventUpdate

    # window_start = datetime.datetime.strptime(window_start, '%Y-%m-%d')
    # window_end = datetime.datetime.strptime(window_end, '%Y-%m-%d')
    print(f"\nPath: {MODEL_PATH}\n")

    files = perform_analysis(
        input=MODEL_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,
        farm_name=FARM_NAME,
    )

    print(MODEL_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


@dashapp.callback(
    Output("graph", "figure"),
    Input("visualisation-select", "value"),
    Input("platter-dropdown", "value"),
    Input("generate-button", "n_clicks"),
)
def change_colorscale(file, palette, n_clicks):
    """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
    """
    if n_clicks == 0 or n_clicks is None or file is None:
        raise PreventUpdate

    band1, lons_a, lats_a = open_image(file)

    # Get the second dimension of the lons
    lats = lats_a[:, 0]
    lons = lons_a[0, :]
    if "quantile" in file:
        value_type = "Percentile"
    else:
        value_type = "SM"

    print(lons.shape, lons)
    print(lats.shape, lats)
    print(band1.shape, band1)
    print(file)
    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": f"<b>{value_type}"
                + "</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/{}"),
    )
    parser.add_argument(
        "-d",
        "--debug",
        help="Debug mode as True or False. Default is True.",
        default=True,
    )
    parser.add_argument(
        "-s",
        "--spaces",
        help="Whether running on huggingface spaces or not. Default is False.",
        action=argparse.BooleanOptionalAction,
    )

    args = parser.parse_args()
    INPUT = args.input

    try:
        # dashapp.run_server(debug=args.debug)
        if args.spaces:
            app.run(debug=args.debug, host="0.0.0.0", port=7860)
        else:
            app.run(debug=args.debug)
    except Exception as e:
        vprint(
            0,
            V,
            V_IGNORE,
            Error="Failed to execute the main function:",
            ErrorMessage=e,
        )
        traceback.print_exc()
        raise e