rolling_mean_dascore_edge.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Rolling Mean Processing of DAS Data in Real-time\n",
    "\n",
    "This Jupyter Notebook is created for performing rolling mean processing on a [spool](https://dascore.org/tutorial/concepts.html#:~:text=read%20the%20docs!-,Data%20structures,-DASCore%20has%20two) of distributed acoustic sensing (DAS) data in real-time. It uses [DASCore](https://dascore.org/) package and the ```lf_das.py``` script. \n",
    "\n",
    "\n",
    "<svg width=\"100%\" height=\"1\">\n",
    "  <line x1=\"0\" y1=\"0\" x2=\"100%\" y2=\"0\" style=\"stroke:rgb(0,0,0);stroke-width:2\" />\n",
    "</svg>\n",
    "\n",
    "\n",
    "#### Notes: \n",
    "1. Before using this notebook, make sure you have included the ```lf_das.py``` script in the current directory with this notebook and successfully installed DASCore using ```pip``` (recommended) or ```conda```:\n",
    "    ```python\n",
    "    pip install dascore\n",
    "    ```\n",
    "    or\n",
    "    ```python\n",
    "    conda install dascore -c conda-forge\n",
    "    ```   \n",
    "2. Please find all supported I/O [here](https://dascore.org/#:~:text=specialized%20analysis/visualization.-,Supported%20file%20formats,-name).\n",
    "3. You need to have at least 1 patch of data in the directory before you start real-time processing.\n",
    " \n",
    "\n",
    "Current DASCore version: 0.0.13 (tested)\n",
    "\n",
    "Date: 09/07/2023\n",
    "\n",
    "\n",
    "Contact: [Ahmad Tourei](https://github.com/ahmadtourei/)\n",
    "\n",
    "ahmadtourei@gmail.com"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import libraries\n",
    "import warnings\n",
    "warnings.simplefilter('ignore')\n",
    "\n",
    "import dascore as dc\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import time \n",
    "\n",
    "from dascore.units import s # s as seconds\n",
    "from datetime import datetime\n",
    "from lf_das import _get_filename\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get a spool of data to work on"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define data path (spool of data) and output folder \n",
    "data_path = '/mnt/h/data'\n",
    "output_data_folder =  '/mnt/h/results'\n",
    "output_figure_folder = '/mnt/h/figures'\n",
    "\n",
    "# get the sorted spool of data form the defined data path (on first run, it will index the patches and subsequently update the index file for future uses)\n",
    "sp = dc.spool(data_path).sort(\"time\").update()\n",
    "\n",
    "# print the contents of first 5 patches\n",
    "content_df = sp.get_contents()\n",
    "content_df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get some metadata and define a sub spool (if needed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get sampling rate, channel spacing, and gauge length from the first patch\n",
    "patch_0 = sp[0]\n",
    "gauge_length = patch_0.attrs['gauge_length']\n",
    "print(\"Gauge length = \", gauge_length)\n",
    "channel_spacing = patch_0.attrs['d_distance']\n",
    "print(\"Channel spacing = \", channel_spacing)\n",
    "sampling_interval = patch_0.attrs['d_time']\n",
    "print(\"Sampling interval = \", sampling_interval)\n",
    "sampling_rate = 1/(sampling_interval / np.timedelta64(1, 's'))\n",
    "print(\"Sampling rate = \", sampling_rate)\n",
    "num_sec = len(patch_0.coords[\"time\"])/sampling_rate\n",
    "print(\"Number of seconds in each patch= \", num_sec)\n",
    "\n",
    "# select a sub-spool\n",
    "ch_start = 400\n",
    "ch_end = 1400\n",
    "d_1 = patch_0.coords['distance'][ch_start] # in meter\n",
    "d_2 = patch_0.coords['distance'][ch_end] # in meter\n",
    "# or:\n",
    "# d_1 = -115 # in meter\n",
    "# d_2 = 2000 # in meter\n",
    "sub_sp = sp.select(distance=(d_1, d_2)) \n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Set real-time processing parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the target sampling interval in seconds\n",
    "d_t = 10.0 # so, cutoff_freq = Nyq_new = 1/(2*d_t) = 0.05 hz\n",
    "\n",
    "# determine window size in sec.\n",
    "window = d_t*s\n",
    "\n",
    "# determine step size in sec.\n",
    "step = d_t*s\n",
    "\n",
    "# define the scale to apply to the raw data\n",
    "scale_iDAS = float((116*sampling_rate/gauge_length)/1e9)\n",
    "\n",
    "# set the desired wait time after each run\n",
    "time_step_for_processing = num_sec # in sec.\n",
    "\n",
    "print(\"time_step_for_processing: \", time_step_for_processing, \"\\n\")\n",
    "print(\"patch length: \", num_sec, \"\\n\")\n",
    "\n",
    "# set the starting time from which low-freq. processing applys (it can be the time_min for the first patch of the spool)\n",
    "start_processing_time = np.datetime64('2023-03-22T06:00:00') # in UTC, or any other time zone that original data are stored \\\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Do rolling mean processing in real-time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# start the for loop for real-time processing\n",
    "initial_run = True\n",
    "while True:\n",
    "    # select a updated sub-spool\n",
    "    sp = dc.spool(data_path).sort(\"time\").update()\n",
    "    sub_sp = sp.select(distance=(d_1, d_2))\n",
    "    len_updated_sp = len(sub_sp)\n",
    "\n",
    "    if not initial_run:\n",
    "        num_added_patches = len_updated_sp - len_last_sp\n",
    "        if len_last_sp == len_updated_sp:\n",
    "            print(\"No new data was detected. Real-time data processing ended successfully.\")\n",
    "            break\n",
    "\n",
    "    if initial_run:\n",
    "        i=1\n",
    "        print(\"\\nrun number: \", i)\n",
    "        for j, patch in enumerate (sub_sp):\n",
    "            print (\"working on patch \", j)\n",
    "            # apply rolling mean function\n",
    "            rolling_mean_patch = patch.rolling(time=window, step=step, engine=\"numpy\").mean()\n",
    "            # scale data\n",
    "            new_scaled_patch = rolling_mean_patch.new(data=rolling_mean_patch.data*scale_iDAS) \n",
    "\n",
    "            # save the result to output folder\n",
    "            filename = _get_filename(new_scaled_patch.attrs['time_min'],\n",
    "                        new_scaled_patch.attrs['time_max'])\n",
    "            filename = output_data_folder + '/' + filename \n",
    "            new_scaled_patch.io.write(filename, \"dasdae\") \n",
    "\n",
    "        initial_run = False \n",
    "        len_last_sp = len(sub_sp)\n",
    "        time.sleep(time_step_for_processing)\n",
    "    else:\n",
    "        i+=1\n",
    "        print(\"\\nrun number: \", i)\n",
    "        for j in range(len_last_sp,len_updated_sp):\n",
    "            patch = sub_sp[j]\n",
    "            print (\"working on patch \", j)\n",
    "            # apply rolling mean function\n",
    "            rolling_mean_patch = patch.rolling(time=window, step=step, engine=\"numpy\").mean()\n",
    "            # scale data\n",
    "            new_scaled_patch = rolling_mean_patch.new(data=rolling_mean_patch.data*scale_iDAS) \n",
    "\n",
    "            # save the result to output folder\n",
    "            filename = _get_filename(new_scaled_patch.attrs['time_min'],\n",
    "                        new_scaled_patch.attrs['time_max'])\n",
    "            filename = output_data_folder + '/' + filename \n",
    "            new_scaled_patch.io.write(filename, \"dasdae\") \n",
    "        len_last_sp = len(sub_sp)\n",
    "        time.sleep(time_step_for_processing)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "dascore",
   "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}