rt_segment_speeds/11_tiger.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "18ada7ff-048d-4c5c-9523-9f191e772137",
   "metadata": {},
   "source": [
    "## Tiger Census\n",
    "* https://www2.census.gov/geo/pdfs/maps-data/data/tiger/tgrshp2019/TGRSHP2019_TechDoc.pdf\n",
    "* S1200 - secondary road\n",
    "* S1100 - primary road\n",
    "* S1400 - local roads\n",
    "* Build off scripts/cut_road_segments.py\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e1409ea6-c44a-45af-9a4b-7752f5f8cfd5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/conda/lib/python3.9/site-packages/geopandas/_compat.py:123: UserWarning: The Shapely GEOS version (3.11.1-CAPI-1.17.1) is incompatible with the GEOS version PyGEOS was compiled with (3.10.1-CAPI-1.16.0). Conversions between both will be slow.\n",
      "  warnings.warn(\n",
      "/opt/conda/lib/python3.9/site-packages/dask_geopandas/backends.py:13: UserWarning: Shapely 2.0 is installed, but because PyGEOS is also installed, GeoPandas will still use PyGEOS by default for now. To force to use and test Shapely 2.0, you have to set the environment variable USE_PYGEOS=0. You can do this before starting the Python process, or in your code before importing geopandas:\n",
      "\n",
      "import os\n",
      "os.environ['USE_PYGEOS'] = '0'\n",
      "import geopandas\n",
      "\n",
      "In a future release, GeoPandas will switch to using Shapely by default. If you are using PyGEOS directly (calling PyGEOS functions on geometries from GeoPandas), this will then stop working and you are encouraged to migrate from PyGEOS to Shapely 2.0 (https://shapely.readthedocs.io/en/latest/migration_pygeos.html).\n",
      "  import geopandas\n"
     ]
    }
   ],
   "source": [
    "import datetime\n",
    "import gcsfs\n",
    "fs = gcsfs.GCSFileSystem()\n",
    "import dask.dataframe as dd\n",
    "import dask_geopandas as dg\n",
    "import geopandas as gpd\n",
    "import pandas as pd\n",
    "from calitp_data_analysis.sql import to_snakecase\n",
    "from dask import compute, delayed\n",
    "from segment_speed_utils import helpers\n",
    "from segment_speed_utils.project_vars import analysis_date\n",
    "from shared_utils import dask_utils\n",
    "from calitp_data_analysis import geography_utils, utils\n",
    "\n",
    "GCS_FILE_PATH = \"gs://calitp-analytics-data/data-analyses/\"\n",
    "SHARED_GCS = f\"{GCS_FILE_PATH}shared_data/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c69abcdb-7509-47c8-b891-a99c899db38b",
   "metadata": {},
   "outputs": [],
   "source": [
    "pd.options.display.max_columns = 100\n",
    "pd.options.display.float_format = \"{:.2f}\".format\n",
    "pd.set_option(\"display.max_rows\", None)\n",
    "pd.set_option(\"display.max_colwidth\", None)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af59e43d-f137-4a0a-9fb9-e6ebd0a6d030",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Tiger - Load Roads\n",
    "* TO DO: remove buffer, do it another step b/c now I have to dissolve twice and that takes too long"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "61786354-ec33-47bd-9891-cb292068c703",
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_roads(road_type_wanted: list) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Load roads based on what you filter.\n",
    "\n",
    "    Args:\n",
    "        road_type_wanted (list): the type of roads you want.\n",
    "\n",
    "        https://www2.census.gov/geo/pdfs/maps-data/data/tiger/tgrshp2019/TGRSHP2019_TechDoc.pdf\n",
    "        buffer_or_not (bool): add a buffer of 200.\n",
    "\n",
    "    Returns:\n",
    "        GDF. As of 4/18/23, returns 953914 nunique linearid\n",
    "    \"\"\"\n",
    "    df = gpd.read_parquet(\n",
    "        f\"{SHARED_GCS}all_roads_2020_state06.parquet\",\n",
    "        filters=[(\"MTFCC\", \"in\", road_type_wanted)],\n",
    "        columns=[\"LINEARID\", \"geometry\", \"FULLNAME\"],\n",
    "    ).to_crs(geography_utils.CA_NAD83Albers_m)\n",
    "\n",
    "    # If a road has mutliple rows but the same\n",
    "    # linear ID, dissolve it so it becomes one row.\n",
    "    df = (\n",
    "        df.drop_duplicates()\n",
    "        .dissolve(by=[\"LINEARID\"])\n",
    "        .reset_index()\n",
    "        .drop_duplicates()\n",
    "        .reset_index(drop=True)\n",
    "    )\n",
    "\n",
    "    df = to_snakecase(df)\n",
    "\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "55408ee8-7dbd-4171-9575-0cb138c8b60c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# len(og_tiger)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "1c2a4b80-9bec-44bb-8a2a-63ce7a11a55c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# og_tiger.linearid.nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "cd327db3-345d-48f5-b673-43edb740a210",
   "metadata": {},
   "outputs": [],
   "source": [
    "# len(og_tiger.drop_duplicates())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "c868617d-b244-47f3-80e4-8c941ec2fe45",
   "metadata": {},
   "outputs": [],
   "source": [
    "# more_than_1 = og_tiger.linearid.value_counts().loc[lambda x: x>1].reset_index()['index']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "f825b43b-1f58-40bc-808c-0a62fae9c510",
   "metadata": {},
   "outputs": [],
   "source": [
    "# more_than_1 = list(more_than_1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "23e094ff-b20d-4d1b-b129-8a2c830ca48c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# len(more_than_1)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d935930a-93da-4609-a8d8-671a4626adde",
   "metadata": {},
   "source": [
    "#### Cesar Chavez Test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "4e0a190a-c26d-48cc-9a57-c9ae2e8fde2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# cesar_chavez = og_tiger[og_tiger.fullname == \"Cesar Chavez\"].reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "58fc873f-bb95-4129-a154-12d64e089264",
   "metadata": {},
   "outputs": [],
   "source": [
    "# cesar_chavez"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "45bdd874-29bf-4e67-8c07-04bd9cdda0d6",
   "metadata": {},
   "source": [
    "### GTFS Shapes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "a67fb46e-c33b-4dee-82c4-dcd611443963",
   "metadata": {},
   "outputs": [],
   "source": [
    "def gtfs_stops_operators(date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Load stops with operator and feed key information.\n",
    "\n",
    "    Args:\n",
    "        date: analysis date\n",
    "    \"\"\"\n",
    "    stops = (\n",
    "        helpers.import_scheduled_stops(\n",
    "            date, (), [\"feed_key\", \"stop_id\", \"stop_key\", \"geometry\"]\n",
    "        )\n",
    "        .compute()\n",
    "        .drop_duplicates()\n",
    "    )\n",
    "\n",
    "    stops = stops.set_crs(geography_utils.CA_NAD83Albers_m)\n",
    "\n",
    "    # Buffer each stop by 50 feet\n",
    "    stops = stops.assign(buffered_geometry=stops.geometry.buffer(50))\n",
    "\n",
    "    # Set geometry\n",
    "    stops = stops.set_geometry(\"buffered_geometry\")\n",
    "\n",
    "    # Merge for operator information\n",
    "    trips = (\n",
    "        helpers.import_scheduled_trips(analysis_date, (), [\"name\", \"feed_key\"])\n",
    "        .compute()\n",
    "        .drop_duplicates()\n",
    "    )\n",
    "\n",
    "    m1 = pd.merge(stops, trips, on=[\"feed_key\"], how=\"left\")\n",
    "\n",
    "    # Fill in na\n",
    "    m1.name = m1.name.fillna(\"None\")\n",
    "\n",
    "    return m1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "0e17d8f7-07ca-48b7-9b9d-e28cbb3b0285",
   "metadata": {},
   "outputs": [],
   "source": [
    "# stops = gtfs_stops_operators(analysis_date)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "af866cf6-65d8-4d4f-96d4-a2eb8f3bf88b",
   "metadata": {},
   "outputs": [],
   "source": [
    "def gtfs_routes_operators(date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Load routes with operator and feed key information.\n",
    "\n",
    "    Args:\n",
    "        date: analysis date\n",
    "    \"\"\"\n",
    "    gtfs_shapes = helpers.import_scheduled_shapes(date).compute().drop_duplicates()\n",
    "\n",
    "    gtfs_shapes = gtfs_shapes.set_crs(geography_utils.CA_NAD83Albers_m)\n",
    "\n",
    "    trips = (\n",
    "        helpers.import_scheduled_trips(date, (), [\"name\", \"shape_array_key\"])\n",
    "        .compute()\n",
    "        .drop_duplicates()\n",
    "    )\n",
    "\n",
    "    m1 = pd.merge(gtfs_shapes, trips, how=\"left\", on=\"shape_array_key\")\n",
    "\n",
    "    return m1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "ee6d2496-861a-4cbb-a3d9-98fab1735da5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def order_operators(date:str) -> list:\n",
    "    \"\"\"\n",
    "    Reorder a list of operators in which the largest\n",
    "    ones will be at the top of the list.\n",
    "\n",
    "    Args:\n",
    "        date: analysis date\n",
    "    \"\"\"\n",
    "    operator_list = (\n",
    "        helpers.import_scheduled_trips(date, (), [\"name\"]).compute().sort_values(\"name\")\n",
    "    )\n",
    "    operator_list = operator_list.name.unique().tolist()\n",
    "\n",
    "    # Reorder list so the biggest operators are at the beginning\n",
    "    # based on NTD services data\n",
    "    big_operators = [\n",
    "        \"LA DOT Schedule\",\n",
    "        \"LA Metro Bus Schedule\",\n",
    "        \"LA Metro Rail Schedule\",\n",
    "        \"Bay Area 511 Muni Schedule\",\n",
    "        \"Bay Area 511 AC Transit Schedule\",\n",
    "        \"Bay Area 511 Santa Clara Transit Schedule\",\n",
    "        \"Bay Area 511 BART Schedule\",\n",
    "        \"San Diego Schedule\",\n",
    "        \"OCTA Schedule\",\n",
    "        \"Sacramento Schedule\",\n",
    "        \"Bay Area 511 Sonoma-Marin Area Rail Transit Schedule\",\n",
    "        \"Bay Area 511 SFO AirTrain Schedule\",\n",
    "        \"Bay Area 511 South San Francisco Shuttle Schedule\",\n",
    "        \"Bay Area 511 Marin Schedule\",\n",
    "        \"Bay Area 511 County Connection Schedule\",\n",
    "        \"Bay Area 511 MVGO Schedule\",\n",
    "        \"Bay Area 511 Commute.org Schedule\",\n",
    "        \"Bay Area 511 Union City Transit Schedule\",\n",
    "        \"Bay Area 511 BART Schedule\",\n",
    "        \"Bay Area 511 Caltrain Schedule\",\n",
    "        \"Bay Area 511 Fairfield and Suisun Transit Schedule\",\n",
    "        \"Bay Area 511 Dumbarton Express Schedule\",\n",
    "        \"Bay Area 511 SamTrans Schedule\",\n",
    "        \"Bay Area 511 Vine Transit Schedule\",\n",
    "        \"Bay Area 511 Tri-Valley Wheels Schedule\",\n",
    "        \"Bay Area 511 Sonoma County Transit Schedule\",\n",
    "        \"Bay Area 511 Santa Rosa CityBus Schedule\",\n",
    "        \"Bay Area 511 Golden Gate Transit Schedule\",\n",
    "        \"Bay Area 511 Golden Gate Ferry Schedule\",\n",
    "        \"Bay Area 511 San Francisco Bay Ferry Schedule\",\n",
    "        \"Bay Area 511 SolTrans Schedule\",\n",
    "        \"Bay Area 511 ACE Schedule\",\n",
    "        \"Bay Area 511 Emery Go-Round Schedule\",\n",
    "        \"Bay Area 511 Tri Delta Schedule\",\n",
    "        \"Bay Area 511 Petaluma Schedule\",\n",
    "        \"Bay Area 511 Capitol Corridor Schedule\",\n",
    "    ]\n",
    "\n",
    "    # Delete off the big operators\n",
    "    operator_list = list(set(operator_list) - set(big_operators))\n",
    "\n",
    "    # Add back in the operators\n",
    "    final_list = big_operators + operator_list\n",
    "\n",
    "    return final_list"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "876aad40-5b71-42b9-b899-11ecd15e2cfe",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Tiger Local Roads"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4002d6ad-0bac-45f5-a7a0-c9358fc39376",
   "metadata": {},
   "source": [
    "#### Cut all roads - stops 1st then routes \n",
    "* Use some small operators to test."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "80e1cb44-e512-4de9-8a8c-d00621db840a",
   "metadata": {},
   "outputs": [],
   "source": [
    "def loop_sjoin(date:str, local_roads_gdf:gpd.GeoDataFrame, gdf_routes_stops:gpd.GeoDataFrame) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    By operator, sjoin either routes or stops to the tiger gdf. \n",
    "    Delete off any linear ids that have already been joined.\n",
    "\n",
    "    Args:\n",
    "        local_roads_gdf: local roads gdf, use the buffered version of Tiger \n",
    "        gdf_routes_stops: stops or routes gdf\n",
    "        date: analysis date\n",
    "    \"\"\"\n",
    "    # Empty dataframe\n",
    "    sjoin_full_results = pd.DataFrame()\n",
    "\n",
    "    # Find all unique operators, ordered by largest operators first\n",
    "    operators_list = order_operators(date)\n",
    "\n",
    "    # Loop through and sjoin by operator\n",
    "    for operator in operators_list:\n",
    "        shapes_filtered = gdf_routes_stops.loc[\n",
    "            gdf_routes_stops.name == operator\n",
    "        ].reset_index(drop=True)\n",
    "\n",
    "        # Delete any local road linear ids that have already been found by an operator\n",
    "        try:\n",
    "            # List of linear IDS\n",
    "            linearid_to_delete = sjoin_full_results.linearid.unique().tolist()\n",
    "\n",
    "            # Filter out the linear IDS in buffered local roads\n",
    "            local_roads_gdf = local_roads_gdf[\n",
    "                ~local_roads_gdf.linearid.isin(linearid_to_delete)\n",
    "            ].reset_index(drop=True)\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "        # Do a sjoin but  keep the linearid as the only column\n",
    "        sjoin1 = (\n",
    "            gpd.sjoin(\n",
    "                local_roads_gdf,\n",
    "                shapes_filtered,\n",
    "                how=\"inner\",\n",
    "                predicate=\"intersects\",\n",
    "            )[[\"linearid\"]]\n",
    "            .drop_duplicates()\n",
    "            .reset_index(drop=True)\n",
    "        )\n",
    "\n",
    "        sjoin_full_results = pd.concat([sjoin_full_results, sjoin1], axis=0)\n",
    "\n",
    "    sjoin_full_results = sjoin_full_results.drop_duplicates()\n",
    "\n",
    "    return sjoin_full_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "98c2aa4c-1260-4b7b-b4f4-8be3c91a0bce",
   "metadata": {},
   "outputs": [],
   "source": [
    "def sjoin_stops(buffered_roads:gpd.GeoDataFrame, original_roads:gpd.GeoDataFrame, date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Sjoin stops to local roads.\n",
    "    \n",
    "    Args:\n",
    "        buffered_roads: local TIGER roads gdf that are buffered\n",
    "        original_roads: original local TIGER roads gdf\n",
    "        date: analysis date\n",
    "\n",
    "    Returns:\n",
    "        A list of linear IDs that have already\n",
    "        been found and a GDF\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "\n",
    "    # Load stops\n",
    "    gtfs_stops = gtfs_stops_operators(date)\n",
    "\n",
    "    # Loop through and sjoin by operator\n",
    "    stops_sjoin = loop_sjoin(date, buffered_roads, gtfs_stops)\n",
    "\n",
    "    # Merge back to original local roads gdf, so we have the\n",
    "    # non buffered geometry.\n",
    "    m1 = pd.merge(original_roads, stops_sjoin, on=\"linearid\", how=\"inner\")\n",
    "\n",
    "    # Fill in null values for fullname\n",
    "    m1.fullname = m1.fullname.fillna(\"None\")\n",
    "\n",
    "    # Find linear ids to delete\n",
    "    linearid_to_delete = m1.linearid.unique().tolist()\n",
    "\n",
    "    # Save\n",
    "    m1.to_parquet(f\"{SHARED_GCS}local_roads_stops_sjoin.parquet\")\n",
    "\n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done with sjoin with stops with local roads. Time lapsed: {end-start}\")\n",
    "\n",
    "    return m1, linearid_to_delete"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "e97a8499-6db4-4f20-ae60-a4845988515e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def sjoin_routes(buffered_roads: gpd.GeoDataFrame, original_roads: gpd.GeoDataFrame, date:str, linearid_to_delete: list) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Sjoin routes to local roads.\n",
    "    \n",
    "    Args:\n",
    "        buffered_roads: local TIGER roads that are buffered\n",
    "        original_roads: original local Tiger roads\n",
    "        date: analysis date\n",
    "        linearid_to_delete: linear ids to delete that have already been found \n",
    "        while applying a sjoin to stops.\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "\n",
    "    # Load stops\n",
    "    gtfs_routes = gtfs_routes_operators(date)\n",
    "\n",
    "    # Delete out linear ids that have already been found\n",
    "    local_roads_buffered = buffered_roads[~buffered_roads.linearid.isin(linearid_to_delete)].reset_index(drop=True)\n",
    "    local_roads_og = original_roads[~original_roads.linearid.isin(linearid_to_delete)].reset_index(drop=True)\n",
    "    \n",
    "    # Sjoin\n",
    "    routes_sjoin = loop_sjoin(date, local_roads_buffered, gtfs_routes)\n",
    "\n",
    "    # Merge back to original local roads, so we have the\n",
    "    # non buffered geometry.\n",
    "    m1 = pd.merge(local_roads_og, routes_sjoin, on=\"linearid\", how=\"inner\")\n",
    "\n",
    "    # Fill in null values for fullname\n",
    "    m1.fullname = m1.fullname.fillna(\"None\")\n",
    "\n",
    "    # Save\n",
    "    m1.to_parquet(f\"{SHARED_GCS}local_roads_routes_sjoin.parquet\")\n",
    "\n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done with sjoin with routes and local roads. Time lapsed: {end-start}\")\n",
    "\n",
    "    return m1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "d6c298fb-28a8-44b2-b423-e3c6ac75ff7c",
   "metadata": {},
   "outputs": [],
   "source": [
    "def sjoin_local_roads(date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Sjoin local roads with stops first then routes.\n",
    "    \n",
    "    Args:\n",
    "        date: analysis date\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Begin sjoin\")\n",
    "    \n",
    "    # Load local roads - not buffered\n",
    "    local_roads_og = load_roads([\"S1400\"])\n",
    "    \n",
    "    # Load local roads - buffered\n",
    "    local_roads_buffered = local_roads_og.assign(geometry=local_roads_og.geometry.buffer(200))\n",
    "    local_roads_buffered = local_roads_buffered.set_geometry('geometry')\n",
    "    \n",
    "    print(f\"Done buffering\")\n",
    "    \n",
    "    # Deal with stops first\n",
    "    stops_sjoin, linear_id_stops = sjoin_stops(\n",
    "        local_roads_buffered, local_roads_og, date\n",
    "    )\n",
    "\n",
    "    # Move onto routes\n",
    "    routes_sjoin = sjoin_routes(\n",
    "        local_roads_buffered, local_roads_og, date, linear_id_stops\n",
    "    )\n",
    "\n",
    "    # Stack\n",
    "    all_local_roads = pd.concat([stops_sjoin, routes_sjoin], axis=0)\n",
    "    \n",
    "    file_date = date.replace('-','_')\n",
    "    all_local_roads.to_parquet(f\"{SHARED_GCS}local_roads_all_routes_stops_sjoin_{file_date}.parquet\")\n",
    "    \n",
    "    end = datetime.datetime.now()\n",
    "\n",
    "    print(f\"Done with doing an sjoin for all local roads. Time lapsed: {end-start}\")\n",
    "    return all_local_roads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "c594aa88-c1ae-4b74-ba1b-0468e383d500",
   "metadata": {},
   "outputs": [],
   "source": [
    "# analysis_date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "992cd1c8-3aa0-40ba-a201-4b840cb02d95",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_ops = sjoin_local_roads(analysis_date)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8c75566-2320-4161-b5a3-cd4ba9c2aab2",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Dask Redo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "9fe405e1-fadb-4d70-b20c-41415fc69c6a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Find all the parquets again\n",
    "def find_files(phrase_to_find: str) -> list:\n",
    "    \"\"\"\n",
    "    Grab a list of files that contain the\n",
    "    phrase inputted. \n",
    "    \"\"\"\n",
    "    folder = f\"{SHARED_GCS}partitioned_tiger\"\n",
    "    \n",
    "    # Create a list of all the files in my folder\n",
    "    all_files_in_folder = fs.ls(folder)\n",
    "\n",
    "    # Grab only files with the string \"Verizon_no_coverage_\"\n",
    "    my_files = [i for i in all_files_in_folder if phrase_to_find in i]\n",
    "\n",
    "    # String to add to read the files\n",
    "    my_string = \"gs://\"\n",
    "    my_files = [my_string + i for i in my_files]\n",
    "    \n",
    "    # Extract digit of parquet \n",
    "    return my_files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "922d054e-ea21-4bd7-9a43-d7672c1850d3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_number(phrase_to_find: str) -> list:\n",
    "    \"\"\"\n",
    "    Extract the numeric portion of a file path.\n",
    "    \"\"\"\n",
    "    files = find_files(phrase_to_find)\n",
    "    all_file_numbers = []\n",
    "    for file in files:\n",
    "    # https://stackoverflow.com/questions/11339210/how-to-get-integer-values-from-a-string-in-python\n",
    "        file_number = \"\".join(i for i in file if i.isdigit())\n",
    "        all_file_numbers.append(file_number)\n",
    "    return all_file_numbers "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "aa27d02b-c75b-47c5-8875-de714d4971a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "def chunk_dask_df(gdf) -> list:\n",
    "    \"\"\"\n",
    "    Break up dataframes by a certain\n",
    "    number of rows, turn them into a dask\n",
    "    dataframe\n",
    "\n",
    "    Args:\n",
    "        gdf: the local roads that intersect w/ stops and routes\n",
    "        chunk_row_size(int): how many rows each dataframe should\n",
    "        be after splitting it out.\n",
    "\n",
    "    Returns:\n",
    "        List of dask dataframes. Length of how many dask dataframes\n",
    "        are returned after cutting.\n",
    "    \"\"\"\n",
    "   # Turn sjoin local roads to dask\n",
    "    ddf1 = dd.from_pandas(gdf, npartitions=1)\n",
    "\n",
    "    # Partition the sjoin stuff automatically\n",
    "    ddf1_partitioned = ddf1.repartition(partition_size=\"1MB\")\n",
    "    \n",
    "    #Save out to GCS\n",
    "    ddf1_partitioned.to_parquet(f\"{SHARED_GCS}partitioned_tiger\", overwrite = True)\n",
    "    \n",
    "    # Read back all the partitioned stuff - grab the file number\n",
    "    #part0.parquet, part1.parquet\n",
    "    file_names_dask = extract_number(\"part\")\n",
    "    \n",
    "    # https://www.geeksforgeeks.org/read-multiple-csv-files-into-separate-dataframes-in-python/\n",
    "    # create empty list\n",
    "    dataframes_list = []\n",
    " \n",
    "    # append datasets into the list\n",
    "    for i in range(len(file_names_dask)):\n",
    "        gcs_file_path = f\"{SHARED_GCS}partitioned_tiger/part.\"\n",
    "        temp_df = dg.read_parquet(f\"{gcs_file_path}{file_names_dask[i]}.parquet\")\n",
    "        dataframes_list.append(temp_df)\n",
    "        \n",
    "    return dataframes_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "985516d3-4616-4690-880c-9b08f9603984",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_geometry_compute(dask_dataframe_list:list) -> gpd.GeoDataFrame:\n",
    "    # Cut geometry\n",
    "    print(\"Cut geometry\")\n",
    "    cut_results = []\n",
    "    for ddf in dask_dataframe_list:\n",
    "        cut_geometry = delayed(geography_utils.cut_segments)(ddf, [\"linearid\", \"fullname\"], 1_000)\n",
    "        cut_results.append(cut_geometry)\n",
    "        \n",
    "    print(f\"Begin computing\")\n",
    "    # Compute \n",
    "    cut_results = [compute(i)[0] for i in cut_results]\n",
    "    cut_df = pd.concat(cut_results, axis=0).reset_index(drop=True)\n",
    "    \n",
    "    return cut_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "bb37b74d-41d0-4cfc-8824-32b45ff0d5c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_local_roads(gdf, date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Cut all the local roads.\n",
    "    \n",
    "    gdf: the local roads to cut\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Begin cutting local roads\")\n",
    "\n",
    "    # Divide the gdf into equal sized chunks (roughly)\n",
    "    # and turn them into dask gdfs\n",
    "    print(\"Split into mulitple parquets\")\n",
    "    ddf_list = chunk_dask_df(gdf)\n",
    "    \n",
    "    cut_df = cut_geometry_compute(ddf_list)\n",
    "    \n",
    "    \"\"\"\n",
    "    # Cut geometry\n",
    "    cut_results = []\n",
    "    for ddf in ddf_list:\n",
    "        cut_geometry = delayed(geography_utils.cut_segments)(ddf, [\"linearid\", \"fullname\"], 1_000)\n",
    "        cut_results.append(cut_geometry)\n",
    "    print(\"Cut geometry\")\n",
    "    \n",
    "    print(f\"Begin computing\")\n",
    "    # Compute \n",
    "    cut_results = [compute(i)[0] for i in cut_results]\n",
    "    cut_df = pd.concat(cut_results, axis=0).reset_index(drop=True)\n",
    "    \n",
    "    \"\"\"\n",
    "    file_date = date.replace('-','_')\n",
    "    cut_df.to_parquet(f\"{SHARED_GCS}segmented_local_rds_{file_date}.parquet\")\n",
    "    \n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done cutting local roads in {end-start} minutes\")\n",
    "    return cut_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "eb8bccb1-8f28-4e91-b37f-6d9939813d7c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# test = cut_local_roads(analysis_date)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "2b7ecbb8-0f54-4bd5-bd33-e0c8ed10f4f7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_ops.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "a821e63b-0878-4eb3-893a-b5aab32a5519",
   "metadata": {},
   "outputs": [],
   "source": [
    "#  all_ops.sample()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "7f5f75cc-3514-41ff-9bc7-5f1fc98d6750",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_ops.plot()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "90c61251-083c-4b0c-a3e4-93448b26afcf",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ddfs, length = chunk_dask_df(all_ops, 10000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "821ed543-0b04-474a-924d-ca4ec5786265",
   "metadata": {},
   "outputs": [],
   "source": [
    "# len(ddfs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "328f7871-6dfa-4eca-bc22-df6b5e9e03ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type(ddfs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "7621c1cc-8499-46ea-85fa-0675d230a5fc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type(ddfs[0]), type(ddfs[15])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "a8245e97-95b3-4db0-aa70-49fce1925f3e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# test = dask_segment(ddfs, [0,1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "260c7c04-6e36-4727-93ab-6ec081155448",
   "metadata": {},
   "outputs": [],
   "source": [
    "# test.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "63f2492d-87b9-4b1c-85c1-516f7096b5cc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# test = cut_local_roads(analysis_date, 10000)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "40536a55-bbbf-4abb-aed2-205223d7d807",
   "metadata": {
    "tags": []
   },
   "source": [
    "###  Monthly run "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "f0e41423-d8b1-45ea-8812-3152ecedf7f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "def monthly_linearids(date:str, last_month_segmented_local_roads: str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Instead of re-cutting all the primary and secondary roads and \n",
    "    local roads found from the last run, only cut the new local roads\n",
    "    that are found. Delete out any local roads that aren't found in \n",
    "    this month's routes. \n",
    "    \n",
    "    Args:\n",
    "        date: analysis_date\n",
    "        last_month_segmented_local_roads: file name of last month's local roads that\n",
    "        have been cut. Don't include .parquet.\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Start: {start}\")\n",
    "    \n",
    "    # Sjoin this month's data to tiger roads\n",
    "    this_month_gdf = sjoin_local_roads(date)\n",
    "    \n",
    "    # Find this month's linearids\n",
    "    this_month_linearid = set(this_month_gdf.linearid.unique().tolist())\n",
    "    \n",
    "    # Grab last month's results that have already been cut - local roads only\n",
    "    last_month_gdf = gpd.read_parquet(f\"{SHARED_GCS}{last_month_segmented_local_roads}.parquet\")\n",
    "    last_month_linearid = set(last_month_gdf.linearid.unique().tolist())\n",
    "    \n",
    "    # Have to cut linear ids that appear in this month but not last month\n",
    "    linearids_to_cut = list(this_month_linearid - last_month_linearid)\n",
    "    print(f\"There are {len(linearids_to_cut)} new linear ids found this month that didn't appear last month.\")\n",
    "    \n",
    "    # Have to delete linear ids that appear in last month but not this month.\n",
    "    linearids_to_delete = list(last_month_linearid - this_month_linearid)\n",
    "    print(f\"There are {len(linearids_to_delete)} that didn't appear this month that will be deleted.\")\n",
    "    \n",
    "    # Filter out linear ids that are no longer relevant to this month\n",
    "    cut_linearid_1= last_month_gdf.loc[~last_month_gdf.linearid.isin(linearids_to_delete)].reset_index(drop = True)\n",
    "    \n",
    "    # Cut the linearids that are only found in this month\n",
    "    cut_linearid_2 = this_month_gdf.loc[this_month_gdf.linearid.isin(linearids_to_cut)].reset_index(drop = True)\n",
    "    cut_linearid_2 = cut_local_roads(cut_linearid_2, date)\n",
    "    \n",
    "    # Compare lengths of last versus this month's local roads\n",
    "    this_month_local_roads = pd.concat([cut_linearid_1, cut_linearid_2], axis = 0)\n",
    "    this_month_len = this_month_local_roads.geometry.length.sum()\n",
    "    last_month_len = last_month_gdf.geometry.length.sum()\n",
    "    print(f\"This month's local roads length: {this_month_len}. Last month: {last_month_len}. Diff: {last_month_len-this_month_len}\")\n",
    "    \n",
    "    # Read in primary & secondary roads that have already been cut\n",
    "    primary_secondary = gpd.read_parquet(f\"{SHARED_GCS}segmented_primary_secondary_roads.parquet\")\n",
    "    \n",
    "    # Concat everything\n",
    "    this_month_segmented = pd.concat([cut_linearid_1, cut_linearid_2, primary_secondary],axis=0)\n",
    "    \n",
    "    # Save\n",
    "    file_date = date.replace('-','_')\n",
    "    this_month_local_roads.to_parquet(f\"{SHARED_GCS}segmented_local_roads_{file_date}.parquet\")\n",
    "    this_month_segmented.to_parquet(f\"{SHARED_GCS}segmented_all_roads_{file_date}.parquet\")\n",
    "    \n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done: {end-start}\")\n",
    "    return this_month_segmented"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "26394e93-201a-4605-8af0-a19763a25dc7",
   "metadata": {},
   "outputs": [],
   "source": [
    "april_month = \"2023-04-12\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "24990a57-3eeb-4352-9a35-5660b10ae992",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start: 2023-05-26 14:34:02.338678\n",
      "Begin sjoin\n",
      "Done buffering\n",
      "Done with sjoin with stops with local roads. Time lapsed: 0:03:01.625797\n",
      "Done with sjoin with routes and local roads. Time lapsed: 0:09:48.755078\n",
      "Done with doing an sjoin for all local roads. Time lapsed: 0:22:02.441069\n",
      "There are 11418 new linear ids found this month that didn't appear last month.\n",
      "There are 8673 that didn't appear this month that will be deleted.\n",
      "Begin cutting local roads\n",
      "Split into mulitple parquets\n",
      "Cut geometry\n",
      "Begin computing\n",
      "Done cutting local roads in 0:02:32.191740 minutes\n",
      "This month's local roads length: 174846164.15138897. Last month: 171163870.38026056. Diff: -3682293.771128416\n",
      "Done: 0:28:42.743699\n"
     ]
    }
   ],
   "source": [
    "april_df = monthly_linearids(april_month, \"segmented_local_rds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ff583c3-1454-4596-9d6a-ada72b7f7363",
   "metadata": {},
   "outputs": [],
   "source": [
    "#march_linearids = set(march_linearis.linearid.unique().tolist())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "305b558f-0d22-4d7c-bdbf-97673036fbd0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_ops = sjoin_local_roads(april_month, \"April_2023\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98abf9a3-428b-47ef-9c21-7ade8da4fefe",
   "metadata": {},
   "outputs": [],
   "source": [
    "#april_linearids = set(april_linearid.linearid.unique().tolist())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c64e216-7560-4fb6-a4d5-7b370fa20801",
   "metadata": {},
   "outputs": [],
   "source": [
    "#len(april_linearids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "579d95e9-82c2-4860-b014-3bb153a3f936",
   "metadata": {},
   "outputs": [],
   "source": [
    "#len(march_linearids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "45785e34-07ca-4ec2-afa1-a60e7183ce9e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#new_linearids_to_cut = list(april_linearids - march_linearids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da00498d-9c91-4e24-8f25-517013dd4f56",
   "metadata": {},
   "outputs": [],
   "source": [
    "# linearids found in this month but not last month's\n",
    "#f\"{len(new_linearids_to_cut)} new linearids\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a217419e-4a86-4678-bd60-8da41b5aac26",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Linearids only found in last month but not this month's\n",
    "#linearids_to_delete = list(march_linearids - april_linearids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e1806cfb-d0c6-4225-a63c-2065c7267000",
   "metadata": {},
   "outputs": [],
   "source": [
    "#f\"{len(linearids_to_delete)} linearids to be deleted\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4848541-5cbc-4d80-b15b-f4b0abe292f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete off roads \n",
    "#march_all_segmented_roads = gpd.read_parquet(f\"{SHARED_GCS}segmented_all_roads.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4cc2060d-56eb-47ab-ba38-53af802ead0a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# arch_all_segmented_roads.sample()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e67d6199-ab22-4d4c-b469-d02dc56ae112",
   "metadata": {},
   "outputs": [],
   "source": [
    "#april_segmented_roads = march_all_segmented_roads[~march_all_segmented_roads.linearid.isin(linearids_to_delete)].reset_index(drop = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36dca459-34db-497c-b905-5983b8d0ebb0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#march_all_segmented_roads.linearid.nunique() - april_segmented_roads.linearid.nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f31a758b-d19f-4de1-8897-88955cfd65ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Find the linearids from last month\n",
    "#april_roads_to_cut = (gpd\n",
    " #             .read_parquet(f\"{SHARED_GCS}local_roads_all_routes_stops_sjoin_April_2023.parquet\")\n",
    " #            )\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "721080c2-6a42-49a8-b0c3-f0dad2d0be11",
   "metadata": {},
   "outputs": [],
   "source": [
    "# april_roads_to_cut = april_roads_to_cut.loc[april_roads_to_cut.linearid.isin(new_linearids_to_cut)].reset_index(drop = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8151a681-aafc-4a80-8b10-55c1d7489ded",
   "metadata": {},
   "outputs": [],
   "source": [
    "# len(april_roads_to_cut)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "de1c173c-d8e0-4ce2-b5bb-7fe2982c4dc8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# april_cut_segments = cut_local_roads_monthly(april_roads_to_cut, april_month)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a8f7592-34dd-4c36-b986-3148a2f9147d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type(april_cut_segments)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b95f934-6331-4913-963f-beb5e39d44cc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# april_local_roads = pd.concat([april_segmented_roads, april_cut_segments], axis=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "15f4522c-7201-4777-89d8-9326bebb06a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# april_local_roads.sample()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76fc064d-991c-4bd1-aa7d-8cf36dac1630",
   "metadata": {},
   "source": [
    "### Concat local roads and primary/secondary ones"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "89d85183-516f-4b88-a6c7-5da78281fc48",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_primary_secondary_roads() -> gpd.GeoDataFrame:\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Cutting primary/secondary roads\")\n",
    "\n",
    "    # Find all primary and secondary roads\n",
    "    # regardless of intersection w/ GTFS shapes\n",
    "    primary_secondary_mtfcc = [\"S1100\", \"S1200\"]\n",
    "    primary_secondary_roads = load_roads(primary_secondary_mtfcc)\n",
    "\n",
    "    segments = geography_utils.cut_segments(\n",
    "        primary_secondary_roads, [\"linearid\", \"fullname\"], 1_000  # 1 km segments\n",
    "    )\n",
    "\n",
    "    segments.to_parquet(f\"{SHARED_GCS}segmented_primary_secondary_roads.parquet\")\n",
    "\n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done cutting primary & secondary roads: {end-start}\")\n",
    "    return segments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d2a7a9c-1102-4a3c-a673-cd4cba2a204b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# primary_secondary = cut_primary_secondary_roads()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44b2ee39-e968-48a5-aa81-d9ec29e670df",
   "metadata": {},
   "source": [
    "### Cut everything from top to bottom"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c0ff123b-81fa-44b3-825d-ea9b8b68625d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_all_roads(date:str) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Cut all roads: primary, secondary, and primary roads\n",
    "    that overlap with bus routes.\n",
    "    \n",
    "    Takes about 1.5 hours.\n",
    "    date (str): analysis date\n",
    "    \"\"\"\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Start cutting all roads: {start}\")\n",
    "    # Find local roads that intersect  with GTFS shapes, then\n",
    "    # segment them\n",
    "    local_roads_unsegmented = sjoin_local_roads(date)\n",
    "    local_roads_gdf = cut_local_roads(local_roads_unsegmented, date)\n",
    "\n",
    "    # Segment primary and secondary roads\n",
    "    segmented_primary_secondary_rds = cut_primary_secondary_roads()\n",
    "\n",
    "    # Concat\n",
    "    file_date = date.replace('-','_')\n",
    "    all_roads = pd.concat([segmented_primary_secondary_rds, local_roads_gdf], axis=0)\n",
    "    all_roads.to_parquet(f\"{SHARED_GCS}segmented_all_roads_{file_date}.parquet\")\n",
    "\n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Time lapsed for cutting all roads: {end-start}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0bfe3493-3988-4cc1-9220-a37e61dc61d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_roads = cut_all_roads(analysis_date)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c213fc28-68eb-431a-ab8b-89fd17e28a43",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_cut_roads = gpd.read_parquet(\"gs://calitp-analytics-data/data-analyses/shared_data/segmented_all_roads.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "64c52831-17fe-4f92-976f-1199cf7f775c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#all_cut_roads.linearid.nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "18bfd78b-41fd-45f7-bd72-37d6b2c7aba0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#all_cut_roads.linearid.value_counts().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "229482ad-851d-4e26-86fa-b70baeeeed13",
   "metadata": {},
   "outputs": [],
   "source": [
    "#all_cut_roads.linearid.value_counts().describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d700c63-15b0-44d5-9974-c50d4edbccac",
   "metadata": {},
   "outputs": [],
   "source": [
    "#linearid_subset = ['11018382472869','1105640135361']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "acd4d684-5fff-45b6-b475-2d78b8b40270",
   "metadata": {},
   "outputs": [],
   "source": [
    "#all_cut_roads.head(100).explore('segment_sequence', cmap = 'tab10', style_kwds = {'weight':10}, legend = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fe813210-3a5d-431b-ae28-6058b33122e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# all_cut_roads.loc[all_cut_roads.linearid.isin(linearid_subset)].explore('segment_sequence', cmap = 'tab10', legend = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf04f211-f2a1-419e-9482-cfad94a26e9d",
   "metadata": {},
   "source": [
    "### If Main ??\n",
    "* As of 5/18, takes 1.5 hours.\n",
    "* "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9901a04f-b428-410a-ba7a-c4891817b24f",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_all_or_month(date:str, last_month_segmented_local_roads: str, run_monthly:bool = True):\n",
    "    if run_monthly:\n",
    "        gdf1 = monthly_linearids(date, last_month_segmented_local_roads)\n",
    "    else:\n",
    "        gdf2 = cut_all_roads(date)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5b07078-1b2c-4153-986c-9618ee632698",
   "metadata": {},
   "outputs": [],
   "source": [
    "analysis_date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4bfca5f3-edd5-4adb-ab4e-574f04932dad",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Cut all the roads top to bottom\n",
    "test = cut_all_or_month(analysis_date, \"\", False)\n",
    "\n",
    "Start cutting all roads: 2023-05-18 10:30:27.686491\n",
    "Begin sjoin\n",
    "Done buffering\n",
    "Done with sjoin with stops with local roads. Time lapsed: 0:03:20.079432\n",
    "Done with sjoin with routes and local roads. Time lapsed: 0:07:06.883445\n",
    "Done with doing an sjoin for all local roads. Time lapsed: 0:18:44.197788\n",
    "Begin cutting local roads\n",
    "Split into mulitple parquets\n",
    "Cut geometry\n",
    "Begin computing\n",
    "Done cutting local roads in 1:04:27.639005 minutes\n",
    "Cutting primary/secondary roads 2023-05-18 11:53:45.703569\n",
    "Done cutting primary & secondary roads: 0:05:24.448465\n",
    "Time lapsed for cutting all roads: 1:29:23.356601\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d296f8d-acea-477e-977c-49d464d162e4",
   "metadata": {},
   "outputs": [],
   "source": [
    "april_month"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c6673a43-4682-4e8e-8f1f-35d2b9c03860",
   "metadata": {
    "scrolled": true,
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Takes about 23 minutes\n",
    "# test2 = cut_all_or_month(april_month, \"segmented_all_roads_2023_03_15\", True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a0248cbe-eb4e-4349-8d9e-1b8d42eb91c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "april = gpd.read_parquet(\"gs://calitp-analytics-data/data-analyses/shared_data/segmented_all_roads_2023_04_12.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1fafff4e-35e1-4119-984c-d4676684dcbe",
   "metadata": {},
   "outputs": [],
   "source": [
    "april.linearid.nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "564cde67-7e8e-43b1-8b0d-c2833652c02e",
   "metadata": {},
   "outputs": [],
   "source": [
    "april.sample()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db1db014-a40b-4961-ad8b-89786bacd438",
   "metadata": {},
   "outputs": [],
   "source": [
    "april.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff99ab01-113a-4fd0-baa7-5a60be6cf725",
   "metadata": {},
   "outputs": [],
   "source": [
    "april.linearid.value_counts().head(30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aa45b63d-6367-4da7-a33a-2b630a20dea7",
   "metadata": {},
   "outputs": [],
   "source": [
    "april2 = april.loc[april.linearid.isin([\"11012815158651\"])].reset_index(drop = True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "555bd6fa-f797-40b3-8060-875294ac57ec",
   "metadata": {},
   "source": [
    "### Draft"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7cf61650-c7e7-4575-bfe9-6e424f1a2847",
   "metadata": {},
   "outputs": [],
   "source": [
    "def cut_local_roads(date, chunk_row_size: int) -> gpd.GeoDataFrame:\n",
    "    start = datetime.datetime.now()\n",
    "    print(f\"Cut local roads {start}\")\n",
    "\n",
    "    # Find all local roads that intersect with\n",
    "    # stops and routes.\n",
    "    local_roads_unsegmented = sjoin_local_roads(date)\n",
    "\n",
    "    # Divide the gdf into equal sized chunks (roughly)\n",
    "    # and turn them into dask gdfs\n",
    "    ddfs, length = chunk_dask_df(local_roads_unsegmented, chunk_row_size)\n",
    "\n",
    "    # Split the list of split dask dataframes\n",
    "    # into half.\n",
    "    length_list = [*range(0, length)]\n",
    "    ddf1 = length_list[: len(length_list) // 2]\n",
    "    ddf2 = length_list[len(length_list) // 2 :]\n",
    "\n",
    "    # Cut geometry\n",
    "    part1 = dask_segment(ddfs, ddf1)\n",
    "    part1.to_parquet(f\"{SHARED_GCS}segmented_local_rds_first_pt.parquet\")\n",
    "    print(\"Done with cutting part1\")\n",
    "\n",
    "    part2 = dask_segment(ddfs, ddf2)\n",
    "    part2.to_parquet(f\"{SHARED_GCS}segmented_local_rds_second_pt.parquet\")\n",
    "    print(\"Done with cutting part2\")\n",
    "\n",
    "    segmented_local_roads = pd.concat([part1, part2])\n",
    "    segmented_local_roads.to_parquet(f\"{SHARED_GCS}segmented_local_rds.parquet\")\n",
    "\n",
    "    end = datetime.datetime.now()\n",
    "    print(f\"Done cutting local roads in {end-start} minutes\")\n",
    "    return segmented_local_roads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f8048ab-d3a9-4a33-9aeb-46879f341a7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "def dask_segment(ddf_list: list, ddfs_range: list) -> gpd.GeoDataFrame:\n",
    "    \"\"\"\n",
    "    Use dask to cut roads into segments. Compute the results\n",
    "    back to a GDF.\n",
    "\n",
    "    Args:\n",
    "        ddf_list: dask dataframes stored in a list.\n",
    "        ddf_list[0] will yield a ddf.\n",
    "\n",
    "        ddfs_range: how many items are in the ddf_list.\n",
    "    \"\"\"\n",
    "    # Empty dataframe\n",
    "    my_results = []\n",
    "\n",
    "    # For each dask dataframe int the list\n",
    "    # cut them and append the results into the empty df.\n",
    "    for i in ddfs_range:\n",
    "        my_df = ddf_list[i]\n",
    "        cut_geometry = delayed(geography_utils.cut_segments)(\n",
    "            my_df, [\"linearid\", \"fullname\"], 1_000\n",
    "        )\n",
    "        my_results.append(cut_geometry)\n",
    "        print(f\"done with {i}\")\n",
    "\n",
    "    # Compute results into a normal gdf\n",
    "    compute_results = [compute(i)[0] for i in my_results]\n",
    "\n",
    "    # Concat results\n",
    "    results_gdf = pd.concat(compute_results)\n",
    "\n",
    "    return results_gdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fa773662-a21c-4ac0-882a-d597b0570570",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read in sjoined files\n",
    "local_roads_unsegmented = gpd.read_parquet(\n",
    "    \"gs://calitp-analytics-data/data-analyses/shared_data/local_roads_all_routes_stops_sjoin.parquet\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1b08a852-2d20-4c01-bd5f-7ff8048b8c45",
   "metadata": {},
   "outputs": [],
   "source": [
    "local_roads_unsegmented.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92ff276e-dd4e-4030-9b48-eadc44b14b2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Turn sjoin local roads to dask\n",
    "ddf1 = dd.from_pandas(local_roads_unsegmented, npartitions=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7a78b420-94db-40ad-9327-7d38882e579b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Partition the sjoin stuff automatically\n",
    "ddf1_partitioned = ddf1.repartition(partition_size=\"1MB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79265a7a-aa75-4005-a625-8cffc4373205",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Save\n",
    "ddf1_partitioned.to_parquet(f\"{SHARED_GCS}daskutilstest\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "702a3e12-19bb-4cb0-a8c5-01b324c0634c",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5042b67-0e65-4f10-a393-0da0e4240b9a",
   "metadata": {},
   "outputs": [],
   "source": [
    "file_names_dask = extract_number(\"part\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e803160-d428-499a-8b54-9538d27856b4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# https://www.geeksforgeeks.org/read-multiple-csv-files-into-separate-dataframes-in-python/\n",
    "# create empty list\n",
    "dataframes_list = []\n",
    " \n",
    "# append datasets into the list\n",
    "for i in range(len(file_names)):\n",
    "    gcs_file_path = \"gs://calitp-analytics-data/data-analyses/shared_data/daskutilstest/part.\"\n",
    "    temp_df = dg.read_parquet(f\"{gcs_file_path}{file_names[i]}.parquet\")\n",
    "    dataframes_list.append(temp_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "68a5c51e-50cc-4bbf-8a18-afc078f69b76",
   "metadata": {},
   "outputs": [],
   "source": [
    "len(dataframes_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f68a473-1ea6-4967-b37e-01d4093da01f",
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframes_list[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2500651b-0560-4497-a3aa-d8f353cb1699",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Empty dataframe\n",
    "my_results = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ba28efc6-3363-4333-93f0-e17bd51ee92f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# For each dask dataframe int the list\n",
    "# cut them and append the results into the empty df.\n",
    "for ddf in dataframes_list:\n",
    "    cut_geometry = delayed(geography_utils.cut_segments)(ddf, [\"linearid\", \"fullname\"], 1_000)\n",
    "    my_results.append(cut_geometry)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2bebed4c-383f-4e96-a77f-7859ef6422b8",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b8e081d5-5755-481c-ad7f-0e20c951ff8e",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "203a1b98-1788-4f53-b9c0-08de21301de4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Took 1:06:37.299317\n",
    "#start = datetime.datetime.now()\n",
    "#print(start)\n",
    "#results2 = [compute(i)[0] for i in my_results]  # 9;12\n",
    "#end = datetime.datetime.now()\n",
    "#print(end)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d6e3ea1d-649c-47aa-8ce8-227cfdcb322a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# print(end-start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "223e08ec-0db3-4977-bd31-199e1f54f679",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type(results2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dc6d6418-d779-4eed-9f85-c65bb84512bd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# testpd = pd.concat(results2, axis=0).reset_index(drop=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1d46e2f-e94b-47a7-9912-543fc67346fc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# type(testpd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2cceec4-f080-467b-b3bc-106b09033aa4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# testpd.to_parquet(f\"{SHARED_GCS}dask_test2.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f552ba87-85c9-4f65-9a98-e003720246ef",
   "metadata": {},
   "outputs": [],
   "source": [
    "# testpd.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "beb6c1ca-ae26-40ed-89c9-53ac2c712e9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2 = gpd.read_parquet(\"gs://calitp-analytics-data/data-analyses/shared_data/dask_test2.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "042fac67-3fa5-49c7-8b5f-d2de023c4e42",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2.linearid.nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4b83270a-1ec3-4e6b-91ee-4f291165f009",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2.linearid.value_counts().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98891845-c705-4266-af51-1e931dce28ca",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_linearids = [ '11019653760031',\n",
    " '1106092764328',\n",
    " '11011135052903',\n",
    " '110411099535',\n",
    " '11011135055229',\n",
    " '11011135055553',\n",
    " '11011135056214',\n",
    " '11012028306122',\n",
    " '11012812027422',\n",
    " '11012812027505',\n",
    " '11012812035943',\n",
    " '11012812038881',\n",
    " '1106073054809']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0234451d-1006-4f7d-bc27-59559c8bc159",
   "metadata": {},
   "outputs": [],
   "source": [
    "og_tiger = load_roads(['S1400'], False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b78d9d13-b64e-459a-8d05-5f51b9c0bbb6",
   "metadata": {},
   "outputs": [],
   "source": [
    "og_tiger_filtered = og_tiger[og_tiger.linearid.isin(test_linearids)].reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e4374a26-ed5e-473a-acdf-95800bb04680",
   "metadata": {},
   "outputs": [],
   "source": [
    "# og_tiger_filtered.explore('linearid',cmap = \"tab20c\", style_kwds = {'weight':5})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f27bbd0-3962-4fcc-8d32-63d814c91476",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2_filtered = dask2[dask2.linearid.isin(test_linearids)].reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f4174574-d5d3-4839-b669-7debb8bac4b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2_filtered.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "704ee9d5-25db-49b4-ac1f-2ae9119087e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask2_filtered.sample()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d9c767a-1028-4ef0-9c10-96c1fa08e832",
   "metadata": {},
   "outputs": [],
   "source": [
    "# dask2_filtered.explore('segment_sequence',cmap = \"tab20c\", style_kwds = {'weight':5})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6e48ae43-89d1-496e-bf64-6d7c23423156",
   "metadata": {},
   "outputs": [],
   "source": [
    "def chunk_dask_df(gdf, chunk_row_size: int):\n",
    "    \"\"\"\n",
    "    Break up dataframes by a certain\n",
    "    number of rows, turn them into a dask\n",
    "    dataframe\n",
    "\n",
    "    Args:\n",
    "        gdf: the local roads that intersect w/ stops and routes\n",
    "        chunk_row_size(int): how many rows each dataframe should\n",
    "        be after splitting it out.\n",
    "\n",
    "    Returns:\n",
    "        List of dask dataframes. Length of how many dask dataframes\n",
    "        are returned after cutting.\n",
    "    \"\"\"\n",
    "    # Specify how many rows I want the gdf to broken into per df\n",
    "    n = chunk_row_size\n",
    "\n",
    "    # Break it out\n",
    "    list_df = [gdf[i : i + n] for i in range(0, gdf.shape[0], n)]\n",
    "\n",
    "    # Turn each dataframe to a dask one\n",
    "    my_ddfs = []\n",
    "    for df in list_df:\n",
    "        ddf = dd.from_pandas(df, npartitions=1)\n",
    "        my_ddfs.append(ddf)\n",
    "\n",
    "    return my_ddfs, len(my_ddfs)"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}