data_report

measurement.dash_apps.data_report

Attributes

app = DjangoDash('DataReport', external_stylesheets=[dbc.themes.BOOTSTRAP, '/static/styles/dashstyle.css']) module-attribute

filters = html.Div(style={'width': '286px'}, children=[html.Label('Temporality:', style={'font-weight': 'bold'}), dcc.Dropdown(id='temporality_drop', options=[{'label': 'Raw measurement', 'value': 'measurement'}, {'label': 'Validated measurement', 'value': 'validated'}, {'label': 'Hourly', 'value': 'hourly'}, {'label': 'Daily', 'value': 'daily'}, {'label': 'Monthly', 'value': 'monthly'}], value='measurement'), html.Br(), html.Label('Station:', style={'font-weight': 'bold'}), dcc.Dropdown(id='station_drop', options=[], value=None), html.Br(), html.Label('Variable:', style={'font-weight': 'bold'}), dcc.Dropdown(id='variable_drop', options=[], value=None), html.Br(), html.Label('Date Range:', style={'font-weight': 'bold'}), dcc.DatePickerRange(id='date_range_picker', display_format='YYYY-MM-DD', start_date=None, end_date=None), html.Br(), html.Div(id='csv_div', style={'margin-top': '30px'})]) module-attribute

Classes

Variable

Bases: PermissionsBase

A variable with a physical meaning.

Such as precipitation, wind speed, wind direction, soil moisture, including the associated unit. It also includes metadata to help identify what is a reasonable value for the data, to flag outliers and to help with the validation process.

The nature of the variable can be one of the following:

• sum: Cumulative value over a period of time.
• average: Average value over a period of time.
• value: One-off value.

Attributes:

Name	Type	Description
variable_id	AutoField	Primary key.
variable_code	CharField	Code of the variable, eg. airtemperature.
name	CharField	Human-readable name of the variable, eg. Air temperature.
unit	ForeignKey	Unit of the variable.
maximum	DecimalField	Maximum value allowed for the variable.
minimum	DecimalField	Minimum value allowed for the variable.
diff_error	DecimalField	If two sequential values in the time-series data of this variable differ by more than this value, the validation process can mark this with an error flag.
outlier_limit	DecimalField	The statistical deviation for defining outliers, in times the standard deviation (sigma).
null_limit	DecimalField	The max % of null values (missing, caused by e.g. equipment malfunction) allowed for hourly, daily, monthly data. Cumulative values are not deemed trustworthy if the number of missing values in a given period is greater than the null_limit.
nature	CharField	Nature of the variable, eg. if it represents a one-off value, the average over a period of time or the cumulative value over a period

Attributes

is_cumulative: bool property

Return True if the nature of the variable is sum.

Functions

__str__()

Return the string representation of the object.

Source code in variable/models.py

def __str__(self) -> str:
    """Return the string representation of the object."""
    return str(self.name)

clean()

Validate the model fields.

Source code in variable/models.py

def clean(self) -> None:
    """Validate the model fields."""
    if self.maximum < self.minimum:
        raise ValidationError(
            {
                "maximum": "The maximum value must be greater than the minimum "
                "value."
            }
        )
    if not self.variable_code.isidentifier():
        raise ValidationError(
            {
                "variable_code": "The variable code must be a valid Python "
                "identifier. Only letters, numbers and underscores are allowed, and"
                " it cannot start with a number."
            }
        )
    return super().clean()

get_absolute_url()

Get the absolute URL of the object.

Source code in variable/models.py

def get_absolute_url(self) -> str:
    """Get the absolute URL of the object."""
    return reverse("variable:variable_detail", kwargs={"pk": self.pk})

Functions

create_empty_plot()

Creates empty plot

Returns:

Type	Description
scatter	px.Scatter: Plot

Source code in measurement/dash_apps/plots.py

def create_empty_plot() -> px.scatter:
    """Creates empty plot

    Returns:
        px.Scatter: Plot
    """
    fig = px.scatter(title="No data to plot")
    fig.update_layout(
        autosize=True,
        margin=dict(
            l=50,
            r=20,
            b=0,
            t=50,
        ),
        title_font=dict(
            size=14,
        ),
    )
    return fig

create_report_plot(data, variable_name, station_code)

Creates plot for Report app

Parameters:

Name	Type	Description	Default
data	DataFrame	Data	required
variable_name	str	Variable name	required
station_code	str	Station code	required

Returns:

Type	Description
scatter	px.Scatter: Plot

Source code in measurement/dash_apps/plots.py

def create_report_plot(
    data: pd.DataFrame, variable_name: str, station_code: str
) -> px.scatter:
    """Creates plot for Report app

    Args:
        data (pd.DataFrame): Data
        variable_name (str): Variable name
        station_code (str): Station code

    Returns:
        px.Scatter: Plot
    """
    fig = px.scatter(
        data,
        x="time",
        y=["value", "minimum", "maximum"],
        title=f"{station_code} - {variable_name}",
        labels={
            "time": "Date",
        },
    )

    fig.for_each_trace(lambda trace: trace.update(name=trace.name.title()))
    fig.update_traces(marker=dict(size=3))
    fig.update_layout(
        legend=dict(
            title=dict(text="", font=dict(size=12)),
            x=1,
            y=1,
            xanchor="auto",
            yanchor="auto",
        ),
        autosize=True,
        margin=dict(
            l=50,
            r=20,
            b=0,
            t=50,
        ),
        yaxis_title=f"{variable_name}",
        title_font=dict(
            size=14,
        ),
    )

    return fig

download_csv_report(n_clicks, temporality, station, variable, start_time, end_time)

Source code in measurement/dash_apps/data_report.py

@app.callback(
    [
        Output("download_csv", "data"),
        Output("csv_alert_div", "children"),
    ],
    Input("csv_button", "n_clicks"),
    [
        State("temporality_drop", "value"),
        State("station_drop", "value"),
        State("variable_drop", "value"),
        State("date_range_picker", "start_date"),
        State("date_range_picker", "end_date"),
    ],
    prevent_initial_call=True,
)
def download_csv_report(
    n_clicks: int,
    temporality: str,
    station: str,
    variable: str,
    start_time: str,
    end_time: str,
):
    if n_clicks and n_clicks > 0:
        try:
            file = (
                get_report_data_from_db(
                    station=station,
                    variable=variable,
                    start_time=start_time,
                    end_time=end_time,
                    report_type=temporality,
                )
                .drop(columns=["station", "variable"])
                .to_csv(index=False)
            )
        except Exception as e:
            alert = dbc.Alert(f"Could not export data to CSV: {e}", color="warning")
            return None, [alert]
        return (
            dict(
                content=file,
                filename=f"{station}_{variable}_{temporality}_{start_time}-{end_time}.csv",
            ),
            [],
        )

get_date_range(station, variable)

Get the date range covered by a chosen station and variable.

Parameters:

Name	Type	Description	Default
station	str	Code for the chosen station	required
variable	str	Code for the chosen variable	required

Returns:

Type	Description
tuple[str, str]	tuple[str, str]: Start date, end date

Source code in measurement/filters.py

def get_date_range(station: str, variable: str) -> tuple[str, str]:
    """Get the date range covered by a chosen station and variable.

    Args:
        station (str): Code for the chosen station
        variable (str): Code for the chosen variable

    Returns:
        tuple[str, str]: Start date, end date
    """
    filter_vals = Measurement.objects.filter(
        station__station_code=station,
        variable__variable_code=variable,
    ).aggregate(
        first_date=Min("time"),
        last_date=Max("time"),
    )

    first_date = (
        filter_vals["first_date"].strftime("%Y-%m-%d")
        if filter_vals["first_date"]
        else None
    )
    last_date = (
        filter_vals["last_date"].strftime("%Y-%m-%d")
        if filter_vals["last_date"]
        else None
    )

    return first_date, last_date

get_report_data_from_db(station, variable, start_time, end_time, report_type)

Retrieves the report data from the database.

Time is set to the station timezone and the time range is inclusive of both start and end times.

Parameters:

Name	Type	Description	Default
station	str	Station of interest.	required
variable	str	Variable of interest.	required
start_time	str	Start time.	required
end_time	str	End time.	required
report_type	str	Type of report to retrieve.	required

Returns:

Type	Description
DataFrame	A dataframe with the report data.

Source code in measurement/reporting.py

def get_report_data_from_db(
    station: str,
    variable: str,
    start_time: str,
    end_time: str,
    report_type: str,
) -> pd.DataFrame:
    """Retrieves the report data from the database.

    Time is set to the station timezone and the time range is inclusive of both
    start and end times.

    Args:
        station: Station of interest.
        variable: Variable of interest.
        start_time: Start time.
        end_time: End time.
        report_type: Type of report to retrieve.

    Returns:
        A dataframe with the report data.
    """
    start_time_, end_time_ = reformat_dates(station, start_time, end_time)

    if report_type == "measurement":
        data = pd.DataFrame.from_records(
            Measurement.objects.filter(
                station__station_code=station,
                variable__variable_code=variable,
                time__gte=start_time_,
                time__lte=end_time_,
            ).values()
        )
        raw_cols = [col for col in data.columns if col.startswith("raw_")]
        normal = [col.strip("raw_") for col in raw_cols]
        data = data.drop(columns=normal).rename(columns=dict(zip(raw_cols, normal)))

    elif report_type == "validated":
        data = pd.DataFrame.from_records(
            Measurement.objects.filter(
                station__station_code=station,
                variable__variable_code=variable,
                time__gte=start_time_,
                time__lte=end_time_,
                is_validated=True,
                is_active=True,
            ).values()
        )
        raw_cols = [col for col in data.columns if col.startswith("raw_")]
        data = data.drop(columns=raw_cols)

    else:
        data = pd.DataFrame.from_records(
            Report.objects.filter(
                station__station_code=station,
                variable__variable_code=variable,
                time__gte=start_time_,
                time__lte=end_time_,
                report_type=report_type,
            ).values()
        )

    data = data.rename(columns={"station_id": "station", "variable_id": "variable"})
    if not data.empty:
        data = data.sort_values("time")

    return data

get_station_options(station_codes)

Get valid station options and default value based on permissions and data availability.

Parameters:

Name	Type	Description	Default
station_codes	list[str]	List of station codes based on permissions	required

Returns:

Type	Description
tuple[list[dict[str, str]], str | None]	tuple[list[dict], str]: Options for the station dropdown, default value

Source code in measurement/filters.py

def get_station_options(
    station_codes: list[str],
) -> tuple[list[dict[str, str]], str | None]:
    """Get valid station options and default value based on permissions and data
    availability.

    Args:
        station_codes (list[str]): List of station codes based on permissions

    Returns:
        tuple[list[dict], str]: Options for the station dropdown, default value
    """
    stations_with_measurements = Measurement.objects.values_list(
        "station__station_code", flat=True
    ).distinct()

    station_options = [
        {"label": station_code, "value": station_code}
        for station_code in station_codes
        if station_code in stations_with_measurements
    ]
    station_value = station_options[0]["value"] if station_options else None
    return station_options, station_value

get_variable_options(station)

Get valid variable options and default value based on the chosen station.

Parameters:

Name	Type	Description	Default
station	str	Code for the chosen station	required

Returns:

Type	Description
tuple[list[dict], str]	tuple[list[dict], str]: Options for the variable dropdown, default value

Source code in measurement/filters.py

def get_variable_options(station: str) -> tuple[list[dict], str]:
    """Get valid variable options and default value based on the chosen station.

    Args:
        station (str): Code for the chosen station

    Returns:
        tuple[list[dict], str]: Options for the variable dropdown, default value
    """
    variable_dicts = (
        Measurement.objects.filter(station__station_code=station)
        .values("variable__name", "variable__variable_code")
        .distinct()
    )

    variable_options = [
        {
            "label": variable["variable__name"],
            "value": variable["variable__variable_code"],
        }
        for variable in variable_dicts
    ]
    variable_value = variable_options[0]["value"] if variable_options else None
    return variable_options, variable_value

populate_stations_dropdown(station_codes)

Populate the station dropdown based on the list of station codes.

Source code in measurement/dash_apps/data_report.py

@app.callback(
    [Output("station_drop", "options"), Output("station_drop", "value")],
    Input("stations_list", "children"),
)
def populate_stations_dropdown(
    station_codes: list[str],
) -> tuple[list[dict[str, str]], str | None]:
    """Populate the station dropdown based on the list of station codes."""
    return get_station_options(station_codes)

populate_variable_dropdown(chosen_station)

Populate the variable dropdown based on the chosen station.

Source code in measurement/dash_apps/data_report.py

@app.callback(
    [Output("variable_drop", "options"), Output("variable_drop", "value")],
    Input("station_drop", "value"),
)
def populate_variable_dropdown(
    chosen_station: str,
) -> tuple[list[dict[str, str]], str | None]:
    """Populate the variable dropdown based on the chosen station."""
    return get_variable_options(chosen_station)

set_date_range(chosen_station, chosen_variable)

Set the default date range based on the chosen station and variable.

Source code in measurement/dash_apps/data_report.py

@app.callback(
    [
        Output("date_range_picker", "start_date"),
        Output("date_range_picker", "end_date"),
    ],
    [
        Input("station_drop", "value"),
        Input("variable_drop", "value"),
    ],
)
def set_date_range(
    chosen_station, chosen_variable
) -> tuple[
    str,
    str,
]:
    """Set the default date range based on the chosen station and variable."""
    return get_date_range(chosen_station, chosen_variable)

update_alert(figure)

Source code in measurement/dash_apps/data_report.py

@app.callback(
    [
        Output("data_alert_div", "children"),
        Output("csv_div", "children"),
    ],
    Input("data_report_graph", "figure"),
)
def update_alert(figure):
    if figure["layout"]["title"]["text"] == "Data not found":
        alert = dbc.Alert(
            "No data was found with the selected criteria", color="warning"
        )
        return [alert], []
    else:
        button = html.Button("Download CSV", id="csv_button")
        return [], [button]

update_graph(temporality, station, variable, start_time, end_time)

Source code in measurement/dash_apps/data_report.py

@app.callback(
    Output("data_report_graph", "figure"),
    [
        Input("temporality_drop", "value"),
        Input("station_drop", "value"),
        Input("variable_drop", "value"),
        Input("date_range_picker", "start_date"),
        Input("date_range_picker", "end_date"),
    ],
)
def update_graph(
    temporality: str,
    station: str,
    variable: str,
    start_time: str,
    end_time: str,
) -> px.line:
    try:
        data = get_report_data_from_db(
            station=station,
            variable=variable,
            start_time=start_time,
            end_time=end_time,
            report_type=temporality,
        )
        plot = create_report_plot(
            data=data,
            variable_name=Variable.objects.get(variable_code=variable).name,
            station_code=station,
        )

    except Exception as e:
        print("Error:", e)
        plot = create_empty_plot()

    return plot