tool_Processing_Location_Features_for_Directory.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please review the python script attached. it was developed in JupyterNotebook, and functions very close to expected.\n",
    "I would like to ensure Idempotency of these processing components, namely I would like to convert this into a BigQuery UDF / Cloud Function that can take in raw_data (timestamp_unix, device_id, latitude, longitude) and execute the functions.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYDZYE7T9BGASTZFT0X3C6CH",
   "metadata": {},
   "outputs": [],
   "source": [
    "# imports\n",
    "import os\n",
    "import glob\n",
    "\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import pyarrow as pa\n",
    "import pyarrow.parquet as pq\n",
    "import geopandas as gpd\n",
    "import movingpandas as mpd\n",
    "\n",
    "from datetime import date\n",
    "from datetime import datetime\n",
    "\n",
    "from tzfpy import get_tz\n",
    "from zoneinfo import ZoneInfo\n",
    "\n",
    "from gpxcsv import gpxtolist\n",
    "import h3\n",
    "import xml.etree.ElementTree as ET\n",
    "\n",
    "import folium\n",
    "from IPython.display import display, HTML\n",
    "import holoviews as hv\n",
    "from holoviews import dim, opts\n",
    "import hvplot.pandas\n",
    "import plotly.express as px\n",
    "from keplergl import KeplerGl\n",
    "\n",
    "import skmob\n",
    "from skmob.preprocessing import detection, clustering, compression, filtering\n",
    "from skmob.measures.individual import home_location\n",
    "from skmob.tessellation import tilers\n",
    "\n",
    "from itables import init_notebook_mode, show\n",
    "from dataprep.eda import create_report\n",
    "from dataprep.eda import plot\n",
    "from dataprep.eda import plot_diff\n",
    "\n",
    "from google.cloud import bigquery\n",
    "from google.api_core import exceptions\n",
    "\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "# notebook extensions\n",
    "init_notebook_mode(all_interactive=False)\n",
    "hv.extension()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYDZY46ZZ2RPEM0WVJ8T7JR0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# gcp-project STAGE 1 \n",
    "\n",
    "# generate datetime_local, filter & compress GPS data, detect stay_locations\n",
    "\n",
    "from stage1_mobility.compression import argcompress_trajectory\n",
    "from stage1_mobility.detection import (\n",
    "    detect_stay_locations,\n",
    "    extract_stay_locations,\n",
    ")\n",
    "from stage1_mobility.filtering import argfilter_trajectory\n",
    "from stage1_mobility.sorting import (\n",
    "    argsort_trajectory_by_time_ascending,\n",
    ")\n",
    "\n",
    "def array_to_str(array: np.array) -> str:\n",
    "    \"\"\"\n",
    "    Converts a numpy array to a string.\n",
    "\n",
    "    To save on storage, superfluous whitespaces and decimal points are removed.\n",
    "\n",
    "    :param array: an input array\n",
    "    :return: string representation of stay location array\n",
    "    \"\"\"\n",
    "    return (\n",
    "        str(array.tolist())\n",
    "        .replace(\" \", \"\")  # get rid of whitespaces to reduce length\n",
    "        .replace(\".0,\", \",\")  # convert from float to int\n",
    "        .replace(\".0]\", \"]\")  # convert from float to int\n",
    "        .replace(\"nan\", \"NaN\")  # convert into a json-readable format\n",
    "        .replace(\"'\", '\"')  # convert single-quotation marks to double quotation marks\n",
    "    )\n",
    "    \n",
    "# ... convert timestamp_utc to local timezone\n",
    "def convert_timestamps_to_datetimes_local(lat_lon_ts: np.array) -> list[str]:\n",
    "    \"\"\"\n",
    "    Converts timestamps to local time zone based on lat/lon location.\n",
    "\n",
    "    :param lat_lon_ts: an (n, 3) numpy array with (lat, lon) location, its timestamp:\n",
    "    - 0th column is latitude,\n",
    "    - 1st column is longitude,\n",
    "    - 2nd column is observation unix timestamp,\n",
    "    where n is the number of observations.\n",
    "\n",
    "    :return: list of timestamps in local timezone\n",
    "    \"\"\"\n",
    "    return [convert_timestamp_to_datetime_local(*lat_lon_ts_row) for lat_lon_ts_row in lat_lon_ts]\n",
    "\n",
    "\n",
    "# ... convert timestamps to local based on lat/lon location\n",
    "def convert_timestamp_to_datetime_local(lat: float, lon: float, ts: float) -> str:\n",
    "    \"\"\"\n",
    "    Converts timestamps to local based on lat/lon location.\n",
    "\n",
    "    :param lat: latitude\n",
    "    :param lon: longitude\n",
    "    :param ts: observation unix timestamp\n",
    "\n",
    "    :return: timestamp in local timezone; 0 if timestamp couldn't be converted between timezones\n",
    "    \"\"\"\n",
    "    date_utc = datetime.fromtimestamp(ts).astimezone(ZoneInfo(\"UTC\"))\n",
    "\n",
    "    new_tz = ZoneInfo(get_tz(lon, lat))\n",
    "    return date_utc.astimezone(new_tz).strftime(\"%Y-%m-%d %H:%M:%S\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYDZXNJS1S02ZHDS9CW60AWA",
   "metadata": {},
   "outputs": [],
   "source": [
    "# load files\n",
    "\n",
    "def rename_latitude_longitude(df):\n",
    "    longitude_cols = ['longitude', 'lon', 'long', 'lng']\n",
    "    latitude_cols = ['latitude', 'lat', 'Lat']\n",
    "\n",
    "    for col in latitude_cols:\n",
    "        if col in df.columns:\n",
    "            df.rename(columns={col: 'latitude'}, inplace=True)\n",
    "\n",
    "    for col in longitude_cols:\n",
    "        if col in df.columns:\n",
    "            df.rename(columns={col: 'longitude'}, inplace=True)\n",
    "\n",
    "    # Print message only if no renaming occurred\n",
    "    if 'longitude' not in df.columns or 'latitude' not in df.columns:\n",
    "        print(\"Warning: Could not find longitude/latitude columns in the DataFrame.\")\n",
    "\n",
    "    return df\n",
    "\n",
    "#-------------------------------------------LOAD GPX, CSV, BIGQUERY or GCS DATA-------------------------------------------------------\n",
    "def gpx_rzr_load(gpx_file):\n",
    "    gpx_rzr_df = pd.DataFrame(gpxtolist(gpx_file))\n",
    "    gpx_rzr_df = rename_latitude_longitude(gpx_rzr_df) \n",
    "    gpx_rzr_df.rename(columns={'rcid': 'device_id', 'type': 'trajectory_id_part1', 'name': 'trajectory_id_part2'}, inplace=True)\n",
    "    gpx_rzr_df['source'] = os.path.basename(gpx_file)\n",
    "    gpx_rzr_df['timestamp_utc'] = pd.to_datetime(gpx_rzr_df['time'])\n",
    "    gpx_rzr_df['timestamp_utc'] = gpx_rzr_df['timestamp_utc'].dt.tz_convert('UTC')\n",
    "    gpx_rzr_df['timestamp_unix'] = gpx_rzr_df['timestamp_utc'].astype('int64') // 10**9\n",
    "    drop_cols = [\"appSku\", \"id\", \"appVersion\", \"time\", \"color\", \"Color\", \"totalDistanceInMeters\", \"totalDurationInSeconds\"]\n",
    "\n",
    "    # optional handle Polaris-specific columns\n",
    "    polaris_cols = [\"averageSpeed\", \"maxSpeed\", \"stoppedTimeInSeconds\"]\n",
    "    drop_cols.extend(col for col in polaris_cols if col in gpx_rzr_df.columns)\n",
    "    gpx_rzr_df.drop(drop_cols, axis=1, inplace=True)\n",
    "    gpx_rzr_df['latitude'] = round(gpx_rzr_df['latitude'], 5)\n",
    "    gpx_rzr_df['longitude'] = round(gpx_rzr_df['longitude'], 5)\n",
    "    gpx_rzr_df = gpx_rzr_df.sort_values('timestamp_unix', ascending=True)\n",
    "    gpx_rzr_df['device_id'] = DEVICE_ID\n",
    "    return gpx_rzr_df[['device_id', 'latitude', 'longitude', 'timestamp_utc', 'timestamp_unix' , 'source']]\n",
    "\n",
    "\n",
    "def gpx_ios_load(gpx_file):\n",
    "    xml_data = open(gpx_file, 'r').read()\n",
    "    root = ET.fromstring(xml_data)\n",
    "\n",
    "    namespace = '{http://www.topografix.com/GPX/1/1}'\n",
    "    trkpt_elements = root.findall(f'.//{namespace}trkpt')\n",
    "\n",
    "    data = []\n",
    "    for trkpt in trkpt_elements:\n",
    "        latitude = float(trkpt.attrib['lat'])\n",
    "        longitude = float(trkpt.attrib['lon'])\n",
    "        timestamp_utc = pd.to_datetime(trkpt.find(f'{namespace}time').text)\n",
    "        timestamp_unix = int(timestamp_utc.timestamp())\n",
    "\n",
    "        data.append({\n",
    "            'latitude': latitude,\n",
    "            'longitude': longitude,\n",
    "            'timestamp_utc': timestamp_utc,\n",
    "            'timestamp_unix': timestamp_unix\n",
    "        })\n",
    "\n",
    "    gpx_ios_df = pd.DataFrame(data)\n",
    "    gpx_ios_df['source'] = os.path.basename(gpx_file)\n",
    "    gpx_ios_df['device_id'] = DEVICE_ID\n",
    "    gpx_ios_df['latitude'] = round(gpx_ios_df['latitude'], 5)\n",
    "    gpx_ios_df['longitude'] = round(gpx_ios_df['longitude'], 5)\n",
    "    gpx_ios_df = gpx_ios_df.sort_values('timestamp_unix', ascending=True)\n",
    "    return gpx_ios_df[['device_id', 'latitude', 'longitude', 'timestamp_utc', 'timestamp_unix' , 'source']]\n",
    "\n",
    "def csv_load(csv_file):\n",
    "    csv_df = pd.read_csv(csv_file)\n",
    "    csv_df = rename_latitude_longitude(csv_df)\n",
    "    csv_df['timestamp_utc'] = pd.to_datetime(csv_df['timestamp_utc'])\n",
    "    csv_df['timestamp_unix'] = pd.to_datetime(csv_df['timestamp_utc']).astype('int64') // 10**9\n",
    "    csv_df['source'] = os.path.basename(csv_file)\n",
    "    csv_df['latitude'] = round(csv_df['latitude'], 5)\n",
    "    csv_df['longitude'] = round(csv_df['longitude'], 5)\n",
    "    csv_df = csv_df.sort_values('timestamp_unix', ascending=True)\n",
    "    if 'device_id' not in csv_df.columns:\n",
    "        csv_df['device_id'] = DEVICE_ID\n",
    "    return csv_df[['device_id', 'latitude', 'longitude', 'timestamp_utc', 'timestamp_unix' , 'source']]\n",
    "\n",
    "def bigquery_load(query):\n",
    "    # ...  (logic to load data from BigQuery)\n",
    "    # ...\n",
    "    # return bigquery_df[['device_id', 'latitude', 'longitude', 'timestamp_utc', 'timestamp_unix' , 'source']]\n",
    "    pass\n",
    "\n",
    "def gcs_bucket_load(bucket_name, file_path):\n",
    "    # ...  (logic to load data from GCS bucket)\n",
    "    # ...\n",
    "    # return gcs_bucket_df[['device_id', 'latitude', 'longitude', 'timestamp_utc', 'timestamp_unix' , 'source']]\n",
    "    pass\n",
    "\n",
    "def ingest_raw_data(filepath, file_type):\n",
    "    if file_type == 'gpx_rzr':\n",
    "        df = gpx_rzr_load(filepath)\n",
    "    elif file_type == 'gpx_ios':\n",
    "        df = gpx_ios_load(filepath)\n",
    "    elif file_type == 'csv':\n",
    "        df = csv_load(filepath)\n",
    "    # elif file_type == 'bigquery':\n",
    "    #     df = bigquery_load(query)  # Replace with actual BigQuery logic\n",
    "    # elif file_type == 'gcs':\n",
    "    #     df = gcs_bucket_load(bucket_name, file_path)  # Replace with actual GCS logic\n",
    "    \n",
    "    else:\n",
    "        raise ValueError(\"Unsupported file type:\", file_type)\n",
    "\n",
    "    if df is not None and not df.empty:\n",
    "        location_data = df.copy()\n",
    "    else:\n",
    "        location_data = pd.read_csv(filepath) #fallback to CSV\n",
    "\n",
    "    return location_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYE03PJTB68PWX2JEYG24D6W",
   "metadata": {},
   "outputs": [],
   "source": [
    "def stage1_preprocess_all_in_one(raw_df):\n",
    "     # to_numpy array\n",
    "    lat_lon_df = raw_df[['latitude', 'longitude', 'timestamp_unix']]\n",
    "    num_attrs_df = raw_df[[]]\n",
    "    str_attrs_df = raw_df[['device_id', 'timestamp_utc', 'source']]\n",
    "\n",
    "    lat_lon_ts = lat_lon_df.to_numpy()\n",
    "    num_attrs = num_attrs_df.to_numpy()\n",
    "    str_attrs = str_attrs_df.to_numpy()\n",
    "\n",
    "    # Sort trajectory\n",
    "    indices_sorted = argsort_trajectory_by_time_ascending(lat_lon_ts)\n",
    "    lat_lon_ts_index_sorted = np.hstack((lat_lon_ts[indices_sorted], np.arange(0, lat_lon_ts.shape[0]).reshape(-1, 1)))\n",
    "    num_attrs_sorted = num_attrs[indices_sorted, :]\n",
    "    str_attrs_sorted = str_attrs[indices_sorted, :]\n",
    "\n",
    "    # Filtering > 400 km/h\n",
    "    indices_filtered = argfilter_trajectory(lat_lon_ts_index_sorted, speed_limit_in_kms=400 / 3600).astype(int)\n",
    "    lat_lon_ts_filtered = lat_lon_ts_index_sorted[indices_filtered, :3]\n",
    "    num_attrs_filtered = num_attrs_sorted[indices_filtered, :]\n",
    "    str_attrs_filtered = str_attrs_sorted[indices_filtered, :]\n",
    "\n",
    "    # Compression (0.01 to 0.05 km range) for location_data\n",
    "    location_data_indices_compressed = argcompress_trajectory(lat_lon_ts_filtered, compression_range_in_km=0.01).astype(int)[0]\n",
    "\n",
    "    location_data_lat_lon_ts_compressed = lat_lon_ts_filtered[location_data_indices_compressed, :]\n",
    "    location_data_num_attrs_compressed = num_attrs_filtered[location_data_indices_compressed, :]\n",
    "    location_data_str_attrs_compressed = str_attrs_filtered[location_data_indices_compressed, :]\n",
    "\n",
    "    # convert compressed numpy arrays to pandas dataframe\n",
    "    datetime_local = convert_timestamps_to_datetimes_local(location_data_lat_lon_ts_compressed)\n",
    "\n",
    "    compressed_lat_lon_df = pd.DataFrame(location_data_lat_lon_ts_compressed)\n",
    "    datetime_local_df = pd.DataFrame(datetime_local)\n",
    "    compressed_num_data_df = pd.DataFrame(location_data_num_attrs_compressed)\n",
    "    compressed_str_data_df = pd.DataFrame(location_data_str_attrs_compressed)\n",
    "\n",
    "    compressed_lat_lon_df.columns = ['latitude', 'longitude', 'timestamp_unix']\n",
    "    datetime_local_df.columns = ['datetime_local']\n",
    "    compressed_num_data_df.columns = []\n",
    "    compressed_str_data_df.columns = ['device_id', 'timestamp_utc', 'source']\n",
    "    compressed_location_data = pd.concat([compressed_lat_lon_df, datetime_local_df, compressed_num_data_df, compressed_str_data_df], axis=1)\n",
    "\n",
    "    compressed_location_data['timestamp_utc'] = compressed_location_data['timestamp_unix'].apply(lambda x: datetime.fromtimestamp(x).astimezone(ZoneInfo(\"UTC\")))\n",
    "    compressed_location_data['datetime_local'] = pd.to_datetime(compressed_location_data['datetime_local'])\n",
    "    compressed_location_data['trajectory_id'] = compressed_location_data.apply(lambda x: f\"{x['datetime_local'].strftime('%Y%m%d')}#{x['device_id']}\", axis=1)\n",
    "    compressed_location_data['trajectory_id'] = compressed_location_data.apply(lambda x: f\"#{x['device_id']}#{x['datetime_local'].strftime('%Y%m%d')}\", axis=1)\n",
    "    compressed_location_data['timezone'] = compressed_location_data.apply(lambda row: get_tz(row['longitude'], row['latitude']), axis=1)\n",
    "    compressed_location_data['last_modified_on'] = TODAY\n",
    "\n",
    "    compressed_location_data = compressed_location_data[['device_id', 'latitude', 'longitude', 'datetime_local', 'timestamp_utc', 'timestamp_unix', 'timezone', 'trajectory_id', 'source', 'last_modified_on']]\n",
    "    compressed_location_data = compressed_location_data.sort_values('datetime_local', ascending=True)\n",
    "    \n",
    "    # stay_locations - compression & extraction\n",
    "    stay_location_indices_compressed = argcompress_trajectory(lat_lon_ts_filtered, compression_range_in_km=0.001).astype(int)[0]\n",
    "    stay_location_lat_lon_ts_compressed = lat_lon_ts_filtered[stay_location_indices_compressed, :]\n",
    "    stay_location_num_attrs_compressed = num_attrs_filtered[stay_location_indices_compressed, :]\n",
    "    stay_location_str_attrs_compressed = str_attrs_filtered[stay_location_indices_compressed, :]\n",
    "\n",
    "    # detect & extract stay_locations_data\n",
    "    start_stop_indices = detect_stay_locations(\n",
    "        stay_location_lat_lon_ts_compressed, stay_range_in_km=0.05, stay_duration_in_s=1200\n",
    "    ).astype(int)[0]\n",
    "\n",
    "    stay_locations = extract_stay_locations(\n",
    "        stay_location_lat_lon_ts_compressed, stay_location_num_attrs_compressed, stay_location_str_attrs_compressed, start_stop_indices\n",
    "    )\n",
    "\n",
    "    stay_locations_data = pd.DataFrame(stay_locations)\n",
    "    stay_locations_data.columns = ['stay_start_unix', 'stay_latitude_ctr', 'stay_longitude_ctr', 'stay_end_unix', 'stay_num_points', 'stay_max_diameter', 'speed', 'altitude', 'horizontal_accuracy', 'vertical_accuracy', 'ipv4', 'ipv6', 'bssids', 'ssids']\n",
    "    stay_locations_data.drop(['horizontal_accuracy', 'vertical_accuracy', 'ipv4', 'ipv6', 'bssids', 'ssids'], axis=1, inplace=True)\n",
    "    stay_locations_data['device_id'] = DEVICE_ID\n",
    "    stay_locations_data['last_modified_on'] = TODAY\n",
    "    \n",
    "    stay_locations_data = stay_locations_data[['device_id', 'stay_latitude_ctr', 'stay_longitude_ctr', 'stay_start_unix', 'stay_end_unix',\n",
    "                                               'stay_num_points', 'stay_max_diameter','last_modified_on']]\n",
    "    stay_locations_data = stay_locations_data.sort_values('stay_start_unix', ascending=True)\n",
    "    \n",
    "    return compressed_location_data, stay_locations_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYE7RME12CAM4XK27SJ0DBQB",
   "metadata": {},
   "outputs": [],
   "source": [
    "# feature engineering\n",
    "\n",
    "def get_period_of_day(timestamp):\n",
    "    hour = timestamp.hour\n",
    "    if 0 <= hour < 6:\n",
    "        return 'early_am'\n",
    "    elif 6 <= hour < 12:\n",
    "        return 'morning'\n",
    "    elif 12 <= hour < 13:\n",
    "        return 'lunch'\n",
    "    elif 13 <= hour < 18:\n",
    "        return 'afternoon'\n",
    "    else:\n",
    "        return 'evening'\n",
    "    \n",
    "def determine_move_activity(row, stay_locations):\n",
    "    device_id = row['device_id']\n",
    "    timestamp_unix = row['timestamp_unix']\n",
    "    \n",
    "    device_stay_locations = stay_locations[stay_locations['device_id'] == device_id]\n",
    "    \n",
    "    # Check if timestamp_unix is between stay_start_unix and stay_end_unix\n",
    "    is_stay = ((device_stay_locations['stay_start_unix'] <= timestamp_unix) & \n",
    "               (device_stay_locations['stay_end_unix'] >= timestamp_unix)).any()\n",
    "    \n",
    "    return 'stay' if is_stay else 'trip'\n",
    "\n",
    "def determine_stay_activity(row, stay_locations):\n",
    "    device_id = row['device_id']\n",
    "    timestamp_unix = row['timestamp_unix']\n",
    "    \n",
    "    device_stay_locations = stay_locations[stay_locations['device_id'] == device_id]\n",
    "    \n",
    "    for _, stay_location in device_stay_locations.iterrows():\n",
    "        stay_start_unix = stay_location['stay_start_unix']\n",
    "        stay_end_unix = stay_location['stay_end_unix']\n",
    "        stay_duration = stay_end_unix - stay_start_unix\n",
    "        \n",
    "        if stay_start_unix <= timestamp_unix <= stay_end_unix:\n",
    "            if timestamp_unix <= stay_start_unix + 0.2 * stay_duration:\n",
    "                return 'arriving'\n",
    "            elif timestamp_unix >= stay_end_unix - 0.2 * stay_duration:\n",
    "                return 'departing'\n",
    "            else:\n",
    "                return 'stopped'\n",
    "    \n",
    "    return None\n",
    "\n",
    "def process_spatiotemporal_features(location_data, stay_locations_data):\n",
    "    # Convert timestamp_utc to datetime_local for location_data\n",
    "    lat_lon_df = location_data[['latitude', 'longitude', 'timestamp_unix']]\n",
    "    lat_lon_ts = lat_lon_df.to_numpy()\n",
    "    location_data['datetime_local'] = convert_timestamps_to_datetimes_local(lat_lon_ts)\n",
    "    location_data['datetime_local'] = pd.to_datetime(location_data['datetime_local'])\n",
    "    \n",
    "    # Convert timestamp_utc to datetime_local for stay_locations_data\n",
    "    stay_lat_lon_df = stay_locations_data[['stay_latitude_ctr', 'stay_longitude_ctr', 'stay_start_unix']]\n",
    "    stay_lat_lon_ts = stay_lat_lon_df.to_numpy()\n",
    "    stay_locations_data['datetime_local'] = convert_timestamps_to_datetimes_local(stay_lat_lon_ts)\n",
    "    stay_locations_data['datetime_local'] = pd.to_datetime(stay_locations_data['datetime_local'])\n",
    "\n",
    "    \n",
    "    #process temporal features\n",
    "    location_data['min_of_day'] = location_data['datetime_local'].dt.hour * 60 + location_data['datetime_local'].dt.minute\n",
    "    location_data['hour_of_day'] = location_data['datetime_local'].dt.hour\n",
    "    location_data['period_of_day'] = location_data['datetime_local'].apply(get_period_of_day)\n",
    "    location_data['date'] = location_data['datetime_local'].dt.date\n",
    "    location_data['time_local'] = location_data['datetime_local'].dt.strftime('%I:%M %p')\n",
    "    location_data['day_of_month'] = location_data['datetime_local'].dt.day\n",
    "    location_data['day_of_year'] = location_data['datetime_local'].dt.day_of_year\n",
    "    location_data['day_of_week'] = location_data['datetime_local'].dt.day_of_week\n",
    "    location_data['day_of_week_name'] = location_data['datetime_local'].dt.day_name()\n",
    "    location_data['is_workday'] = location_data['datetime_local'].dt.day_of_week.between(0, 4)\n",
    "    location_data['is_weekend'] = location_data['datetime_local'].dt.weekday >= 5\n",
    "    location_data['is_business_hours'] = location_data['datetime_local'].dt.hour.between(9, 16)\n",
    "    location_data['month_name'] = location_data['datetime_local'].dt.month_name()\n",
    "    location_data['month'] = location_data['datetime_local'].dt.month\n",
    "    location_data['quarter'] = location_data['datetime_local'].dt.quarter\n",
    "    location_data['datetime_index'] = location_data['datetime_local'].dt.tz_localize(None, ambiguous=\"infer\", nonexistent='raise')\n",
    "    \n",
    "    #spatial pre-processing features\n",
    "    location_data['h3_lvl10_index'] = location_data.apply(lambda row: h3.geo_to_h3(row['latitude'], row['longitude'], 10), axis=1)\n",
    "    location_data['h3_lvl4_index'] = location_data.apply(lambda row: h3.geo_to_h3(row['latitude'], row['longitude'], 4), axis=1)\n",
    "    location_data['altitude1_minOverlap'] = (location_data['day_of_month'] * 240) + (location_data['min_of_day'])\n",
    "    location_data['altitude2_hourOverlap'] = (location_data['day_of_month'] * 240) + (location_data['hour_of_day'] * 10)\n",
    "    \n",
    "    min_datetime = location_data['datetime_local'].min()\n",
    "    max_datetime = location_data['datetime_local'].max()\n",
    "    num_intervals = location_data.shape[0]\n",
    "    interval_size = (max_datetime - min_datetime) / (num_intervals - 1)\n",
    "    location_data['altitude3_min2max'] = ((location_data['datetime_local'] - min_datetime) / interval_size).astype(int) * 100\n",
    "    max_value = location_data['altitude3_min2max'].max()\n",
    "    location_data['altitude3_min2max'] = location_data['altitude3_min2max'] * 10000 / max_value\n",
    "    \n",
    "    #move_activity & stay_activity features\n",
    "    location_data['move_activity'] = location_data.apply(lambda row: determine_move_activity(row, stay_locations_data), axis=1)\n",
    "    location_data['stay_activity'] = location_data.apply(lambda row: determine_stay_activity(row, stay_locations_data), axis=1)\n",
    "    location_data['last_modified_on'] = TODAY\n",
    "    \n",
    "    #location_data.drop([], axis=1, inplace=True)\n",
    "    \n",
    "    #formatting output dataframe columns\n",
    "    first_columns = ['device_id', 'latitude', 'longitude', 'datetime_local']\n",
    "    last_columns = ['move_activity','stay_activity','altitude1_minOverlap','altitude2_hourOverlap','altitude3_min2max','h3_lvl10_index','h3_lvl4_index','timestamp_unix','timestamp_utc','timezone','trajectory_id','source','last_modified_on']\n",
    "    other_columns = [col for col in location_data.columns if col not in (first_columns + last_columns)]\n",
    "    ordered_output = first_columns + other_columns + last_columns\n",
    "    \n",
    "    location_features_output = location_data[ordered_output]\n",
    "    location_features_output.sort_values(by='datetime_local', inplace=True, ascending=True)\n",
    "    \n",
    "    location_features_output.to_csv(ANALYTIC_LOCATION_FEATURES, index=False)\n",
    "    #location_features_output.to_parquet()\n",
    "    \n",
    "    return location_features_output\n",
    "\n",
    "#-------------------------------------PREPROCESSING STAY_LOCATION FEATURES----------------------------------------------------------------------------------\n",
    "def process_stay_location_features(stay_locations_data, DEVICE_ID):\n",
    "    # Check if stay_locations_data has any data\n",
    "    if len(stay_locations_data) <= 0:\n",
    "        return display(f\"No stay locations detected in device_id: {DEVICE_ID}'s data\")\n",
    "    \n",
    "    #convert timestamp_utc to datetime_local \n",
    "    if 'stay_start_datetime_local' not in stay_locations_data.columns and 'stay_end_datetime_local' not in stay_locations_data.columns:\n",
    "        lat_lon_start_df = stay_locations_data[['stay_latitude_ctr', 'stay_longitude_ctr', 'stay_start_unix']]\n",
    "        lat_lon_start_ts = lat_lon_start_df.to_numpy()\n",
    "        lat_lon_end_df = stay_locations_data[['stay_latitude_ctr', 'stay_longitude_ctr','stay_end_unix']]\n",
    "        lat_lon_end_ts = lat_lon_end_df.to_numpy()\n",
    "        stay_locations_data['stay_start_datetime_local'] = convert_timestamps_to_datetimes_local(lat_lon_start_ts)\n",
    "        stay_locations_data['stay_start_datetime_local'] = pd.to_datetime(stay_locations_data['stay_start_datetime_local'])\n",
    "        stay_locations_data['stay_end_datetime_local'] = convert_timestamps_to_datetimes_local(lat_lon_end_ts)\n",
    "        stay_locations_data['stay_end_datetime_local'] = pd.to_datetime(stay_locations_data['stay_end_datetime_local'])\n",
    "    else:\n",
    "        pass\n",
    "    \n",
    "    stay_locations_data['stay_duration_min'] = np.floor((stay_locations_data['stay_end_unix'] - stay_locations_data['stay_start_unix'])/60)\n",
    "    stay_locations_data['stay_of_day'] = stay_locations_data['stay_start_unix'].rank(method='dense').astype(int)\n",
    "    stay_locations_data['stay_id'] = stay_locations_data.apply(lambda x: f\"#{x['device_id']}#{x['stay_start_datetime_local'].strftime('%Y%m%d')}#{x['stay_of_day']:04d}\", axis=1)\n",
    "    \n",
    "    stay_locations_data['last_modified_on'] = TODAY\n",
    "    \n",
    "    #formatting the output dataframe\n",
    "    stay_locations_features = stay_locations_data[['device_id','stay_latitude_ctr','stay_longitude_ctr','stay_start_unix','stay_start_datetime_local','stay_end_unix','stay_end_datetime_local',\n",
    "                                           'stay_duration_min','stay_num_points','stay_max_diameter','stay_of_day','stay_id','last_modified_on']]\n",
    "    \n",
    "    return stay_locations_features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYE2MXDW7B8A1CWKD5AR6WKA",
   "metadata": {},
   "outputs": [],
   "source": [
    "#stack all daily files & append into CSV & BQ tables\n",
    "def stack_all_COMPRESSED_location_data(DATA_DIR, file_pattern, output_csv, output_parquet):\n",
    "    files = glob.glob(os.path.join(DATA_DIR, file_pattern))\n",
    "\n",
    "    merged_compressed_location_data = []\n",
    "    processed_compressed_location_data = []  # List to store processed file names\n",
    "    \n",
    "    for file in files:\n",
    "        if file.endswith('.csv') and file not in processed_compressed_location_data:\n",
    "            df = pd.read_csv(file)\n",
    "            merged_compressed_location_data.append(df)\n",
    "            processed_compressed_location_data.append(file)  # Add the processed file name to the list\n",
    "        \n",
    "    all_compressed_data_for_export = pd.concat(merged_compressed_location_data, ignore_index=True)\n",
    "    all_compressed_data_for_export = all_compressed_data_for_export.sort_values('timestamp_unix', ascending=True)\n",
    "\n",
    "    # Check if the file exists and delete it\n",
    "    if os.path.exists(output_csv):\n",
    "        os.remove(output_csv)\n",
    "\n",
    "    if os.path.exists(output_parquet):\n",
    "        os.remove(output_parquet)\n",
    "        \n",
    "    all_compressed_data_for_export.to_csv(output_csv, index=False)\n",
    "    all_compressed_data_for_export.to_parquet(output_parquet, index=False)\n",
    "\n",
    "    # export to BigQuery - COMPRESSED_location_features\n",
    "    client = bigquery.Client()\n",
    "    dataset_id = \"compressed_location_data\"\n",
    "    table_id = \"JC_2024_rzr_compressed_location_data\"\n",
    "    table_ref = client.dataset(dataset_id).table(table_id)\n",
    "\n",
    "    # Check if the table exists\n",
    "    try:\n",
    "        client.get_table(table_ref)\n",
    "        table_exists = True\n",
    "    except exceptions.NotFound:\n",
    "        table_exists = False\n",
    "        \n",
    "    # Create the table if it doesn't exist\n",
    "    if not table_exists:\n",
    "        location_data_schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"latitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"longitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_utc\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timezone\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"trajectory_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"source\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "        table = bigquery.Table(table_ref, schema=location_data_schema)\n",
    "        table = client.create_table(table)\n",
    "        print(f\"Table {table.project}.{table.dataset_id}.{table.table_id} created.\")\n",
    "\n",
    "    if table_exists:\n",
    "        # Delete rows with matching 'last_modified_on' values\n",
    "        last_modified_on_values = pd.to_datetime(all_compressed_data_for_export['last_modified_on']).dt.date.unique().tolist()\n",
    "        if last_modified_on_values:\n",
    "            delete_query = f\"\"\"\n",
    "                DELETE FROM `{dataset_id}.{table_id}`\n",
    "                WHERE DATE(last_modified_on) IN UNNEST(@last_modified_on_values)\n",
    "            \"\"\"\n",
    "            job_config = bigquery.QueryJobConfig(\n",
    "                query_parameters=[\n",
    "                    bigquery.ArrayQueryParameter(\"last_modified_on_values\", \"DATE\", last_modified_on_values)\n",
    "                ]\n",
    "            )\n",
    "            delete_job = client.query(delete_query, job_config=job_config)\n",
    "            delete_job.result()\n",
    "            print(f\"Deleted rows with 'last_modified_on' values: {last_modified_on_values}\")\n",
    "        else:\n",
    "            print(\"No rows to delete based on 'last_modified_on' values.\")\n",
    "\n",
    "    # Set the job configuration to overwrite the table if it exists\n",
    "    job_config = bigquery.LoadJobConfig()\n",
    "    job_config.autodetect = False\n",
    "    job_config.schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"latitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"longitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_utc\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timezone\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"trajectory_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"source\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "    job_config.source_format = bigquery.SourceFormat.CSV\n",
    "    job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND\n",
    "    job_config.create_disposition = bigquery.CreateDisposition.CREATE_IF_NEEDED\n",
    "    job_config.skip_leading_rows = 1  # Skip the first row (header row)\n",
    "\n",
    "    # Load the merged DataFrame into BigQuery\n",
    "    with open(output_csv, \"rb\") as source_file:\n",
    "        job = client.load_table_from_file(source_file, table_ref, job_config=job_config)\n",
    "\n",
    "    job.result()\n",
    "\n",
    "    print(\"Compressed Location Data loaded into BigQuery table: {}\".format(table_ref.path))\n",
    "    pass\n",
    "\n",
    "def stack_all_ANALYTICAL_location_features(DATA_DIR, file_pattern, output_csv, output_parquet):\n",
    "    files = glob.glob(os.path.join(DATA_DIR, file_pattern))\n",
    "\n",
    "    merged_analytical_location_features = []\n",
    "    analytical_features_processed = []\n",
    "\n",
    "    for file in files:\n",
    "        if file.endswith('.csv') and file not in analytical_features_processed:\n",
    "            df = pd.read_csv(file)\n",
    "            merged_analytical_location_features.append(df)\n",
    "            analytical_features_processed.append(file)\n",
    "        \n",
    "    all_features_for_export = pd.concat(merged_analytical_location_features, ignore_index=True)\n",
    "    all_features_for_export = all_features_for_export.sort_values('timestamp_unix', ascending=True)\n",
    "\n",
    "    # Check if the file exists and delete it\n",
    "    if os.path.exists(output_csv):\n",
    "        os.remove(output_csv)\n",
    "\n",
    "    if os.path.exists(output_parquet):\n",
    "        os.remove(output_parquet)\n",
    "        \n",
    "    all_features_for_export.to_csv(output_csv, index=False)\n",
    "    all_features_for_export.to_parquet(output_parquet, index=False)\n",
    "\n",
    "    # export to BigQuery - ANALYTIC_location_features\n",
    "    client = bigquery.Client()\n",
    "    dataset_id = \"analytic_location_features\"\n",
    "    table_id = \"JC_2024_rzr_analytic_location_features\"\n",
    "    table_ref = client.dataset(dataset_id).table(table_id)\n",
    "\n",
    "    # Check if the table exists\n",
    "    try:\n",
    "        client.get_table(table_ref)\n",
    "        table_exists = True\n",
    "    except exceptions.NotFound:\n",
    "        table_exists = False\n",
    "        \n",
    "    # Create the table if it doesn't exist\n",
    "    if not table_exists:\n",
    "        location_features_schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"latitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"longitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"min_of_day\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"hour_of_day\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"period_of_day\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"date\", \"DATE\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"time_local\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_month\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_year\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_week\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_week_name\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_workday\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_weekend\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_business_hours\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"month_name\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"month\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"quarter\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_index\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"move_activity\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_activity\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude1_minOverlap\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude2_hourOverlap\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude3_min2max\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"h3_lvl10_index\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"h3_lvl4_index\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_utc\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timezone\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"trajectory_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"source\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "        table = bigquery.Table(table_ref, schema=location_features_schema)\n",
    "        table = client.create_table(table)\n",
    "        print(f\"Table {table.project}.{table.dataset_id}.{table.table_id} created.\")\n",
    "\n",
    "    if table_exists:\n",
    "        # Delete rows with matching 'last_modified_on' values\n",
    "        last_modified_on_values = pd.to_datetime(all_features_for_export['last_modified_on']).dt.date.unique().tolist()\n",
    "        if last_modified_on_values:\n",
    "            delete_query = f\"\"\"\n",
    "                DELETE FROM `{dataset_id}.{table_id}`\n",
    "                WHERE DATE(last_modified_on) IN UNNEST(@last_modified_on_values)\n",
    "            \"\"\"\n",
    "            job_config = bigquery.QueryJobConfig(\n",
    "                query_parameters=[\n",
    "                    bigquery.ArrayQueryParameter(\"last_modified_on_values\", \"DATE\", last_modified_on_values)\n",
    "                ]\n",
    "            )\n",
    "            delete_job = client.query(delete_query, job_config=job_config)\n",
    "            delete_job.result()\n",
    "            print(f\"Deleted rows with 'last_modified_on' values: {last_modified_on_values}\")\n",
    "        else:\n",
    "            print(\"No rows to delete based on 'last_modified_on' values.\")\n",
    "\n",
    "    # Set the job configuration to overwrite the table if it exists\n",
    "    job_config = bigquery.LoadJobConfig()\n",
    "    job_config.autodetect = False\n",
    "    job_config.schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"latitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"longitude\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"min_of_day\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"hour_of_day\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"period_of_day\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"date\", \"DATE\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"time_local\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_month\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_year\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_week\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"day_of_week_name\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_workday\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_weekend\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"is_business_hours\", \"BOOLEAN\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"month_name\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"month\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"quarter\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"datetime_index\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"move_activity\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_activity\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude1_minOverlap\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude2_hourOverlap\", \"INTEGER\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"altitude3_min2max\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"h3_lvl10_index\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"h3_lvl4_index\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timestamp_utc\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"timezone\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"trajectory_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"source\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "    job_config.source_format = bigquery.SourceFormat.CSV\n",
    "    job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND\n",
    "    job_config.create_disposition = bigquery.CreateDisposition.CREATE_IF_NEEDED\n",
    "    job_config.skip_leading_rows = 1  # Skip the first row (header row)\n",
    "\n",
    "    # Load the merged DataFrame into BigQuery\n",
    "    with open(output_csv, \"rb\") as source_file:\n",
    "        job = client.load_table_from_file(source_file, table_ref, job_config=job_config)\n",
    "\n",
    "    job.result()\n",
    "\n",
    "    print(\"Analytic Location Features loaded into BigQuery table: {}\".format(table_ref.path))\n",
    "    pass \n",
    "\n",
    "def stack_all_STAY_location_features(DATA_DIR, file_pattern, output_csv, output_parquet):\n",
    "    files = glob.glob(os.path.join(DATA_DIR, file_pattern))\n",
    "\n",
    "    merged_stay_location_features = []\n",
    "    stay_processed_files = []\n",
    "\n",
    "    for file in files:\n",
    "        if file.endswith('.csv') and file not in stay_processed_files:\n",
    "            df = pd.read_csv(file, dtype={\n",
    "                'stay_duration_min': str,\n",
    "                'stay_num_points': str,\n",
    "                'stay_of_day': str\n",
    "            })\n",
    "            merged_stay_location_features.append(df)\n",
    "            stay_processed_files.append(file)\n",
    "        \n",
    "    stay_data_for_export = pd.concat(merged_stay_location_features, ignore_index=True)\n",
    "    stay_data_for_export = stay_data_for_export.sort_values('stay_start_unix', ascending=True)\n",
    "\n",
    "    # Convert specific columns to integers\n",
    "    stay_data_for_export['stay_duration_min'] = pd.to_numeric(stay_data_for_export['stay_duration_min'], errors='coerce')\n",
    "    stay_data_for_export['stay_num_points'] = pd.to_numeric(stay_data_for_export['stay_num_points'], errors='coerce')\n",
    "    stay_data_for_export['stay_of_day'] = pd.to_numeric(stay_data_for_export['stay_of_day'], errors='coerce')\n",
    "    \n",
    "    # Check if the file exists and delete it\n",
    "    if os.path.exists(output_csv):\n",
    "        os.remove(output_csv)\n",
    "\n",
    "    if os.path.exists(output_parquet):\n",
    "        os.remove(output_parquet)\n",
    "        \n",
    "    stay_data_for_export.to_csv(output_csv, index=False)\n",
    "    stay_data_for_export.to_parquet(output_parquet, index=False)\n",
    "\n",
    "    # export to BigQuery - STAY_location_features\n",
    "    client = bigquery.Client()\n",
    "    dataset_id = \"stay_locations_features\"\n",
    "    table_id = \"JC_2024_rzr_stay_locations_features\"\n",
    "    table_ref = client.dataset(dataset_id).table(table_id)\n",
    "\n",
    "    # Check if the table exists\n",
    "    try:\n",
    "        client.get_table(table_ref)\n",
    "        table_exists = True\n",
    "    except exceptions.NotFound:\n",
    "        table_exists = False\n",
    "        \n",
    "    # Create the table if it doesn't exist\n",
    "    if not table_exists:\n",
    "        stay_locations_schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_latitude_ctr\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_longitude_ctr\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_start_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_start_datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_end_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_end_datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_duration_min\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_num_points\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_max_diameter\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_of_day\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "        table = bigquery.Table(table_ref, schema=stay_locations_schema)\n",
    "        table = client.create_table(table)\n",
    "        print(f\"Table {table.project}.{table.dataset_id}.{table.table_id} created.\")\n",
    "\n",
    "    if table_exists:\n",
    "        # Delete rows with matching 'last_modified_on' values\n",
    "        last_modified_on_values = pd.to_datetime(stay_data_for_export['last_modified_on']).dt.date.unique().tolist()\n",
    "        if last_modified_on_values:\n",
    "            delete_query = f\"\"\"\n",
    "                DELETE FROM `{dataset_id}.{table_id}`\n",
    "                WHERE DATE(last_modified_on) IN UNNEST(@last_modified_on_values)\n",
    "            \"\"\"\n",
    "            job_config = bigquery.QueryJobConfig(\n",
    "                query_parameters=[\n",
    "                    bigquery.ArrayQueryParameter(\"last_modified_on_values\", \"DATE\", last_modified_on_values)\n",
    "                ]\n",
    "            )\n",
    "            delete_job = client.query(delete_query, job_config=job_config)\n",
    "            delete_job.result()\n",
    "            print(f\"Deleted rows with 'last_modified_on' values: {last_modified_on_values}\")\n",
    "        else:\n",
    "            print(\"No rows to delete based on 'last_modified_on' values.\")\n",
    "\n",
    "    # Set the job configuration to overwrite the table if it exists\n",
    "    job_config = bigquery.LoadJobConfig()\n",
    "    job_config.autodetect = False\n",
    "    job_config.schema = [\n",
    "            bigquery.SchemaField(\"device_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_latitude_ctr\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_longitude_ctr\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_start_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_start_datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_end_unix\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_end_datetime_local\", \"TIMESTAMP\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_duration_min\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_num_points\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_max_diameter\", \"FLOAT\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_of_day\", \"NUMERIC\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"stay_id\", \"STRING\", mode=\"NULLABLE\"),\n",
    "            bigquery.SchemaField(\"last_modified_on\", \"TIMESTAMP\", mode=\"NULLABLE\")\n",
    "        ]\n",
    "    job_config.source_format = bigquery.SourceFormat.CSV\n",
    "    job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND\n",
    "    job_config.create_disposition = bigquery.CreateDisposition.CREATE_IF_NEEDED\n",
    "    job_config.skip_leading_rows = 1  # Skip the first row (header row)\n",
    "\n",
    "    # Load the merged DataFrame into BigQuery\n",
    "    with open(output_csv, \"rb\") as source_file:\n",
    "        job = client.load_table_from_file(source_file, table_ref, job_config=job_config)\n",
    "\n",
    "    job.result()\n",
    "\n",
    "    print(\"Data loaded into BigQuery table: {}\".format(table_ref.path))\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### [0] Set Source of RAW_location_data objects\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01HYEA5Q2NHHAGM35DAV7FZEG4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Location of RAW_location_data objects \n",
    "DATA_DIR = \"/Users/jonathancachat/..../trajectory_data_processing_cleaned2analytic/data/gpx_rzr/\"\n",
    "\n",
    "# Set the combined filename prefix\n",
    "COMBINED_FILENAME = \"JC_2024_rzr\"\n",
    "\n",
    "TODAY = datetime.now()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### [1] RAW_location_data --> COMPRESSED_location_data & STAY_location_data (Cleaned Phase)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "01HYDZXK3QS2A9TBTA7E85YDWY",
   "metadata": {},
   "outputs": [],
   "source": [
    "# RAW_location_data --> COMPRESSED_location_data & STAY_location_data (Cleaned Phase)\n",
    "\n",
    "# Get a list of all GPX files in the directory\n",
    "gpx_files = glob.glob(os.path.join(DATA_DIR, \"*.gpx\"))\n",
    "\n",
    "# Process each GPX file for stage1_preprocess_all_in_one\n",
    "for gpx_file in gpx_files:\n",
    "    # Extract the device ID from the GPX file name\n",
    "    DEVICE_ID = os.path.splitext(os.path.basename(gpx_file))[0]\n",
    "    device_id = DEVICE_ID\n",
    "    \n",
    "    COMPRESSED_LOCATION_DATA = DATA_DIR+DEVICE_ID+'-COMPRESSED_location_data.csv'\n",
    "    STAY_LOCATIONS_DATA = DATA_DIR+DEVICE_ID+'-STAY_locations_data.csv'\n",
    "    \n",
    "    # Perform the data processing steps for stage1_preprocess_all_in_one\n",
    "    raw_df = ingest_raw_data(gpx_file, 'gpx_rzr')\n",
    "    \n",
    "    compressed_location_data, stay_locations_data = stage1_preprocess_all_in_one(raw_df)\n",
    "    compressed_location_data.to_csv(os.path.join(DATA_DIR, f\"{device_id}-COMPRESSED_location_data.csv\"), index=False)\n",
    "    #stay_locations_data.to_csv(os.path.join(DATA_DIR, f\"{device_id}-STAY_locations_data.csv\"), index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### [2] COMPRESSED_location_data & STAY_location_data --> ANALYTICAL_location_features & STAY_location_features\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "01HYEH7T06DR953W1PTDGE97Q0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# IN = compressed_location_data & stay_locations_data for features engineering\n",
    "\n",
    "# Get a list of all GPX files in the directory\n",
    "gpx_files = glob.glob(os.path.join(DATA_DIR, \"*.gpx\"))\n",
    "\n",
    "# Process each GPX file for process_spatiotemporal_features and process_stay_location_features\n",
    "for gpx_file in gpx_files:\n",
    "    # Extract the device ID from the GPX file name\n",
    "    DEVICE_ID = os.path.splitext(os.path.basename(gpx_file))[0]\n",
    "    device_id = DEVICE_ID\n",
    "    \n",
    "    COMPRESSED_LOCATION_DATA = DATA_DIR+DEVICE_ID+'-COMPRESSED_location_data.csv'\n",
    "    STAY_LOCATIONS_DATA = DATA_DIR+DEVICE_ID+'-STAY_locations_data.csv'\n",
    "    ANALYTIC_LOCATION_FEATURES = DATA_DIR+DEVICE_ID+'-ANALYTIC_location_features.csv'\n",
    "    STAY_LOCATIONS_FEATURES = DATA_DIR+DEVICE_ID+'-STAY_locations_features.csv'\n",
    "    \n",
    "    # Load the compressed_location_data and stay_locations_data from the output files\n",
    "    compressed_location_data = pd.read_csv(COMPRESSED_LOCATION_DATA)\n",
    "    #stay_locations_data = pd.read_csv(STAY_LOCATIONS_DATA)\n",
    "    \n",
    "    # Perform process_spatiotemporal_features\n",
    "    analytic_location_features = process_spatiotemporal_features(compressed_location_data, stay_locations_data)\n",
    "    analytic_location_features.to_csv(os.path.join(DATA_DIR, f\"{device_id}-ANALYTIC_location_features.csv\"), index=False)\n",
    "    \n",
    "    # Perform process_stay_location_features\n",
    "    stay_location_features = process_stay_location_features(stay_locations_data, DEVICE_ID)\n",
    "    stay_location_features.to_csv(os.path.join(DATA_DIR, f\"{device_id}-STAY_locations_features.csv\"), index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "01HYE3EFR99YDHQZ4P5F4SF581",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted rows with 'last_modified_on' values: [datetime.date(2024, 5, 21)]\n",
      "Compressed Location Data loaded into BigQuery table: /projects/tl-dev-datascience/datasets/compressed_location_data/tables/JC_2024_rzr_compressed_location_data\n",
      "Deleted rows with 'last_modified_on' values: [datetime.date(2024, 5, 21)]\n",
      "Analytic Location Features loaded into BigQuery table: /projects/tl-dev-datascience/datasets/analytic_location_features/tables/JC_2024_rzr_analytic_location_features\n"
     ]
    }
   ],
   "source": [
    "# Set the output file paths\n",
    "ALL_COMPRESSED_LOCATION_DATA_CSV = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-COMPRESSED_location_data.csv\")\n",
    "ALL_STAY_LOCATIONS_FEATURES_CSV = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-STAY_locations_features.csv\")\n",
    "ALL_ANALYTIC_LOCATION_FEATURES_CSV = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-ANALYTIC_location_features.csv\")\n",
    "\n",
    "ALL_COMPRESSED_LOCATION_DATA_PARQUET = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-COMPRESSED_location_data.parquet\")\n",
    "ALL_STAY_LOCATIONS_FEATURES_PARQUET = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-STAY_locations_features.parquet\")\n",
    "ALL_ANALYTIC_LOCATION_FEATURES_PARQUET = os.path.join(DATA_DIR, f\"{COMBINED_FILENAME}-ANALYTIC_location_features.parquet\")\n",
    "\n",
    "# Stack all the processed data files\n",
    "stack_all_COMPRESSED_location_data(DATA_DIR, \"*-COMPRESSED_location_data.csv\", ALL_COMPRESSED_LOCATION_DATA_CSV, ALL_COMPRESSED_LOCATION_DATA_PARQUET)\n",
    "stack_all_ANALYTICAL_location_features(DATA_DIR, \"*-ANALYTIC_location_features.csv\", ALL_ANALYTIC_LOCATION_FEATURES_CSV, ALL_ANALYTIC_LOCATION_FEATURES_PARQUET)\n",
    "#stack_all_STAY_location_features(DATA_DIR, \"*-STAY_locations_data.csv\", ALL_STAY_LOCATIONS_FEATURES_CSV, ALL_STAY_LOCATIONS_FEATURES_PARQUET)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "skmob-dev",
   "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.1.-1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}