Flag_LoadNumbers.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "pd.options.display.float_format = '{:,.4f}'.format"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## User Settings\n",
    "Here is every tunable parameter..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# -------------------------\n",
    "# ---Input/Output Files ---\n",
    "# -------------------------\n",
    "input_csv_path = r\"/Users/carynv5/Documents/NV5/USACE/dredge/Dodge Island 2023 Data.csv\"\n",
    "\n",
    "output_csv_path = '/Users/carynv5/Documents/NV5/USACE/dredge/OUTLIER_DETECTION/load_number_fix/Dodge_Data_LoadFlags.csv'\n",
    "\n",
    "# ----------------------\n",
    "# ---Time Thresholds ---\n",
    "# ----------------------\n",
    "# this is the maximum allowable gap between loads on a contract. This threshold is used to \n",
    "# remove outliers when differencing time stamps between rows to avoid outliers in analysis.\n",
    "contract_gap_threshold = pd.to_timedelta('5 days') # 5 days\n",
    "\n",
    "# if the average time deltas of a load (or group) exceed this threshold, the entire load is \n",
    "# flagged as having potential data gaps. Additionally, this threshold is used to identify\n",
    "# non-contiguous loads within a single contract\n",
    "load_gap_threshold = pd.to_timedelta('0 days 00:00:20') # 20 seconds\n",
    "\n",
    "# -------------------------\n",
    "# ---Min/Max Thresholds ---\n",
    "# -------------------------\n",
    "# any loads (or groups) whose durations that fall outside these thresholds will be flagged as\n",
    "# having invalid durations\n",
    "duration_threshold_min = pd.to_timedelta('0 days 01:00:00') # 1 hour\n",
    "duration_threshold_max = pd.to_timedelta('1 days 00:00:00') # 1 day\n",
    "\n",
    "# any loads (or groups) whose total point count falls outside these thresholds will be flagged\n",
    "# as having an invalid number of points\n",
    "point_threshold_min = 375 # set to approximate duration_threshold_min (assuming 10 second ping rate)\n",
    "point_threshold_max = 8750 # set to approximate  duration_threshold_max (assuming 10 second ping rate)\n",
    "\n",
    "# we estimate ping rate for each group based on the group's duration and the total number of points.\n",
    "# If the actual ping rate is different, we flag that group (or load). In seconds.\n",
    "ping_rate_threshold = 1.0 # 1 second"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Data Preprocessing\n",
    "Ingest CSV, compute per-point statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "orig_df = pd.read_csv(input_csv_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "orig_df['msg_time'] = pd.to_datetime(orig_df['msg_time']).dt.tz_localize(None)\n",
    "orig_df = orig_df.sort_values('msg_time')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# determine the length of time between each data transmission for each datapoint in each contract,\n",
    "# and set the outliers to pd.NA (i.e. gaps within a contract)\n",
    "\n",
    "# to be implemented in Spark\n",
    "orig_df['time_delta'] = orig_df.groupby('contract_id')['msg_time'].diff()\n",
    "orig_df.loc[orig_df['time_delta'] > contract_gap_threshold, 'time_delta'] = pd.NA"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# assign load group numbers based on contract_id and contiguous load numbers\n",
    "# This sorts the DataFrame orig_df based on the values in the msg_time column in ascending order. This ensures that the data is in chronological order, which is important for the subsequent grouping and calculations.\n",
    "orig_df = orig_df.sort_values('msg_time')\n",
    "# spark - sorted_df = orig_df.orderBy(col(\"msg_time\").asc())\n",
    "\n",
    "\n",
    "# This function is defined to operate on each group of the DataFrame.\n",
    "# group['load_number'].shift() shifts the load_number column by one position, creating a new Series where each value is the previous row's load_number.\n",
    "# group['load_number'].ne(group['load_number'].shift()) compares each load_number with the previous one, resulting in a boolean Series where True indicates a change in load_number from the previous row.\n",
    "# .cumsum() computes the cumulative sum of this boolean Series. This essentially assigns a new group number every time there is a change in load_number.\n",
    "# The result is stored in a new column group_number in the DataFrame group.\n",
    "def calculate_group_number(group):\n",
    "    group['group_number'] = group['load_number'].ne(group['load_number'].shift()).cumsum()\n",
    "    return group\n",
    "\n",
    "orig_df = orig_df.groupby('contract_id').apply(calculate_group_number)\n",
    "\n",
    "# Quash the multindex back down to a single index with reset_index()\n",
    "orig_df = orig_df.reset_index(drop=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2. Create Load Group Intervals\n",
    "\n",
    "Create a dataframe of \"load group intervals\". These intervals represent contiguous points in time\n",
    "that all share a single load number. These are useful because load numbers can be mis-reported,\n",
    "so partitioning data by load number alone will lead to misleading statistics."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create an interval dataframe from each load group which describes\n",
    "# high-level statistics about each group, then compute an IndexInterval for each\n",
    "# A dictionary interval_dict is initialized with empty lists for each of the required summary statistics.\n",
    "interval_dict = {'contract': [], 'group': [], 'load': [], 'msg_time_min': [], \n",
    "                 'msg_time_max': [], 'time_delta_avg': [], 'total_points':[]}\n",
    "\n",
    "# The orig_df DataFrame is grouped by contract_id and group_number.\n",
    "# For each group, the following statistics are calculated and appended to the respective lists in the dictionary:\n",
    "# contract: The contract_id of the group.\n",
    "# group: The group_number of the group.\n",
    "# load: The minimum load_number within the group.\n",
    "# msg_time_min: The minimum msg_time within the group.\n",
    "# msg_time_max: The maximum msg_time within the group.\n",
    "# time_delta_avg: The average time_delta within the group.\n",
    "# total_points: The total number of points (rows) in the group.\n",
    "# Example:\n",
    "\n",
    "# SELECT \n",
    "#     contract_id AS contract,\n",
    "#     group_number AS `group`,\n",
    "#     MIN(load_number) AS load,\n",
    "#     MIN(msg_time) AS msg_time_min,\n",
    "#     MAX(msg_time) AS msg_time_max,\n",
    "#     AVG(time_delta) AS time_delta_avg,\n",
    "#     COUNT(*) AS total_points\n",
    "# FROM orig_table\n",
    "# GROUP BY contract_id, group_number;\n",
    "\n",
    "for idx, df in orig_df.groupby(['contract_id', 'group_number']):\n",
    "    interval_dict['contract'].append(idx[0])\n",
    "    interval_dict['group'].append(idx[1])\n",
    "    interval_dict['load'].append(df['load_number'].min())\n",
    "    interval_dict['msg_time_min'].append(df['msg_time'].min())\n",
    "    interval_dict['msg_time_max'].append(df['msg_time'].max())\n",
    "    interval_dict['time_delta_avg'].append(df['time_delta'].mean())\n",
    "    interval_dict['total_points'].append(df.shape[0])\n",
    "\n",
    "# Creating dataframe of the result - SQL would create a table or view\n",
    "group_intervals = pd.DataFrame(interval_dict)\n",
    "#group_intervals = pd.DataFrame(interval_dict).set_index(['contract', 'group'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>contract</th>\n",
       "      <th>group</th>\n",
       "      <th>load</th>\n",
       "      <th>msg_time_min</th>\n",
       "      <th>msg_time_max</th>\n",
       "      <th>time_delta_avg</th>\n",
       "      <th>total_points</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>445</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>2023-10-30 19:55:32</td>\n",
       "      <td>2023-10-31 00:11:32</td>\n",
       "      <td>0 days 00:00:05.245901639</td>\n",
       "      <td>2929</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>445</td>\n",
       "      <td>2</td>\n",
       "      <td>2</td>\n",
       "      <td>2023-10-31 00:11:34</td>\n",
       "      <td>2023-10-31 02:59:54</td>\n",
       "      <td>0 days 00:00:05.226073460</td>\n",
       "      <td>1933</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>445</td>\n",
       "      <td>3</td>\n",
       "      <td>3</td>\n",
       "      <td>2023-10-31 03:00:09</td>\n",
       "      <td>2023-10-31 07:00:39</td>\n",
       "      <td>0 days 00:00:05.243194192</td>\n",
       "      <td>2755</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>445</td>\n",
       "      <td>4</td>\n",
       "      <td>4</td>\n",
       "      <td>2023-10-31 07:00:49</td>\n",
       "      <td>2023-10-31 10:11:36</td>\n",
       "      <td>0 days 00:00:05.142280071</td>\n",
       "      <td>2228</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>445</td>\n",
       "      <td>5</td>\n",
       "      <td>5</td>\n",
       "      <td>2023-10-31 10:11:46</td>\n",
       "      <td>2023-10-31 15:14:46</td>\n",
       "      <td>0 days 00:00:05.339007924</td>\n",
       "      <td>3407</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   contract  group  load        msg_time_min        msg_time_max  \\\n",
       "0       445      1     1 2023-10-30 19:55:32 2023-10-31 00:11:32   \n",
       "1       445      2     2 2023-10-31 00:11:34 2023-10-31 02:59:54   \n",
       "2       445      3     3 2023-10-31 03:00:09 2023-10-31 07:00:39   \n",
       "3       445      4     4 2023-10-31 07:00:49 2023-10-31 10:11:36   \n",
       "4       445      5     5 2023-10-31 10:11:46 2023-10-31 15:14:46   \n",
       "\n",
       "             time_delta_avg  total_points  \n",
       "0 0 days 00:00:05.245901639          2929  \n",
       "1 0 days 00:00:05.226073460          1933  \n",
       "2 0 days 00:00:05.243194192          2755  \n",
       "3 0 days 00:00:05.142280071          2228  \n",
       "4 0 days 00:00:05.339007924          3407  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# What the output looks like\n",
    "group_intervals.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# compute useful columns:\n",
    "# interval index\n",
    "\n",
    "# SQL databases do not have a direct equivalent of pandas' IntervalIndex, but you can achieve a similar result by using the minimum and maximum timestamps to define intervals.\n",
    "# Example:\n",
    "\n",
    "# CREATE VIEW interval_view AS\n",
    "# SELECT \n",
    "#     contract,\n",
    "#     group,\n",
    "#     msg_time_min,\n",
    "#     msg_time_max,\n",
    "#     CONCAT('[',\n",
    "#            DATE_FORMAT(msg_time_min, '%Y-%m-%d %H:%i:%s'),\n",
    "#            ', ',\n",
    "#            DATE_FORMAT(msg_time_max, '%Y-%m-%d %H:%i:%s'),\n",
    "#            ']') AS interval\n",
    "# FROM group_intervals;\n",
    "group_intervals['interval'] = pd.IntervalIndex.from_arrays(np.array(group_intervals['msg_time_min']), \n",
    "                                                      np.array(group_intervals['msg_time_max']), \n",
    "                                                      closed='both')\n",
    "\n",
    "\n",
    "\n",
    "# This calculates the total time duration for each interval in the group_intervals DataFrame and storing the result in a new column called total_time.\n",
    "\n",
    "# You can calculate the duration between the msg_time_min and msg_time_max columns in SQL.\n",
    "# Example:\n",
    "\n",
    "# SELECT \n",
    "#     contract,\n",
    "#     group,\n",
    "#     msg_time_min,\n",
    "#     msg_time_max,\n",
    "#     TIMESTAMPDIFF(SECOND, msg_time_min, msg_time_max) AS total_time_seconds\n",
    "# FROM \n",
    "#     group_intervals;\n",
    "\n",
    "\n",
    "# total length of an interval\n",
    "group_intervals['total_time'] = group_intervals['interval'].apply(lambda x: (x.right - x.left))\n",
    "\n",
    "\n",
    "\n",
    "#  This calculates the midpoint of each interval in the group_intervals DataFrame and storing the result in a new column called msg_time_mid\n",
    "\n",
    "# To achieve the same result in SQL, you can calculate the midpoint between the msg_time_min and msg_time_max columns. \n",
    "# Example:\n",
    "\n",
    "# SELECT \n",
    "#     contract,\n",
    "#     group,\n",
    "#     msg_time_min,\n",
    "#     msg_time_max,\n",
    "#     DATE_ADD(msg_time_min, INTERVAL TIMESTAMPDIFF(SECOND, msg_time_min, msg_time_max) / 2 SECOND) AS msg_time_mid\n",
    "# FROM \n",
    "#     group_intervals;\n",
    "\n",
    "# midpoint of an interval\n",
    "group_intervals['msg_time_mid'] = group_intervals['interval'].apply(lambda x: x.mid)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Flag Any Outliers in the Load Groups"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# flag any groups where the duration is either too short or two long... 1 means the group \n",
    "# is positive for exceeding threshold, 0 means group is within valid range\n",
    "\n",
    "\n",
    "# Example:\n",
    "\n",
    "# Variable: duration_threshold_min defined in seconds (1hr)\n",
    "# SET @duration_threshold_min = 3600; or pass a variable into the query.\n",
    "\n",
    "# SELECT \n",
    "#     contract,\n",
    "#     group,\n",
    "#     total_time,\n",
    "#     CASE \n",
    "#         WHEN total_time < @duration_threshold_min THEN 1\n",
    "#         ELSE 0\n",
    "#     END AS invalid_load_duration_short\n",
    "# FROM \n",
    "#     group_intervals;\n",
    "\n",
    "group_intervals['invalid_load_duration_short'] = np.where(\n",
    "    (group_intervals['total_time'] < duration_threshold_min), 1, 0)\n",
    "\n",
    "group_intervals['invalid_load_duration_long'] = np.where(\n",
    "    (group_intervals['total_time'] > duration_threshold_max), 1, 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_duration_short\n",
       "0    1887\n",
       "1      53\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# You can use the COUNT function along with a GROUP BY clause to count the occurrences of each value in the invalid_load_duration_short column. \n",
    "\n",
    "# SELECT \n",
    "#     invalid_load_duration_short,\n",
    "#     COUNT(*) AS count\n",
    "# FROM \n",
    "#     group_intervals\n",
    "# GROUP BY \n",
    "#     invalid_load_duration_short;\n",
    "\n",
    "group_intervals['invalid_load_duration_short'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_duration_long\n",
       "0    1920\n",
       "1      20\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "group_intervals['invalid_load_duration_long'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# flag any groups where the point count is either too low or too high... 1 means the group \n",
    "# is positive for falling outside the threshold, 0 means group is within valid range\n",
    "group_intervals['invalid_load_point_count_low'] = np.where(\n",
    "    (group_intervals['total_points'] < point_threshold_min), 1, 0)\n",
    "\n",
    "group_intervals['invalid_load_point_count_high'] = np.where(\n",
    "    (group_intervals['total_points'] > point_threshold_max), 1, 0)\n",
    "\n",
    "\n",
    "# defined above ^\n",
    "# duration_threshold_min = pd.to_timedelta('0 days 01:00:00') # 1 hour\n",
    "# duration_threshold_max = pd.to_timedelta('1 days 00:00:00') # 1 day"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_point_count_low\n",
       "0    1900\n",
       "1      40\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "group_intervals['invalid_load_point_count_low'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_point_count_high\n",
       "0    1906\n",
       "1      34\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "group_intervals['invalid_load_point_count_high'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_reporting_gap\n",
       "0    1938\n",
       "1       2\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# flag groups with long durations between points\n",
    "group_intervals['invalid_load_reporting_gap'] = np.where(\n",
    "    group_intervals['time_delta_avg'] > load_gap_threshold, 1, 0)\n",
    "\n",
    "group_intervals['invalid_load_reporting_gap'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_ping_rate\n",
       "0    1914\n",
       "1      26\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# check if the estimated ping rate and the actual ping rates are within a threshold of each other\n",
    "def flag_invalid_ping_rates(row, ping_rate_threshold):\n",
    "    est_ping_rate = row['total_time'].total_seconds() / row['total_points']\n",
    "    time_delta_seconds = row['time_delta_avg'].total_seconds()\n",
    "    if abs(est_ping_rate - time_delta_seconds) > ping_rate_threshold:\n",
    "        return 1\n",
    "    else:\n",
    "        return 0\n",
    "\n",
    "group_intervals['invalid_load_ping_rate'] = group_intervals.apply(\n",
    "    flag_invalid_ping_rates, args=(ping_rate_threshold,), axis=1)\n",
    "\n",
    "group_intervals['invalid_load_ping_rate'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_duplicate\n",
       "0    1940\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# look for load numbers that are duplicated on a contract\n",
    "def flag_duplicated_load_numbers(df):\n",
    "    mask = df['load'].duplicated(keep=False)\n",
    "    df.loc[mask, 'invalid_duplicate_loads'] = 1\n",
    "    return df\n",
    "\n",
    "group_intervals['invalid_load_duplicate'] = 0\n",
    "group_intervals = group_intervals.groupby('contract').apply(flag_duplicated_load_numbers)\n",
    "group_intervals = group_intervals.reset_index(drop=True)\n",
    "\n",
    "group_intervals['invalid_load_duplicate'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_overlap\n",
       "0    1940\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# check if any load groups within a contract overlap one another\n",
    "def check_overlaps(df):\n",
    "    for i in df.itertuples():\n",
    "        for j in df.itertuples():\n",
    "            if i.Index != j.Index and i.interval.overlaps(j.interval):\n",
    "                df.loc[i.Index, 'invalid_load_overlap'] = 1\n",
    "                print(f\"Contract {i.contract}: Load {i.load} overlaps {j.load}\")\n",
    "    return df\n",
    "\n",
    "group_intervals['invalid_load_overlap'] = 0\n",
    "group_intervals = group_intervals.groupby('contract').apply(check_overlaps)\n",
    "group_intervals = group_intervals.reset_index(drop=True)\n",
    "\n",
    "group_intervals['invalid_load_overlap'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_noncontiguous\n",
       "0    1913\n",
       "1      27\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# look for non-contiguous loads\n",
    "group_intervals = group_intervals.sort_values(['contract', 'interval'])\n",
    "\n",
    "group_intervals['time_diff'] = abs(group_intervals['msg_time_min'] - \n",
    "                                            group_intervals.groupby('contract')['msg_time_max'].shift(1))\n",
    "\n",
    "group_intervals['invalid_load_noncontiguous'] = np.where(group_intervals['time_diff'] > load_gap_threshold, 1, 0)\n",
    "\n",
    "group_intervals['invalid_load_noncontiguous'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load_duplicate\n",
       "0    1940\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# look for load numbers that are duplicated on a contract\n",
    "def flag_duplicated_load_numbers(df):\n",
    "    mask = df['load'].duplicated(keep=False)\n",
    "    df.loc[mask, 'invalid_duplicate_loads'] = 1\n",
    "    return df\n",
    "\n",
    "group_intervals['invalid_load_duplicate'] = 0\n",
    "group_intervals = group_intervals.groupby('contract').apply(flag_duplicated_load_numbers)\n",
    "group_intervals = group_intervals.reset_index(drop=True)\n",
    "\n",
    "group_intervals['invalid_load_duplicate'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "invalid_load\n",
       "0    1829\n",
       "1     111\n",
       "Name: count, dtype: int64"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Flag loads where ANY of the outlier flags are positive (=1)\n",
    "outlier_columns = ['invalid_load_duration_short', 'invalid_load_duration_long',\n",
    "                   'invalid_load_point_count_low', 'invalid_load_point_count_high',\n",
    "                   'invalid_load_reporting_gap', 'invalid_load_ping_rate', \n",
    "                   'invalid_load_overlap', 'invalid_load_noncontiguous', 'invalid_load_duplicate']\n",
    "\n",
    "group_intervals['invalid_load'] = group_intervals[outlier_columns].any(axis=1).astype(int)\n",
    "\n",
    "group_intervals['invalid_load'].value_counts()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Rejoin Intervals with Outlier Flags to Original Data (map intervals back to points)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# NOTE: this should be easier to do with a pd.cut() but their was a conflict in the time formats\n",
    "#       between points and intervals. Need to troubleshoot that...\n",
    "\n",
    "# create a key field shared between the two data frames\n",
    "group_intervals['msg_time'] = group_intervals['msg_time_min']\n",
    "\n",
    "# pd.merge_asof requires the data to be sorted\n",
    "orig_df = orig_df.sort_values('msg_time')\n",
    "group_intervals = group_intervals.sort_values('msg_time')\n",
    "\n",
    "# merge only the needed fields into the original dataframe. \n",
    "# NOTE: pd.merge_asof() does a \"closest time match\" and be default only matches \"backwards\", so \n",
    "#       using the minimum time of each load should work out just fine.\n",
    "group_intervals_out = group_intervals[['msg_time', \n",
    "                                       'invalid_load_duration_short', 'invalid_load_duration_long',\n",
    "                                       'invalid_load_point_count_low', 'invalid_load_point_count_high',\n",
    "                                       'invalid_load_reporting_gap', 'invalid_load_ping_rate', \n",
    "                                       'invalid_load_overlap', 'invalid_load_noncontiguous', \n",
    "                                       'invalid_load_duplicate', 'invalid_load']]\n",
    " \n",
    "orig_df_out = pd.merge_asof(orig_df, group_intervals_out, on='msg_time')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# write the df to CSV with the repaired load numbers, drop the columns we created\n",
    "orig_df_out.drop(columns=['group_number', 'time_delta']).to_csv(output_csv_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "geopandas",
   "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.8.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}