micro_path.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MicroPath Reconstruction From AIS Broadcast Points\n",
    "\n",
    "Micropathing is the construction of a target's path from a limited set of a consecutive sequence of target points. Typically, the sequence is time-based, and the collection is limited to 2 or 3 target points.  The following is an illustration of 2 micropaths derived from 3 target points:\n",
    "\n",
    "![](media/Micropath0.png)\n",
    "\n",
    "Micropathing is different than path reconstruction, in such that the latter produced one polyline for the path of a target. Path reconstruction losses insightful in-path behavior, as a large number of attributes cannot be associated with the path parts. Some can argue that the points along the path can be enriched with these attributes. However, with the current implementations of Point objects, we are limited to only the extra `M` and `Z` to the necessary `X` and `Y`. You can also join the `PathID` and `M` to a lookup table and gain back that insight, but that joining is typically expensive and is difficult to derive from it the \"expression\" of the path using traditional mapping. A micropath overcomes today's limitations with today's traditional means to express the path insight better.\n",
    "\n",
    "So, a micropath is a line composed of typically 2 points only and is associated with a set of attributes that describe that line.  These attributes are typical enrichment metrics derived from its two ends. An attribute can be, for example, the traveled distance, time, or speed.\n",
    "\n",
    "In this notebook, we will construct \"clean\" micropaths using SparkSQL.  What do I mean by clean? As we all know, emitted target points are notoriously affected by noise, so using SparkSQL, we will eliminate that noise during the micropath construction."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import required modules."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import arcpy\n",
    "from spark_esri import spark_start, spark_stop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start Spark instance.\n",
    "\n",
    "Note the `config` argument to [configure the Spark instance](https://spark.apache.org/docs/latest/configuration.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "config = {\"spark.driver.memory\":\"2G\"}\n",
    "spark = spark_start(config=config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define the output spatial reference and fields.\n",
    "\n",
    "Here we are emitting the `x` and `y` values in web mercator meters for easier displacement calculations later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sp_ref = arcpy.SpatialReference(3857)\n",
    "fields = [\"MMSI\",\"SHAPE@X\",\"SHAPE@Y\",\"BaseDateTime\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the selected `Broadcast` features and create a Spark dataframe that is mapped to a SparkSQL view.\n",
    "\n",
    "Note that we are extracting the hour value from the timestamp, and converting the timestamp to an epoch in seconds since 1970."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with arcpy.da.SearchCursor(\"Broadcast\", fields, spatial_reference=sp_ref) as rows:\n",
    "    spark\\\n",
    "        .createDataFrame(rows, \"mmsi string,x double,y double,t timestamp\")\\\n",
    "        .selectExpr(\"mmsi\",\"x\",\"y\",\"hour(t) h\",\"unix_timestamp(t) t\")\\\n",
    "        .createOrReplaceTempView(\"v0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start the micopath construction by using the SparkSQL window function to find the leading record to the current record."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark\\\n",
    "    .sql(\"\"\"\n",
    "select mmsi,h,\n",
    "x x1,\n",
    "y y1,\n",
    "t t1,\n",
    "lead(x,1,0.0) over (partition by mmsi order by t) x2,\n",
    "lead(y,1,0.0) over (partition by mmsi order by t) y2,\n",
    "lead(t,1,0) over (partition by mmsi order by t) t2\n",
    "from v0\n",
    "\"\"\")\\\n",
    "    .createOrReplaceTempView(\"v1\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Enrich the micropath with vertical and horizontal displacements in meters, and add the endpoint time difference in seconds."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"select *,(x2-x1) dx,(y2-y1) dy,(t2-t1) dt from v1 where t1 < t2\").createOrReplaceTempView(\"v2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculate the travel distance in meters.\n",
    "\n",
    "```\n",
    "dd = sqrt(dx*dx+dy*dy)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"select mmsi,h,x1,y1,x2,y2,sqrt(dx*dx+dy*dy) dd,dt from v2\").createOrReplaceTempView(\"v3\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculate the travel speed in meters per second."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"select *,dd/dt mps from v3\").cache().createOrReplaceTempView(\"v4\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Noise Elimination.\n",
    "\n",
    "Here we approximate the 99th percentile of the distance, time and speed values, and we will use the resulting values as the noise reduction thresholds."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"\n",
    "select\n",
    "percentile_approx(mps,0.99) mps,\n",
    "percentile_approx(dt,0.99) dt,\n",
    "percentile_approx(dd,0.99) dd\n",
    "from v4\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Filter the data to create \"clean\" micropaths.\n",
    "\n",
    "Here, we roughly doubled the `mps` (12.3) and `dd` (778.6) as a \"tolerant\" threshold."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"\n",
    "select mmsi,h,dd,dt,mps,x1,y1,x2,y2\n",
    "from v4\n",
    "where dd between 1 and 1500\n",
    "and mps < 25\n",
    "and dt < 130\n",
    "\"\"\")\\\n",
    "    .cache()\\\n",
    "    .createOrReplaceTempView(\"v5\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collect the micropaths as features where the shape is in WKT format.\n",
    "\n",
    "The usage of the WKT format will relevant in the subsquent cell during the insert process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rows = spark.sql(\"\"\"\n",
    "select mmsi,h,dd,dt,mps,concat('LINESTRING(',x1,' ',y1,',',x2,' ',y2,')') wkt\n",
    "from v5\n",
    "\"\"\")\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an in-memory line feature class in the TOC."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ws = \"memory\"\n",
    "nm = \"MicroPaths\"\n",
    "\n",
    "fc = os.path.join(ws,nm)\n",
    "\n",
    "arcpy.management.Delete(fc)\n",
    "\n",
    "sp_ref = arcpy.SpatialReference(3857)\n",
    "arcpy.management.CreateFeatureclass(ws,nm,\"POLYLINE\",spatial_reference=sp_ref)\n",
    "arcpy.management.AddField(fc, \"MMSI\", \"TEXT\")\n",
    "arcpy.management.AddField(fc, \"HH\", \"LONG\")\n",
    "arcpy.management.AddField(fc, \"DD\", \"DOUBLE\")\n",
    "arcpy.management.AddField(fc, \"DT\", \"DOUBLE\")\n",
    "arcpy.management.AddField(fc, \"MPS\", \"DOUBLE\")\n",
    "\n",
    "# Note shape is expected to be in WKT\n",
    "with arcpy.da.InsertCursor(fc, [\"MMSI\",\"HH\",\"DD\",\"DT\",\"MPS\",\"SHAPE@WKT\"]) as cursor:\n",
    "    for row in rows:\n",
    "        cursor.insertRow(row)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stop the spark instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}