warragamba.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Warragamba"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Install dependencies\n",
    "\n",
    "```\n",
    "poetry install\n",
    "```\n",
    "\n",
    "Use poetry env\n",
    "```\n",
    "poetry shell\n",
    "```\n",
    "\n",
    "If using VSCode use the python env denoted as 'Poetry env'"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Imports and Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np # Numerical python lib\n",
    "import pandas as pd # Data frame lib\n",
    "import altair as alt # Chart plotting lib\n",
    "import tensorflow as tf # Neural Network lib\n",
    "import requests\n",
    "import json\n",
    "import pprint\n",
    "\n",
    "from tensorflow.keras import layers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "water_nsw_api_endpoint = \"https://realtimedata.waternsw.com.au/cgi/webservice.pl\"\n",
    "\n",
    "def get_ts_trace(station_id: str, start_time: str, end_time: str, variable: str, interval: str, aggregate: str, datasource=\"A\", multiplier=\"1\"):\n",
    "    \n",
    "    payload = json.dumps({\n",
    "        \"function\": \"get_ts_traces\",\n",
    "        \"version\": 2,\n",
    "        \"params\": {\n",
    "            \"site_list\": station_id,\n",
    "            \"datasource\": datasource,\n",
    "            \"start_time\": start_time,\n",
    "            \"end_time\": end_time,\n",
    "            \"var_list\": variable,\n",
    "            \"interval\": interval,\n",
    "            \"multiplier\": multiplier,\n",
    "            \"data_type\": aggregate\n",
    "        }        \n",
    "    })\n",
    "    \n",
    "    json_response = requests.post(water_nsw_api_endpoint, payload).json()\n",
    "\n",
    "    try:\n",
    "        return json_response[\"_return\"][\"traces\"][0][\"trace\"]\n",
    "    except KeyError:\n",
    "        return json_response"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fetch Data from API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "weather_station_ids = [\"563035\", \"563046\", \"563079\", \"568045\", \"568051\"]\n",
    "stream_station_ids = [\"212250\", \"212270\"]\n",
    "\n",
    "def cast_data_types(df):\n",
    "    df[\"v\"] = df[\"v\"].astype(float)\n",
    "    df[\"q\"] = df[\"q\"].astype(float)\n",
    "    df[\"t\"] = pd.to_datetime(df[\"t\"], format=\"%Y%m%d%H%M%S\")\n",
    "    return df\n",
    "    \n",
    "\n",
    "json_response = get_ts_trace(\n",
    "    \"212242\",\n",
    "    \"20080130000000\", # when the warragamba dam measurements start\n",
    "    \"20220111000000\", # when warragamba dam measurements end\n",
    "    variable=\"130.00\",\n",
    "    interval=\"day\",\n",
    "    aggregate=\"mean\",\n",
    "    datasource=\"CP\"\n",
    "    \n",
    ")\n",
    "water_level_df = pd.json_normalize(json_response)\n",
    "cast_data_types(water_level_df)\n",
    "\n",
    "\n",
    "rainfall_dfs = []\n",
    "for station in weather_station_ids:\n",
    "    json_response = get_ts_trace(\n",
    "        station,\n",
    "        \"20080130000000\", \n",
    "        \"20220111000000\",\n",
    "        variable=\"10.00\",\n",
    "        interval=\"day\",\n",
    "        aggregate=\"mean\",\n",
    "        datasource=\"CP\"   \n",
    "    )\n",
    "    \n",
    "    rainfall_dfs.append(\n",
    "        cast_data_types(\n",
    "            pd.json_normalize(json_response)\n",
    "        )\n",
    "    )\n",
    "\n",
    "stream_dfs = []\n",
    "for station in stream_station_ids:\n",
    "    json_response = get_ts_trace(\n",
    "        station,\n",
    "        \"20080130000000\", \n",
    "        \"20220111000000\",\n",
    "        variable=\"100.00\",\n",
    "        interval=\"day\",\n",
    "        aggregate=\"mean\",\n",
    "        datasource=\"CP\"   \n",
    "    )\n",
    "    \n",
    "    stream_dfs.append(\n",
    "        cast_data_types(\n",
    "            pd.json_normalize(json_response)\n",
    "        )\n",
    "    )"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deriving 'Water Level Difference'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "water_level = water_level_df[\"v\"]\n",
    "water_level_tomorrow = water_level_df[\"v\"].copy().shift(1, fill_value=0.0)\n",
    "water_level_df[\"water_level_difference\"] = water_level - water_level_tomorrow\n",
    "water_level_df = water_level_df[water_level_df[\"water_level_difference\"] > -0.3]"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cleaning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "water_level_df = water_level_df[~water_level_df[\"q\"].isin([201,205])]\n",
    "water_level_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for stream_df in stream_dfs:\n",
    "    stream_df.loc[stream_df[\"q\"].isin([201,255]), \"v\"] = 0.0\n",
    "    \n",
    "for rainfall_df in rainfall_dfs:\n",
    "    rainfall_df.loc[rainfall_df[\"q\"].isin([201,255]), \"v\"] = 0.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def rename_cols_with_id(dfs, station_ids: list, variable_no: str):\n",
    "    for df, station_id in zip(dfs, station_ids):\n",
    "        df.rename(\n",
    "            columns={\n",
    "                \"v\": f\"v_{station_id}_{variable_no}\",\n",
    "                \"q\": f\"q_{station_id}_{variable_no}\",\n",
    "            },\n",
    "            inplace=True,\n",
    "        )\n",
    "\n",
    "\n",
    "water_level_df = water_level_df.rename(\n",
    "    columns={\"v\": \"v_212242_130\", \"q\": \"q_212242_130\"}\n",
    ")\n",
    "\n",
    "rename_cols_with_id(rainfall_dfs, weather_station_ids, \"10\")\n",
    "rename_cols_with_id(stream_dfs, stream_station_ids, \"100\")\n",
    "\n",
    "df = water_level_df\n",
    "\n",
    "for rainfall_df in rainfall_dfs:\n",
    "    df = pd.merge(left=df, right=rainfall_df, how=\"inner\", on=\"t\")\n",
    "\n",
    "for stream_df in stream_dfs:\n",
    "    df = pd.merge(left=df, right=stream_df, how=\"inner\", on=\"t\")\n",
    "\n",
    "df.columns"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare Data for Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_columns = [\n",
    "  \"v_568051_10\", \"v_568045_10\", \"v_563079_10\",\n",
    "  \"v_563046_10\", \"v_563035_10\", \"v_212250_100\",\n",
    "  \"v_212270_100\"]\n",
    "\n",
    "label_column = \"water_level_difference\"\n",
    "model_columns = feature_columns + [label_column]\n",
    "\n",
    "train_data = df.copy()[model_columns].sample(frac=0.8, random_state=12345)\n",
    "test_data = df.copy()[model_columns].drop(train_data.index)\n",
    "\n",
    "train_features = train_data.copy()\n",
    "test_features = test_data.copy()\n",
    "\n",
    "train_labels = train_features.pop(\"water_level_difference\").values.reshape(-1, 1)\n",
    "test_labels = test_features.pop(\"water_level_difference\").values.reshape(-1, 1)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building the Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "normaliser = tf.keras.layers.Normalization(axis=1)\n",
    "normaliser.adapt(train_features)\n",
    "\n",
    "test_model = tf.keras.Sequential(name=\"stream_and_rain_model\", layers=[\n",
    "    layers.Input(shape=(7,)),\n",
    "    normaliser,\n",
    "    layers.Dropout(rate=0.5),\n",
    "    layers.Dense(units=16, activation=\"relu\"),\n",
    "    layers.Dropout(rate=0.5),\n",
    "    layers.Dense(units=1)\n",
    "])\n",
    "\n",
    "test_model.summary()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_model.compile(\n",
    "    optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),\n",
    "    loss='mean_absolute_error')\n",
    "\n",
    "history = test_model.fit(\n",
    "    train_features.values,\n",
    "    train_labels,\n",
    "    epochs=30,\n",
    "    # Suppress logging.\n",
    "    verbose=1,\n",
    "    # Calculate validation results on 20% of the training data.\n",
    "    validation_split = 0.2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist_df = pd.DataFrame(history.history)\n",
    "hist_df[\"epoch\"] = history.epoch\n",
    "hist_df.rename(\n",
    "    columns={\"loss\":\"training_loss\", \"val_loss\":\"validation_loss\"},\n",
    "    inplace=True\n",
    ")\n",
    "\n",
    "alt.Chart(hist_df).mark_line().transform_fold(\n",
    "    fold=['training_loss', 'validation_loss'], \n",
    "    as_=['variable', 'loss']\n",
    ").encode(\n",
    "    x=\"epoch:Q\",\n",
    "    y=\"loss:Q\",\n",
    "    color=\"variable:N\"\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_model.evaluate(test_features, test_labels)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y = test_model.predict(df[feature_columns])\n",
    "\n",
    "compare_df = pd.DataFrame({\n",
    "    \"t\": df[[\"t\"]].values.flatten(),\n",
    "    \"actual\": df[[\"water_level_difference\"]].values.flatten(),\n",
    "    \"prediction\": y.flatten()\n",
    "})\n",
    "\n",
    "base = alt.Chart(compare_df.reset_index()[0:5000]).encode(\n",
    "    x=\"index:Q\"\n",
    ")\n",
    "\n",
    "(base.mark_line().encode(\n",
    "    y=\"actual:Q\"\n",
    ") + base.mark_line(color=\"orange\").encode(\n",
    "    y=\"prediction:Q\"\n",
    ")).interactive()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "warragamba-blogpost-FG-3Jor--py3.10",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.9"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "32af6703f6de197a785ed248fd050900b54929d05b2cd99dec9c67c83648d162"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}