Build a Methane Inspection Pipeline

Introduction

30 Second Summary

Mobile survey data can look convincing on a map even when duplicate, missing, or malformed readings are mixed into the feed. Inspection teams need a clear way to separate trustworthy signals from noise before sending vehicles into the field.

In this project, you will build a local Python workflow that turns imperfect synthetic methane telemetry into a tested two-van inspection plan. You will present the result on an interactive operations map.

What You'll Build

You will launch a dashboard that transforms raw survey readings into a color-coded depot-to-depot inspection plan.

By the end of this project, you'll have:

• An ingest report showing 20 accepted records plus 3 rejected records with explicit reasons.
• An interactive operations map linking every accepted reading to its nearest fictional pipeline. Three DBSCAN plume candidates display centroids plus methane summaries.
• A two-van dispatch plan that visits every plume candidate before returning to the depot. An automated test suite verifies the workflow's critical transformations.
• Secret Mission: Add dashboard controls for the methane threshold and DBSCAN radius. Watch the plume count plus inspection routes recompute from your settings.

Are there any prerequisites?

You need a Windows x86_64 computer with internet access. The first step verifies Python 3.11 or newer plus 64-bit architecture before installing the required packages.

Before We Start

Before you begin building, take a moment to lock in the purpose of GeoMethane-Ops. You are committing to turn imperfect mobile methane telemetry into defensible inspection priorities for a utility operator.

Prepare the Windows Python Environment

GeoMethane-Ops depends on a compiled Python geospatial stack. An incompatible interpreter can turn an environment problem into a failure halfway through the pipeline.

Here, you will verify a 64-bit interpreter running version 3.11 or newer. You will isolate every pinned package inside a virtual environment so the remaining steps use a consistent toolchain.

In this step, get ready to:

• Confirm the Python interpreter meets the version and architecture requirements.
• Create an isolated environment with every pinned dependency.
• Prove Google OR-Tools loads from the active environment.

Verify Python and the interpreter architecture

The pinned package stack requires Python 3.11 or newer. The routing dependency also needs a 64-bit Windows interpreter.

• Press the Windows key to open the search bar.
• Search for File Explorer.
• Select File Explorer from the search results.
• Navigate to the GeoMethane-Ops project folder.
• Right-click the GeoMethane-Ops project folder.
• Select Open in Terminal.
• Choose the Windows PowerShell profile if another terminal profile is active.
• Check the installed Python version by running this command:

python --version

What does this check do?

This command asks the active Python interpreter to report its version. The package pins in this project require version 3.11 or newer.

✔️ I see version 3.11 or higher

Good. Your installed Python version meets the project minimum.

• Continue to the architecture check below.

ⓧ I see an older version

Your current interpreter is below the minimum required by the pinned packages. Install a 64-bit Python release at version 3.11 or newer before continuing.

• Open the official Python downloads for Windows page.
• Download a 64-bit Windows installer for Python 3.11 or newer.
• Run the downloaded installer.
• Choose the installer options that enable launcher support plus command-path access.
• Close the current terminal after installation finishes.
• Return to the GeoMethane-Ops project folder in File Explorer.
• Select Open in Terminal from the folder context menu.
• Confirm the new Python version by running this command:

python --version

What confirms the upgrade?

The reported version should now be 3.11 or newer. This confirms that PowerShell can reach a compatible interpreter.

ⓧ Command not found

PowerShell cannot find a Python interpreter through its current command path. Install a 64-bit Python release with launcher support.

• Open the official Python downloads for Windows page.
• Download a 64-bit Windows installer for Python 3.11 or newer.
• Run the downloaded installer.
• Choose the installer options that enable launcher support plus command-path access.
• Close the current terminal after installation finishes.
• Return to the GeoMethane-Ops project folder in File Explorer.
• Select Open in Terminal from the folder context menu.
• Confirm Python is available by running this command:

python --version

What confirms the installation?

A version number confirms that PowerShell can now find Python. The reported version must be 3.11 or newer for this project.

• Check whether the interpreter uses 64-bit architecture by running this command:

python -c "import platform; print(platform.architecture()[0])"

Why check the architecture?

This command reads the architecture of the active interpreter. A 64-bit interpreter can load the native routing libraries required by OR-Tools on Windows.

You should see 64bit in the output. That result confirms the interpreter architecture is compatible.

Not seeing a 64-bit result?

• Install a 64-bit build from the official Python downloads for Windows page.
• Open a fresh PowerShell terminal in the GeoMethane-Ops project folder after installation.
• Repeat the architecture check before creating the environment.

Ask for help diagnosing the interpreter selected by PowerShell: Why is my Windows PowerShell session using the wrong Python architecture?

Create the environment and install dependencies

An isolated environment keeps this project's package versions separate from other Python projects. The activation step makes its interpreter the active one in your current PowerShell session.

• Create .venv and activate it by running these commands:

py -m venv .venv
.venv\Scripts\activate

What do these commands do?

• The first command uses the Windows Python launcher to create an isolated environment inside .venv.
• The second command activates that environment for the current PowerShell session.

Your PowerShell prompt should now include the environment name .venv. That visible prompt change confirms activation.

Environment not active?

• Confirm PowerShell is still inside the GeoMethane-Ops project folder.
• Rerun the environment-creation command above if the .venv folder was not created.
• Use the execution policy approved for your machine if PowerShell blocks the activation script.

Ask for help with the activation failure: Why will my .venv environment not activate in Windows PowerShell?

The dependency file gives every learner the same package versions. This makes later spatial operations and routing results reproducible.

• Press the Windows key to open the search bar.
• Search for Visual Studio Code.
• Select Visual Studio Code from the search results.
• Use the folder picker to select the GeoMethane-Ops project folder.
• Create requirements.txt inside the GeoMethane-Ops project folder with this content:

pandas==3.0.5
geopandas==1.1.4
shapely==2.1.2
scikit-learn==1.9.1
ortools==9.15.6755
streamlit==1.63.0
folium==0.20.0
streamlit-folium==0.27.4
pytest==9.1.1

What does this file install?

• pandas loads the telemetry table.
• GeoPandas manages projected spatial data.
• Shapely calculates the exact point-to-pipeline connectors.
• scikit-learn provides DBSCAN clustering.
• Streamlit serves the local operations dashboard.
• Folium renders the interactive map.
• pytest runs the automated checks.

• Save requirements.txt.
• Confirm the file appears in the Visual Studio Code file sidebar with nine pinned package lines.

Dependency file missing?

• Confirm Visual Studio Code has the GeoMethane-Ops project folder selected.
• Check that the filename is exactly requirements.txt.
• Remove any extra file extension added by Windows.

Ask for help checking the file location: Why can PowerShell not find my requirements.txt file?

✔️ Awesome, I've got everything!

Your dependency file now matches the pinned stack for GeoMethane-Ops.

ⓧ I'd like to double check the full code

pandas==3.0.5
geopandas==1.1.4
shapely==2.1.2
scikit-learn==1.9.1
ortools==9.15.6755
streamlit==1.63.0
folium==0.20.0
streamlit-folium==0.27.4
pytest==9.1.1

How to use this reference

Compare each line with your saved requirements.txt file. The package names and version pins must match exactly.

The first dependency install can take several minutes while Python downloads the compiled packages. Scrolling output means the installation is still working.

• Return to the active PowerShell session in the GeoMethane-Ops project folder.
• Install every pinned dependency by running this command:

py -m pip install -r requirements.txt

What does this command do?

The Python launcher runs pip against the dependency file. The active environment keeps the installed packages inside .venv.

You should see installation progress for the pinned packages followed by a successful completion message. The active environment now contains the complete project stack.

Package installation failed?

• Confirm the PowerShell prompt still includes .venv.
• Confirm requirements.txt is saved inside the GeoMethane-Ops project folder.
• Rerun the installation after restoring your internet connection if a package download was interrupted.

Ask for help with the failed package installation: Why did my pinned GeoMethane-Ops dependencies fail to install?

Verify OR-Tools in the active environment

A successful package download does not prove that a compiled library can load. Importing OR-Tools checks the installation inside the active environment before the routing work begins.

Before you run the check, which OR-Tools version do you expect the active environment to report?

• Import OR-Tools and print its installed version by running this command:

python -c "import ortools; print(ortools.__version__)"

What does this check prove?

Python imports OR-Tools from the active environment. Printing the package version confirms that the routing library is available to later pipeline code.

You should see 9.15.6755. That is the hard setup complete: the isolated Windows environment can load the routing dependency.

OR-Tools will not import?

• Confirm the PowerShell prompt still includes .venv.
• Repeat the installation command if OR-Tools was skipped during the package install.
• Follow Google's Windows installation guide to install the x64 Microsoft Visual C++ Redistributable for Visual Studio 2022 if the import reports a native-library problem.

Ask for help with the native dependency: Why can my active .venv environment not import OR-Tools on 64-bit Windows?

Your Windows Python environment is ready for geospatial analysis and vehicle routing. Next, you will put the synthetic methane readings and fictional utility lines onto an interactive operations map.

Map Raw Telemetry

That setup work paid off. Your 64-bit Python environment now has every package needed to turn survey records into an operations dashboard.

A map can make unreviewed data look convincing. In this step, you will use Streamlit with Folium to plot every raw reading beside a fictional pipeline network.

In this step, get ready to:

• Create a CSV fixture containing 23 synthetic telemetry records.
• Add a fictional pipeline network as GeoJSON data.
• Build a raw map in Visual Studio Code that displays every reading beside four pipeline segments.

Create the telemetry fixture

The telemetry fixture represents two survey vehicles collecting location and methane readings. Three deliberately flawed rows remain mixed into the file so you can see how a raw dashboard treats them.

• Open the current GeoMethane-Ops folder in Visual Studio Code by running this command in the activated PowerShell session:

code .

What does this command do?

The code . command opens the current PowerShell folder as a Visual Studio Code workspace. The period refers to the folder you are already inside.

Visual Studio Code Did Not Open?

Your terminal may not have the Visual Studio Code command on its path. Open Visual Studio Code through Windows search instead.

Once it opens, select the GeoMethane-Ops folder shown in PowerShell. Help me open my current PowerShell folder in Visual Studio Code.

• Create a folder named data inside the GeoMethane-Ops folder using the file tree controls.
• Create telemetry.csv inside the data folder.
• Fill data/telemetry.csv by pasting this dataset:

reading_id,timestamp,vehicle_id,latitude,longitude,methane_ppm,speed_kph
R001,2026-09-12T08:00:00Z,SURVEY-01,37.7950,-122.2630,1.91,28
R002,2026-09-12T08:00:10Z,SURVEY-01,37.7960,-122.2600,1.95,27
R003,2026-09-12T08:00:20Z,SURVEY-01,37.7970,-122.2570,2.01,26
R004,2026-09-12T08:00:30Z,SURVEY-01,37.8000,-122.2510,3.70,18
R005,2026-09-12T08:00:40Z,SURVEY-01,37.8004,-122.2506,3.95,16
R006,2026-09-12T08:00:50Z,SURVEY-01,37.8008,-122.2502,4.10,14
R007,2026-09-12T08:01:00Z,SURVEY-01,37.8011,-122.2498,3.80,15
R008,2026-09-12T08:01:10Z,SURVEY-01,37.8040,-122.2460,2.04,24
R009,2026-09-12T08:01:20Z,SURVEY-01,37.8070,-122.2400,1.98,25
R010,2026-09-12T08:01:30Z,SURVEY-01,37.8110,-122.2250,4.30,17
R011,2026-09-12T08:01:40Z,SURVEY-01,37.8113,-122.2245,4.50,15
R012,2026-09-12T08:01:50Z,SURVEY-01,37.8117,-122.2240,4.15,13
R013,2026-09-12T08:02:00Z,SURVEY-01,37.8120,-122.2236,4.70,12
R014,2026-09-12T08:02:10Z,SURVEY-02,37.8140,-122.2180,1.96,29
R015,2026-09-12T08:02:20Z,SURVEY-02,37.8170,-122.2140,2.00,30
R016,2026-09-12T08:02:30Z,SURVEY-02,37.7888,-122.2360,3.60,19
R017,2026-09-12T08:02:40Z,SURVEY-02,37.7892,-122.2354,3.90,18
R018,2026-09-12T08:02:50Z,SURVEY-02,37.7896,-122.2349,4.20,16
R019,2026-09-12T08:03:00Z,SURVEY-02,37.7900,-122.2345,3.80,17
R020,2026-09-12T08:03:10Z,SURVEY-02,37.7920,-122.2300,2.02,23
R020,2026-09-12T08:03:20Z,SURVEY-02,37.7923,-122.2295,2.06,22
R021,not-a-time,SURVEY-02,37.8050,-122.2410,2.10,20
R022,2026-09-12T08:03:40Z,SURVEY-02,37.8060,-122.2380,,24

What Does This Dataset Contain?

• Each row combines a reading identifier with a timestamp.
• The coordinates provide the marker position for each survey reading.
• The methane and speed fields provide measurements captured at that position.
• The repeated R020 identifier creates a duplicate record.
• The R021 row contains the invalid timestamp not-a-time.
• The R022 row has a blank methane_ppm value.

• Save data/telemetry.csv.
• Check the editor shows one header row followed by 23 telemetry records.
• Confirm the final record begins with R022.

Telemetry Rows Look Misaligned?

Make sure every record stays on one line. Each field must remain separated by a comma.

Keep the empty space between the two commas in the R022 row because it represents the missing methane value. Help me check the structure of my telemetry CSV.

Add the fictional pipeline network

Each pipeline is a line made from longitude and latitude positions. The file uses WGS 84 coordinates so Folium can place the lines directly on its basemap.

• Create pipelines.geojson inside the data folder.
• Fill data/pipelines.geojson by pasting this network:

{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "properties": {"pipeline_id": "P-100", "pressure_class": "medium"},
      "geometry": {"type": "LineString", "coordinates": [[-122.268, 37.800], [-122.242, 37.800]]}
    },
    {
      "type": "Feature",
      "properties": {"pipeline_id": "P-200", "pressure_class": "low"},
      "geometry": {"type": "LineString", "coordinates": [[-122.247, 37.785], [-122.225, 37.796]]}
    },
    {
      "type": "Feature",
      "properties": {"pipeline_id": "P-300", "pressure_class": "medium"},
      "geometry": {"type": "LineString", "coordinates": [[-122.232, 37.805], [-122.212, 37.820]]}
    },
    {
      "type": "Feature",
      "properties": {"pipeline_id": "P-400", "pressure_class": "high"},
      "geometry": {"type": "LineString", "coordinates": [[-122.262, 37.790], [-122.218, 37.814]]}
    }
  ]
}

How Is the Network Structured?

• The FeatureCollection groups all four pipeline features into one file.
• Each LineString contains positions ordered as longitude followed by latitude.
• The pipeline_id property gives each fictional asset a stable identifier.
• The pressure_class property supplies descriptive map metadata.

• Save data/pipelines.geojson.
• Confirm the editor shows pipeline identifiers P-100 through P-400.
• Check that every geometry uses the value LineString.

GeoJSON Showing a Syntax Problem?

Check the commas between the four feature objects. Confirm the file ends with both closing square and curly brackets.

Keep longitude first inside every coordinate pair. Help me find the syntax problem in my pipeline GeoJSON.

Build and test the raw map

The first dashboard slice proves both fixture files can be loaded before you draw the map. It gives you a quick count for each source so file problems surface immediately.

• Create streamlit_app.py inside the GeoMethane-Ops folder.
• Start streamlit_app.py with this data-loading code:

from pathlib import Path

import folium
import geopandas as gpd
import pandas as pd
import streamlit as st
from streamlit_folium import st_folium

# Anchor the fixture paths to this project folder.
DATA_DIR = Path(__file__).resolve().parent / "data"
raw = pd.read_csv(DATA_DIR / "telemetry.csv")
pipelines = gpd.read_file(DATA_DIR / "pipelines.geojson")

# Surface both source counts before drawing the map.
st.set_page_config(page_title="GeoMethane-Ops", layout="wide")
st.title("GeoMethane-Ops")
st.caption("Synthetic operations demo with unvalidated telemetry.")
metric_columns = st.columns(2)
metric_columns[0].metric("Raw telemetry records", len(raw))
metric_columns[1].metric("Pipeline segments", len(pipelines))

What Does This Code Do?

• The path setup keeps both fixture locations tied to streamlit_app.py.
• Pandas loads all 23 CSV records without evaluating their quality.
• GeoPandas loads the four pipeline features with their geometries.
• The two metrics expose the raw source counts in the browser.

• Save streamlit_app.py.
• Return to the activated PowerShell session.
• Launch the first dashboard slice by running:

streamlit run streamlit_app.py

What Does This Command Do?

Streamlit starts a local dashboard server for streamlit_app.py. Your browser loads the page generated by the Python script.

You should see 23 raw telemetry records beside 4 pipeline segments. Your two fixture files are loading successfully.

Dashboard Did Not Load?

• Confirm the activated environment marker still appears in PowerShell.
• Check that streamlit_app.py sits beside the data folder.
• Check that both data filenames match the Python paths exactly.

Help me diagnose why my first GeoMethane-Ops Streamlit page does not show 23 records and four pipelines.

The next chunk turns the pipeline geometry into a visible map layer. The default basemap needs an internet connection to download its tiles.

• Return to streamlit_app.py in Visual Studio Code.
• Add this map code below the two metric lines:

# Center the operations view over the synthetic survey area.
operations_map = folium.Map(
    location=[37.802, -122.240],
    zoom_start=14,
    tiles="CartoDB Positron",
)

# Draw the fictional assets as one pipeline layer.
folium.GeoJson(
    pipelines,
    name="Synthetic utility pipelines",
    style_function=lambda feature: {"color": "#475569", "weight": 5},
    tooltip=folium.GeoJsonTooltip(
        fields=["pipeline_id", "pressure_class"],
        aliases=["Pipeline", "Pressure class"],
    ),
).add_to(operations_map)

# Render the Folium map inside the Streamlit page.
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

How Does the Map Layer Work?

• The map starts at a fixed center and zoom level covering the fixture coordinates.
• The GeoJSON layer draws every pipeline with the same visible line style.
• The tooltip exposes each pipeline identifier and pressure class.
• The final call embeds the completed Folium object in the Streamlit page.

• Save streamlit_app.py.
• Return to the dashboard in your browser.
• Reload the dashboard if the saved change has not appeared.

You should see four dark pipeline segments crossing the map. Hover over a segment to see its pipeline identifier and pressure class.

Map or Pipelines Missing?

A blank basemap often means the browser cannot download its tiles. Check your internet connection while keeping the dashboard server running.

If the basemap appears without pipeline lines, inspect the terminal for a GeoJSON parsing error. Help me troubleshoot my missing Folium pipeline layer.

The pipeline layer provides geographic context. The final chunk plots every raw telemetry row without filtering it first.

• Find the st_folium( call near the bottom of streamlit_app.py.
• Add this marker loop directly above that call:

# Plot every source row so the raw-data problem stays visible.
for row in raw.itertuples():
    methane_text = (
        "missing"
        if pd.isna(row.methane_ppm)
        else f"{row.methane_ppm:.2f} ppm"
    )
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=5,
        color="#2563eb",
        fill=True,
        fill_color="#2563eb",
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.timestamp} | {methane_text}"
        ),
    ).add_to(operations_map)

What Does the Marker Loop Show?

• The loop creates one marker from each raw CSV row.
• The tooltip combines the reading identifier with its timestamp and methane value.
• The missing methane value receives visible text instead of numeric formatting.
• Every marker uses the same style because the data has not been classified.

• Save streamlit_app.py.
• Return to the running PowerShell session.
• Stop the current dashboard process by pressing Ctrl+C.

Before you run the completed raw dashboard, do you expect the flawed rows to look less trustworthy than the valid readings?

• Launch the completed raw map by running:

streamlit run streamlit_app.py

What Does This Check Prove?

The command reruns the saved dashboard source against both fixture files. It proves the full raw-map slice works from disk to browser.

You should see 23 blue telemetry markers with four dark pipeline segments. The raw map gives every record the same visual trust.

• Hover over the R021 marker to find the timestamp not-a-time.
• Hover over the R022 marker to find its missing methane value.
• Find the two separate markers carrying the identifier R020.
• Confirm none of the readings shows a nearest pipeline assignment or snap distance.

That shortfall is intentional. The map is working, but raw proximity and polished visuals cannot establish which records deserve operational trust.

Do You See Fewer Than 23 Markers?

Several points sit close together, so nearby markers can visually overlap. Zoom in around the survey area before assuming a row is missing.

If the dashboard metric is below 23, compare the CSV against the complete fixture. Help me find why my raw telemetry map has the wrong record count.

✔️ Awesome, I've got everything!

Great. Save streamlit_app.py and keep the virtual environment active.

ⓧ I'd like to double check the full code

from pathlib import Path

import folium
import geopandas as gpd
import pandas as pd
import streamlit as st
from streamlit_folium import st_folium

# Anchor the fixture paths to this project folder.
DATA_DIR = Path(__file__).resolve().parent / "data"
raw = pd.read_csv(DATA_DIR / "telemetry.csv")
pipelines = gpd.read_file(DATA_DIR / "pipelines.geojson")

# Surface both source counts before drawing the map.
st.set_page_config(page_title="GeoMethane-Ops", layout="wide")
st.title("GeoMethane-Ops")
st.caption("Synthetic operations demo with unvalidated telemetry.")
metric_columns = st.columns(2)
metric_columns[0].metric("Raw telemetry records", len(raw))
metric_columns[1].metric("Pipeline segments", len(pipelines))

# Center the operations view over the synthetic survey area.
operations_map = folium.Map(
    location=[37.802, -122.240],
    zoom_start=14,
    tiles="CartoDB Positron",
)

# Draw the fictional assets as one pipeline layer.
folium.GeoJson(
    pipelines,
    name="Synthetic utility pipelines",
    style_function=lambda feature: {"color": "#475569", "weight": 5},
    tooltip=folium.GeoJsonTooltip(
        fields=["pipeline_id", "pressure_class"],
        aliases=["Pipeline", "Pressure class"],
    ),
).add_to(operations_map)

# Plot every source row so the raw-data problem stays visible.
for row in raw.itertuples():
    methane_text = (
        "missing"
        if pd.isna(row.methane_ppm)
        else f"{row.methane_ppm:.2f} ppm"
    )
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=5,
        color="#2563eb",
        fill=True,
        fill_color="#2563eb",
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.timestamp} | {methane_text}"
        ),
    ).add_to(operations_map)

# Render the Folium map inside the Streamlit page.
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

Your first operations dashboard is live, and its weakness is now visible. Next, you will quarantine flawed records and connect every accepted reading to its nearest pipeline using meter-based distances.

Validate and Snap Readings

Your raw Streamlit map proved that the telemetry and fictional pipelines can load together. However, its 23 readings currently include malformed records that look as trustworthy as valid observations.

Operations also need each accepted reading tied to a nearby pipeline. In this step, you will quarantine three flawed records before using GeoPandas to calculate meter-based pipeline assignments for the remaining 20.

In this step, get ready to:

• Validate every telemetry record against a clear data contract.
• Snap each accepted reading to its nearest pipeline in projected coordinates.
• Display the accepted records, rejected records, and snap connectors on the dashboard.

Validate the telemetry records

A data contract defines the fields and value rules that every usable record must satisfy. Explicit rejection reasons preserve flawed inputs for review while keeping them out of later spatial analysis.

• Select the project folder in the Visual Studio Code Explorer sidebar.
• Click the New File button in the Explorer toolbar.
• Name the new file geomethane_ops.py.
• Add the module imports, project constants, and data loader by pasting this code:

from pathlib import Path

import geopandas as gpd
import pandas as pd
from shapely import Point, shortest_line

DATA_CRS = "EPSG:4326"
DEPOT_LON_LAT = (-122.2600, 37.7900)
DEFAULT_DATA_DIR = Path(__file__).resolve().parent / "data"
REQUIRED_COLUMNS = {
    "reading_id",
    "timestamp",
    "vehicle_id",
    "latitude",
    "longitude",
    "methane_ppm",
    "speed_kph",
}


def load_data(data_dir: str | Path = DEFAULT_DATA_DIR):
    data_dir = Path(data_dir)
    telemetry = pd.read_csv(data_dir / "telemetry.csv")
    pipelines = gpd.read_file(data_dir / "pipelines.geojson")
    return telemetry, pipelines

What does this code set up?

• DATA_CRS identifies the source coordinates as WGS 84 longitude and latitude values.
• REQUIRED_COLUMNS records the seven fields that every telemetry file must provide.
• load_data reads the telemetry table and the fictional pipeline features from the existing data folder.

• Continue in geomethane_ops.py by starting the validation function with this code:

def validate_telemetry(frame: pd.DataFrame):
    missing_columns = REQUIRED_COLUMNS.difference(frame.columns)
    if missing_columns:
        raise ValueError(f"Missing required columns: {sorted(missing_columns)}")

    work = frame.copy()
    work["timestamp"] = pd.to_datetime(work["timestamp"], errors="coerce", utc=True)
    for column in ["latitude", "longitude", "methane_ppm", "speed_kph"]:
        work[column] = pd.to_numeric(work[column], errors="coerce")

    seen_ids = set()
    rejection_reasons = []

    for _, row in work.iterrows():
        issues = []
        reading_id = "" if pd.isna(row["reading_id"]) else str(row["reading_id"]).strip()
        vehicle_id = "" if pd.isna(row["vehicle_id"]) else str(row["vehicle_id"]).strip()

        if not reading_id:
            issues.append("missing_reading_id")
        elif reading_id in seen_ids:
            issues.append("duplicate_reading_id")
        else:
            seen_ids.add(reading_id)

        if not vehicle_id:
            issues.append("missing_vehicle_id")
        if pd.isna(row["timestamp"]):
            issues.append("invalid_timestamp")

How does validation begin?

• The required-column check stops processing when the CSV structure is incomplete.
• Invalid timestamps become missing UTC timestamp values that the row checks can identify.
• Numeric coercion turns malformed measurements into missing values instead of allowing unpredictable comparisons.
• The seen_ids set detects the second occurrence of a reading identifier while preserving the first.

• Finish validate_telemetry by adding this code directly below the timestamp check:

        if (
            pd.isna(row["latitude"])
            or pd.isna(row["longitude"])
            or not -90 <= row["latitude"] <= 90
            or not -180 <= row["longitude"] <= 180
        ):
            issues.append("invalid_coordinates")
        if pd.isna(row["methane_ppm"]) or row["methane_ppm"] <= 0:
            issues.append("invalid_methane")
        if pd.isna(row["speed_kph"]) or row["speed_kph"] < 0:
            issues.append("invalid_speed")

        rejection_reasons.append("; ".join(issues))

    work["rejection_reason"] = rejection_reasons
    clean = work.loc[work["rejection_reason"] == ""].drop(columns="rejection_reason")
    rejected = work.loc[work["rejection_reason"] != ""]
    return clean.reset_index(drop=True), rejected.reset_index(drop=True)

How are records separated?

• Coordinate checks enforce valid latitude and longitude ranges.
• Methane and speed checks reject missing values plus measurements outside the accepted demo rules.
• Each rejected row keeps a rejection_reason value that explains why it was quarantined.
• The function returns separate accepted and rejected tables with clean row indexes.

• Save geomethane_ops.py.
• Check the Visual Studio Code Problems panel. You should see no Python syntax problems for geomethane_ops.py.

Seeing a Python syntax problem?

• Check that the coordinate condition remains inside the row loop with eight leading spaces.
• Check that the clean and rejected assignments align with work at four leading spaces.
• Compare every closing parenthesis and quotation mark with the snippets above.

Ask for help with my telemetry validation syntax error.

Snap accepted readings to pipelines

Longitude and latitude describe positions on a globe, so differences between their degree values are not meter distances. A projected coordinate reference system gives GeoPandas flat coordinates that support local distance calculations.

The nearest join uses a spatial index to avoid comparing every point with every line. Shapely then identifies the exact location on the matched pipeline where the connector should end.

• Add the point construction, CRS transformation, and nearest spatial join below validate_telemetry by pasting this code:

def snap_readings(clean: pd.DataFrame, pipelines: gpd.GeoDataFrame):
    points = gpd.GeoDataFrame(
        clean.copy(),
        geometry=gpd.points_from_xy(
            clean["longitude"], clean["latitude"], crs=DATA_CRS
        ),
        crs=DATA_CRS,
    )
    pipelines_wgs = pipelines.to_crs(DATA_CRS).reset_index(drop=True)
    metric_crs = pipelines_wgs.estimate_utm_crs()
    points_metric = points.to_crs(metric_crs)
    pipelines_metric = pipelines_wgs.to_crs(metric_crs)

    joined = gpd.sjoin_nearest(
        points_metric,
        pipelines_metric[["pipeline_id", "geometry"]],
        how="left",
        distance_col="snap_distance_m",
    )
    joined = joined.loc[~joined.index.duplicated(keep="first")].copy()

How does the nearest join work?

• points_from_xy converts each accepted longitude and latitude pair into point geometry.
• estimate_utm_crs selects a local projected CRS from the pipeline bounds.
• sjoin_nearest assigns the closest pipeline_id and records its projected distance in snap_distance_m.
• The duplicate-index filter keeps one match if two pipelines are exactly the same distance from a reading.

• Complete snap_readings by adding the exact snap-point calculation below the nearest join:

    snapped_points = []
    for _, row in joined.iterrows():
        pipeline_geometry = pipelines_metric.loc[int(row["index_right"]), "geometry"]
        connector = shortest_line(row.geometry, pipeline_geometry)
        snapped_points.append(Point(*connector.coords[-1]))

    snapped_wgs = gpd.GeoDataFrame(
        geometry=snapped_points, crs=metric_crs
    ).to_crs(DATA_CRS)
    joined["snapped_longitude"] = snapped_wgs.geometry.x.to_numpy()
    joined["snapped_latitude"] = snapped_wgs.geometry.y.to_numpy()
    joined["snap_distance_m"] = joined["snap_distance_m"].round(1)
    return joined.reset_index(drop=True), pipelines_wgs, metric_crs

How are connector endpoints calculated?

• shortest_line creates a segment from each reading to the nearest point on its matched pipeline.
• The last coordinate of that segment becomes the snapped pipeline location.
• The snapped points return to WGS 84 so Folium can place them on the browser map.
• The rounded snap_distance_m value remains in meters because it was calculated in the projected CRS.

• Save geomethane_ops.py.
• Compare the completed module with the reference below.

✔️ My module matches

Your module now contains the complete validation and snapping pipeline for this step.

ⓧ I'd like to double check the full code

from pathlib import Path

import geopandas as gpd
import pandas as pd
from shapely import Point, shortest_line

DATA_CRS = "EPSG:4326"
DEPOT_LON_LAT = (-122.2600, 37.7900)
DEFAULT_DATA_DIR = Path(__file__).resolve().parent / "data"
REQUIRED_COLUMNS = {
    "reading_id",
    "timestamp",
    "vehicle_id",
    "latitude",
    "longitude",
    "methane_ppm",
    "speed_kph",
}


def load_data(data_dir: str | Path = DEFAULT_DATA_DIR):
    data_dir = Path(data_dir)
    telemetry = pd.read_csv(data_dir / "telemetry.csv")
    pipelines = gpd.read_file(data_dir / "pipelines.geojson")
    return telemetry, pipelines


def validate_telemetry(frame: pd.DataFrame):
    missing_columns = REQUIRED_COLUMNS.difference(frame.columns)
    if missing_columns:
        raise ValueError(f"Missing required columns: {sorted(missing_columns)}")

    work = frame.copy()
    work["timestamp"] = pd.to_datetime(work["timestamp"], errors="coerce", utc=True)
    for column in ["latitude", "longitude", "methane_ppm", "speed_kph"]:
        work[column] = pd.to_numeric(work[column], errors="coerce")

    seen_ids = set()
    rejection_reasons = []

    for _, row in work.iterrows():
        issues = []
        reading_id = "" if pd.isna(row["reading_id"]) else str(row["reading_id"]).strip()
        vehicle_id = "" if pd.isna(row["vehicle_id"]) else str(row["vehicle_id"]).strip()

        if not reading_id:
            issues.append("missing_reading_id")
        elif reading_id in seen_ids:
            issues.append("duplicate_reading_id")
        else:
            seen_ids.add(reading_id)

        if not vehicle_id:
            issues.append("missing_vehicle_id")
        if pd.isna(row["timestamp"]):
            issues.append("invalid_timestamp")
        if (
            pd.isna(row["latitude"])
            or pd.isna(row["longitude"])
            or not -90 <= row["latitude"] <= 90
            or not -180 <= row["longitude"] <= 180
        ):
            issues.append("invalid_coordinates")
        if pd.isna(row["methane_ppm"]) or row["methane_ppm"] <= 0:
            issues.append("invalid_methane")
        if pd.isna(row["speed_kph"]) or row["speed_kph"] < 0:
            issues.append("invalid_speed")

        rejection_reasons.append("; ".join(issues))

    work["rejection_reason"] = rejection_reasons
    clean = work.loc[work["rejection_reason"] == ""].drop(columns="rejection_reason")
    rejected = work.loc[work["rejection_reason"] != ""]
    return clean.reset_index(drop=True), rejected.reset_index(drop=True)


def snap_readings(clean: pd.DataFrame, pipelines: gpd.GeoDataFrame):
    points = gpd.GeoDataFrame(
        clean.copy(),
        geometry=gpd.points_from_xy(
            clean["longitude"], clean["latitude"], crs=DATA_CRS
        ),
        crs=DATA_CRS,
    )
    pipelines_wgs = pipelines.to_crs(DATA_CRS).reset_index(drop=True)
    metric_crs = pipelines_wgs.estimate_utm_crs()
    points_metric = points.to_crs(metric_crs)
    pipelines_metric = pipelines_wgs.to_crs(metric_crs)

    joined = gpd.sjoin_nearest(
        points_metric,
        pipelines_metric[["pipeline_id", "geometry"]],
        how="left",
        distance_col="snap_distance_m",
    )
    joined = joined.loc[~joined.index.duplicated(keep="first")].copy()

    snapped_points = []
    for _, row in joined.iterrows():
        pipeline_geometry = pipelines_metric.loc[int(row["index_right"]), "geometry"]
        connector = shortest_line(row.geometry, pipeline_geometry)
        snapped_points.append(Point(*connector.coords[-1]))

    snapped_wgs = gpd.GeoDataFrame(
        geometry=snapped_points, crs=metric_crs
    ).to_crs(DATA_CRS)
    joined["snapped_longitude"] = snapped_wgs.geometry.x.to_numpy()
    joined["snapped_latitude"] = snapped_wgs.geometry.y.to_numpy()
    joined["snap_distance_m"] = joined["snap_distance_m"].round(1)
    return joined.reset_index(drop=True), pipelines_wgs, metric_crs

What should the complete module contain?

The file should end with snap_readings returning the snapped records, WGS 84 pipelines, and the projected CRS. Keep the names exact because the dashboard uses these returned values.

Having trouble with the spatial code?

• Check that the source fields are named longitude and latitude in the point construction.
• Check that both the points and pipelines use metric_crs before the nearest join.
• Check that shortest_line receives the reading geometry before the matched pipeline geometry.

Ask for help with my nearest-pipeline snapping code.

Connect validation and snapping to the dashboard

The analytics module now produces the two states the operator needs to compare. Folium can draw a connector from every accepted reading to the exact point selected on its nearest pipeline.

• Switch back to streamlit_app.py in Visual Studio Code.
• Select the existing contents of streamlit_app.py.
• Replace the selected contents by starting with this data-loading and metrics code:

import folium
import streamlit as st
from streamlit_folium import st_folium

from geomethane_ops import load_data, snap_readings, validate_telemetry

st.set_page_config(page_title="GeoMethane-Ops", layout="wide")

raw, pipelines = load_data()
clean, rejected = validate_telemetry(raw)
accepted, pipelines, _ = snap_readings(clean, pipelines)

st.title("GeoMethane-Ops")
st.caption(
    "Synthetic operations demo. Pipeline assignments use projected meter distances."
)

metric_columns = st.columns(2)
metric_columns[0].metric("Accepted readings", len(accepted))
metric_columns[1].metric("Rejected readings", len(rejected))

What does the dashboard load?

• load_data retrieves the same synthetic CSV and pipeline network shown in the raw dashboard.
• validate_telemetry separates trusted inputs from quarantined records.
• snap_readings enriches every accepted record with its pipeline assignment and snap coordinates.
• The two metrics expose the validation split before the operator inspects either table.

• Add the map and fictional pipeline layer directly below the metrics by pasting this code:

operations_map = folium.Map(
    location=[37.802, -122.240],
    zoom_start=14,
    tiles="CartoDB Positron",
)

folium.GeoJson(
    pipelines,
    name="Synthetic utility pipelines",
    style_function=lambda feature: {"color": "#475569", "weight": 5},
    tooltip=folium.GeoJsonTooltip(
        fields=["pipeline_id", "pressure_class"],
        aliases=["Pipeline", "Pressure class"],
    ),
).add_to(operations_map)

What does the map layer show?

The map uses the same center and fictional utility lines as the raw dashboard. Pipeline tooltips preserve each asset identifier and pressure class for operational context.

• Add the accepted reading markers and connector lines below the pipeline layer by pasting this code:

for row in accepted.itertuples():
    folium.PolyLine(
        [
            [row.latitude, row.longitude],
            [row.snapped_latitude, row.snapped_longitude],
        ],
        color="#94a3b8",
        weight=1,
        dash_array="4",
    ).add_to(operations_map)
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=4,
        color="#2563eb",
        fill=True,
        fill_color="#2563eb",
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {row.snap_distance_m:.1f} m"
        ),
    ).add_to(operations_map)

What do the connectors prove?

• Each dashed line starts at an accepted telemetry coordinate.
• Each line ends at the exact snapped coordinate on the matched pipeline.
• Each marker tooltip reports the reading, methane value, assigned pipeline, and meter-based distance.

• Finish the dashboard by adding the map renderer and both telemetry tables:

folium.LayerControl().add_to(operations_map)
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

st.caption("Accepted readings and nearest synthetic pipelines")
st.dataframe(
    accepted[
        [
            "reading_id",
            "vehicle_id",
            "methane_ppm",
            "pipeline_id",
            "snap_distance_m",
        ]
    ],
    hide_index=True,
)

st.caption("Rejected records")
st.dataframe(
    rejected[["reading_id", "rejection_reason"]],
    hide_index=True,
)

What do the tables reveal?

• The accepted table exposes the nearest pipeline and rounded snap distance for every usable reading.
• The rejected table keeps each quarantined reading visible with its deterministic rejection reason.
• The map and tables now use the same validated dataset.

• Save streamlit_app.py.
• Compare the completed dashboard with the reference below.

✔️ My dashboard matches

Your dashboard is ready to display validated readings, rejected records, nearest-pipeline distances, and snap connectors.

ⓧ I'd like to double check the full code

import folium
import streamlit as st
from streamlit_folium import st_folium

from geomethane_ops import load_data, snap_readings, validate_telemetry

st.set_page_config(page_title="GeoMethane-Ops", layout="wide")

raw, pipelines = load_data()
clean, rejected = validate_telemetry(raw)
accepted, pipelines, _ = snap_readings(clean, pipelines)

st.title("GeoMethane-Ops")
st.caption(
    "Synthetic operations demo. Pipeline assignments use projected meter distances."
)

metric_columns = st.columns(2)
metric_columns[0].metric("Accepted readings", len(accepted))
metric_columns[1].metric("Rejected readings", len(rejected))

operations_map = folium.Map(
    location=[37.802, -122.240],
    zoom_start=14,
    tiles="CartoDB Positron",
)

folium.GeoJson(
    pipelines,
    name="Synthetic utility pipelines",
    style_function=lambda feature: {"color": "#475569", "weight": 5},
    tooltip=folium.GeoJsonTooltip(
        fields=["pipeline_id", "pressure_class"],
        aliases=["Pipeline", "Pressure class"],
    ),
).add_to(operations_map)

for row in accepted.itertuples():
    folium.PolyLine(
        [
            [row.latitude, row.longitude],
            [row.snapped_latitude, row.snapped_longitude],
        ],
        color="#94a3b8",
        weight=1,
        dash_array="4",
    ).add_to(operations_map)
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=4,
        color="#2563eb",
        fill=True,
        fill_color="#2563eb",
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {row.snap_distance_m:.1f} m"
        ),
    ).add_to(operations_map)

folium.LayerControl().add_to(operations_map)
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

st.caption("Accepted readings and nearest synthetic pipelines")
st.dataframe(
    accepted[
        [
            "reading_id",
            "vehicle_id",
            "methane_ppm",
            "pipeline_id",
            "snap_distance_m",
        ]
    ],
    hide_index=True,
)

st.caption("Rejected records")
st.dataframe(
    rejected[["reading_id", "rejection_reason"]],
    hide_index=True,
)

What should the complete dashboard contain?

The file should validate and snap the loaded records before creating any metrics, map layers, or tables. This keeps every dashboard view aligned with the same accepted dataset.

Before you launch the dashboard, predict which three source rows will be quarantined. What rejection reason do you expect beside each one?

• Launch the updated dashboard from the active PowerShell environment by running:

streamlit run streamlit_app.py

What does this command do?

Streamlit starts the local dashboard server and loads streamlit_app.py in your browser. Keep PowerShell running while you inspect the map.

You should see 20 accepted readings and 3 rejected readings. The rejected table should show duplicate_reading_id, invalid_timestamp, and invalid_methane.

Every accepted row should contain a pipeline_id and snap_distance_m value. The map should show one dashed connector from each accepted reading to its snapped pipeline location.

Dashboard not showing the 20 and 3 split?

• Check that the second R020 row receives duplicate_reading_id while the first R020 remains accepted.
• Check that timestamp parsing uses UTC mode plus coercion for the not-a-time value.
• Check that the missing R022 methane value becomes invalid_methane after numeric coercion.

Ask for help with my unexpected telemetry validation split.

Map missing connectors or basemap tiles?

• Check your internet connection if the pipeline overlays appear over a blank basemap.
• Check that snapped_latitude and snapped_longitude are transformed back to DATA_CRS before Folium uses them.
• Check that the connector uses the source coordinates first and the snapped coordinates second.

Ask for help with my missing Folium snap connectors.

That is a strong operations checkpoint. Your dashboard now distinguishes trusted telemetry from quarantined records while linking every accepted reading to a specific pipeline in meters.

Next, you will turn the elevated readings into spatial plume candidates while preserving isolated observations as noise.

Detect Plume Candidates

Your dashboard now separates flawed telemetry from 20 accepted readings. It also links every accepted point to its nearest synthetic pipeline using meter-based distances.

Those accepted readings still appear as isolated observations. An operator needs spatial groups that turn elevated readings into inspection candidates while preserving sparse readings as noise.

In this step, get ready to:

• Identify elevated readings using a synthetic methane threshold.
• Group spatially dense readings into plume candidates.
• Display three plume candidates with summary metrics on the operations map.

Inspect the isolated readings

The current map treats every accepted reading as an individual point. Before changing that behavior, check whether the elevated readings already form visible operational groups.

Before you launch the dashboard, do you expect it to identify distinct plume candidates from the accepted readings?

• Launch the current dashboard from the active PowerShell environment by running:

streamlit run streamlit_app.py

What should you see?

The browser shows the accepted readings with their pipeline connectors. Elevated points remain individual observations because the pipeline has not grouped them by spatial density.

Dashboard not opening?

• Confirm the PowerShell prompt still shows that the project virtual environment is active.
• Confirm PowerShell is using the folder that contains streamlit_app.py.

Troubleshoot the local launch with help me diagnose why my Streamlit dashboard is not opening from its active virtual environment.

How will the readings become candidates?

The fixture uses 2.8 ppm as a synthetic candidate threshold. This value supports the demonstration dataset. It is not a safety threshold or leak-confirmation threshold.

DBSCAN groups nearby candidate points without requiring a predefined cluster count. It labels sparse candidate points as -1 so the dashboard can preserve them as noise.

The algorithm receives points in the estimated local UTM coordinate reference system. That projection lets eps=180 represent a radius of 180 meters.

Build the plume detector

The detector first labels elevated readings. It then summarizes every non-noise cluster into one plume candidate with a centroid and methane metrics.

• Return to geomethane_ops.py in Visual Studio Code.
• Replace the import group from import geopandas as gpd through the Shapely import with this updated group:

import geopandas as gpd
import pandas as pd
from ortools.constraint_solver import pywrapcp, routing_enums_pb2
from shapely import Point, shortest_line
from sklearn.cluster import DBSCAN

What does this import add?

The DBSCAN class provides density-based clustering. The existing geospatial imports continue to supply projected point geometry for the model.

• Add the first part of detect_plumes below snap_readings by pasting this code:

def detect_plumes(
    snapped: gpd.GeoDataFrame,
    methane_threshold: float = 2.8,
    eps_m: int = 180,
    min_samples: int = 3,
):
    labeled = snapped.copy()
    labeled["plume_id"] = -1
    candidate_indices = labeled.index[labeled["methane_ppm"] >= methane_threshold]

    if len(candidate_indices) >= min_samples:
        coordinates = [
            [labeled.at[index, "geometry"].x, labeled.at[index, "geometry"].y]
            for index in candidate_indices
        ]
        raw_labels = DBSCAN(eps=eps_m, min_samples=min_samples).fit_predict(coordinates)
        plume_labels = [label + 1 if label >= 0 else -1 for label in raw_labels]
        labeled.loc[candidate_indices, "plume_id"] = plume_labels

How are readings labeled?

• The plume_id column starts at -1. This preserves ordinary readings and unclustered candidates as noise.
• The candidate filter keeps readings at or above the synthetic methane_threshold value.
• The model uses eps_m as its search radius. It requires at least min_samples nearby readings to create a cluster.
• Adding one to each non-negative model label produces plume IDs beginning at 1.

• Continue the same function directly below the label assignment by pasting this empty-result handling:

    clustered = labeled.loc[labeled["plume_id"] >= 0]
    if clustered.empty:
        empty = gpd.GeoDataFrame(
            columns=[
                "plume_id",
                "reading_count",
                "mean_methane_ppm",
                "peak_methane_ppm",
                "pipeline_id",
                "geometry",
            ],
            geometry="geometry",
            crs=labeled.crs,
        )
        return labeled, empty

Why handle an empty result?

Some threshold and radius combinations may produce no clusters. Returning an empty GeoDataFrame with the expected columns keeps the dashboard data contract stable.

• Complete detect_plumes directly below the empty-result branch by pasting this aggregation code:

    plume_records = []
    for plume_id, group in clustered.groupby("plume_id"):
        plume_records.append(
            {
                "plume_id": int(plume_id),
                "reading_count": len(group),
                "mean_methane_ppm": round(group["methane_ppm"].mean(), 2),
                "peak_methane_ppm": round(group["methane_ppm"].max(), 2),
                "pipeline_id": group["pipeline_id"].mode().iat[0],
                "geometry": Point(group.geometry.x.mean(), group.geometry.y.mean()),
            }
        )

    plumes = gpd.GeoDataFrame(plume_records, geometry="geometry", crs=labeled.crs)
    return labeled, plumes

What does each summary contain?

• The reading_count records how many elevated readings support the candidate.
• The mean and peak methane fields describe the candidate readings without declaring a confirmed leak.
• The modal pipeline_id links the candidate to the pipeline associated with most readings in its cluster.
• The mean projected coordinates create the centroid used by the map.

• Save geomethane_ops.py.

✔️ Awesome, I've got everything!

Your detector now labels candidate readings and returns plume summaries. Keep geomethane_ops.py saved before updating the dashboard.

ⓧ I'd like to double check the full code

from pathlib import Path

import geopandas as gpd
import pandas as pd
from shapely import Point, shortest_line
from sklearn.cluster import DBSCAN

DATA_CRS = "EPSG:4326"
DEPOT_LON_LAT = (-122.2600, 37.7900)
DEFAULT_DATA_DIR = Path(__file__).resolve().parent / "data"
REQUIRED_COLUMNS = {
    "reading_id",
    "timestamp",
    "vehicle_id",
    "latitude",
    "longitude",
    "methane_ppm",
    "speed_kph",
}


def load_data(data_dir: str | Path = DEFAULT_DATA_DIR):
    data_dir = Path(data_dir)
    telemetry = pd.read_csv(data_dir / "telemetry.csv")
    pipelines = gpd.read_file(data_dir / "pipelines.geojson")
    return telemetry, pipelines


def validate_telemetry(frame: pd.DataFrame):
    missing_columns = REQUIRED_COLUMNS.difference(frame.columns)
    if missing_columns:
        raise ValueError(f"Missing required columns: {sorted(missing_columns)}")

    work = frame.copy()
    work["timestamp"] = pd.to_datetime(work["timestamp"], errors="coerce", utc=True)
    for column in ["latitude", "longitude", "methane_ppm", "speed_kph"]:
        work[column] = pd.to_numeric(work[column], errors="coerce")

    seen_ids = set()
    rejection_reasons = []

    for _, row in work.iterrows():
        issues = []
        reading_id = "" if pd.isna(row["reading_id"]) else str(row["reading_id"]).strip()
        vehicle_id = "" if pd.isna(row["vehicle_id"]) else str(row["vehicle_id"]).strip()

        if not reading_id:
            issues.append("missing_reading_id")
        elif reading_id in seen_ids:
            issues.append("duplicate_reading_id")
        else:
            seen_ids.add(reading_id)

        if not vehicle_id:
            issues.append("missing_vehicle_id")
        if pd.isna(row["timestamp"]):
            issues.append("invalid_timestamp")
        if (
            pd.isna(row["latitude"])
            or pd.isna(row["longitude"])
            or not -90 <= row["latitude"] <= 90
            or not -180 <= row["longitude"] <= 180
        ):
            issues.append("invalid_coordinates")
        if pd.isna(row["methane_ppm"]) or row["methane_ppm"] <= 0:
            issues.append("invalid_methane")
        if pd.isna(row["speed_kph"]) or row["speed_kph"] < 0:
            issues.append("invalid_speed")

        rejection_reasons.append("; ".join(issues))

    work["rejection_reason"] = rejection_reasons
    clean = work.loc[work["rejection_reason"] == ""].drop(columns="rejection_reason")
    rejected = work.loc[work["rejection_reason"] != ""]
    return clean.reset_index(drop=True), rejected.reset_index(drop=True)


def snap_readings(clean: pd.DataFrame, pipelines: gpd.GeoDataFrame):
    points = gpd.GeoDataFrame(
        clean.copy(),
        geometry=gpd.points_from_xy(
            clean["longitude"], clean["latitude"], crs=DATA_CRS
        ),
        crs=DATA_CRS,
    )
    pipelines_wgs = pipelines.to_crs(DATA_CRS).reset_index(drop=True)
    metric_crs = pipelines_wgs.estimate_utm_crs()
    points_metric = points.to_crs(metric_crs)
    pipelines_metric = pipelines_wgs.to_crs(metric_crs)

    joined = gpd.sjoin_nearest(
        points_metric,
        pipelines_metric[["pipeline_id", "geometry"]],
        how="left",
        distance_col="snap_distance_m",
    )
    joined = joined.loc[~joined.index.duplicated(keep="first")].copy()

    snapped_points = []
    for _, row in joined.iterrows():
        pipeline_geometry = pipelines_metric.loc[int(row["index_right"]), "geometry"]
        connector = shortest_line(row.geometry, pipeline_geometry)
        snapped_points.append(Point(*connector.coords[-1]))

    snapped_wgs = gpd.GeoDataFrame(
        geometry=snapped_points, crs=metric_crs
    ).to_crs(DATA_CRS)
    joined["snapped_longitude"] = snapped_wgs.geometry.x.to_numpy()
    joined["snapped_latitude"] = snapped_wgs.geometry.y.to_numpy()
    joined["snap_distance_m"] = joined["snap_distance_m"].round(1)
    return joined.reset_index(drop=True), pipelines_wgs, metric_crs


def detect_plumes(
    snapped: gpd.GeoDataFrame,
    methane_threshold: float = 2.8,
    eps_m: int = 180,
    min_samples: int = 3,
):
    labeled = snapped.copy()
    labeled["plume_id"] = -1
    candidate_indices = labeled.index[labeled["methane_ppm"] >= methane_threshold]

    if len(candidate_indices) >= min_samples:
        coordinates = [
            [labeled.at[index, "geometry"].x, labeled.at[index, "geometry"].y]
            for index in candidate_indices
        ]
        raw_labels = DBSCAN(eps=eps_m, min_samples=min_samples).fit_predict(coordinates)
        plume_labels = [label + 1 if label >= 0 else -1 for label in raw_labels]
        labeled.loc[candidate_indices, "plume_id"] = plume_labels

    clustered = labeled.loc[labeled["plume_id"] >= 0]
    if clustered.empty:
        empty = gpd.GeoDataFrame(
            columns=[
                "plume_id",
                "reading_count",
                "mean_methane_ppm",
                "peak_methane_ppm",
                "pipeline_id",
                "geometry",
            ],
            geometry="geometry",
            crs=labeled.crs,
        )
        return labeled, empty

    plume_records = []
    for plume_id, group in clustered.groupby("plume_id"):
        plume_records.append(
            {
                "plume_id": int(plume_id),
                "reading_count": len(group),
                "mean_methane_ppm": round(group["methane_ppm"].mean(), 2),
                "peak_methane_ppm": round(group["methane_ppm"].max(), 2),
                "pipeline_id": group["pipeline_id"].mode().iat[0],
                "geometry": Point(group.geometry.x.mean(), group.geometry.y.mean()),
            }
        )

    plumes = gpd.GeoDataFrame(plume_records, geometry="geometry", crs=labeled.crs)
    return labeled, plumes

Map and verify the plume candidates

The detector returns labeled readings in projected coordinates. The dashboard keeps those coordinates for clustering while transforming plume centroids back to WGS 84 for Folium.

• Return to streamlit_app.py.
• Extend the existing import from geomethane_ops to include DATA_CRS and detect_plumes.
• Add the detector call immediately after snap_readings returns the accepted readings:

accepted, plumes_metric = detect_plumes(accepted)
plumes = (
    plumes_metric.to_crs(DATA_CRS)
    if not plumes_metric.empty
    else plumes_metric
)

Why keep two plume frames?

The detector uses plumes_metric for meter-based analysis. The transformed plumes frame supplies longitude and latitude coordinates for the browser map.

• Replace the existing dashboard caption with this analytical-use warning:

st.caption(
    "Synthetic operations demo. Plume candidates are analytical priorities, "
    "not confirmed leaks or safety thresholds."
)

Why show this warning?

The 2.8 ppm setting exists to demonstrate the synthetic fixture. The caption keeps analytical prioritization separate from safety decisions.

• Replace the accepted-reading marker loop with this cluster-aware version:

plume_colors = {-1: "#2563eb", 1: "#dc2626", 2: "#ea580c", 3: "#a21caf"}
for row in accepted.itertuples():
    plume_id = int(row.plume_id)
    color = plume_colors.get(plume_id, "#dc2626")
    folium.PolyLine(
        [
            [row.latitude, row.longitude],
            [row.snapped_latitude, row.snapped_longitude],
        ],
        color="#94a3b8",
        weight=1,
        dash_array="4",
    ).add_to(operations_map)
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=7 if plume_id >= 0 else 4,
        color=color,
        fill=True,
        fill_color=color,
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {row.snap_distance_m:.1f} m"
        ),
    ).add_to(operations_map)

How does the map distinguish clusters?

Noise readings use blue markers with a smaller radius. Clustered readings use larger markers with a separate color for each plume_id.

The connector lines remain visible. This keeps each reading linked to its snapped pipeline while adding the new cluster layer.

• Add the plume centroid markers after the inspection depot marker:

for row in plumes.itertuples():
    folium.Marker(
        location=[row.geometry.y, row.geometry.x],
        tooltip=(
            f"Plume {row.plume_id} | "
            f"{row.reading_count} readings | "
            f"mean {row.mean_methane_ppm:.2f} ppm | "
            f"peak {row.peak_methane_ppm:.2f} ppm | "
            f"{row.pipeline_id}"
        ),
    ).add_to(operations_map)

What does each marker prove?

Each marker represents one cluster centroid. Its tooltip exposes the reading count plus mean methane plus peak methane plus nearest pipeline.

The default fixture produces plume IDs 1, 2, and 3.

• Save streamlit_app.py.

Before you check the browser, how many centroid markers do you expect the default threshold and radius to produce?

• Return to the running Streamlit dashboard in your browser.
• Refresh the dashboard to load the saved code.
• Confirm the map shows three numbered plume candidate centroids.
• Hover over each centroid to confirm its tooltip includes a reading count plus mean methane plus peak methane plus a pipeline ID.

You should see three plume candidates with IDs 1, 2, and 3. Sparse readings remain blue noise points with plume_id set to -1.

Missing plume markers?

• Check that detect_plumes receives the projected accepted frame returned by snap_readings.
• Check that the default detector values remain 2.8 for the threshold plus 180 for the radius plus 3 for the minimum sample count.
• Check that the plume centroid frame is transformed to EPSG:4326 before Folium receives it.

Investigate the detector with help me debug why GeoMethane-Ops does not show three DBSCAN plume candidates.

That is the analytical layer working: elevated readings now become three spatial inspection candidates while sparse observations remain visible as noise. Next, you will assign those candidates across two vans and protect the workflow with automated tests.

Optimize Dispatch and Test the Pipeline

Your dashboard now turns validated readings into three spatial plume candidates with DBSCAN. Each candidate identifies a place that deserves inspection.

A priority map still leaves an operations team deciding which van visits each candidate. This step uses Google OR-Tools to create a two-van plan. You will also add repeatable tests that protect every critical transformation.

In this step, get ready to:

• Optimize two depot-to-depot inspection routes across the three plume candidates.
• Display color-coded routes with exactly-once plume assignments on the completed dashboard.
• Test validation, spatial snapping, plume detection, and dispatch coverage with pytest.

Build the two-van dispatch optimizer

A vehicle routing problem assigns destinations across a fleet while accounting for travel distance. Here, the depot and plume centroids become nodes in an integer meter-distance matrix.

Why use OR-Tools for dispatch?

A greedy nearest-stop loop makes one local choice at a time. The OR-Tools routing model evaluates complete routes across both vans.

The distance dimension measures each route. Its global-span cost encourages a more balanced split between the vans.

• In geomethane_ops.py, replace the current import section with this completed import section:

from math import hypot
from pathlib import Path

import geopandas as gpd
import pandas as pd
from ortools.constraint_solver import pywrapcp, routing_enums_pb2
from shapely import Point, shortest_line
from sklearn.cluster import DBSCAN

What do these imports add?

• The hypot function calculates straight-line distances between projected coordinates.
• The pywrapcp module supplies the routing manager and model.
• The routing_enums_pb2 module supplies the first-solution strategy.

• Add the opening section of optimize_dispatch below detect_plumes:

def optimize_dispatch(
    plumes_metric: gpd.GeoDataFrame,
    metric_crs,
    num_vans: int = 2,
):
    assignment_columns = ["van_id", "stop_order", "plume_id"]
    if plumes_metric.empty:
        return pd.DataFrame(columns=assignment_columns), []

    depot_metric = gpd.GeoDataFrame(
        geometry=[Point(*DEPOT_LON_LAT)], crs=DATA_CRS
    ).to_crs(metric_crs).geometry.iloc[0]
    node_geometries = [depot_metric, *plumes_metric.geometry.tolist()]

    distance_matrix = [
        [int(round(hypot(a.x - b.x, a.y - b.y))) for b in node_geometries]
        for a in node_geometries
    ]

How does the distance matrix work?

• The empty-result branch keeps the dashboard usable when clustering produces no plume candidates.
• The depot is transformed into the same projected CRS as the plume centroids.
• Each matrix cell stores a rounded meter distance between two nodes.

• Continue optimize_dispatch directly below the distance matrix with the routing model configuration:

    manager = pywrapcp.RoutingIndexManager(len(distance_matrix), num_vans, 0)
    routing = pywrapcp.RoutingModel(manager)

    def distance_callback(from_index, to_index):
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        return distance_matrix[from_node][to_node]

    transit_callback_index = routing.RegisterTransitCallback(distance_callback)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
    routing.AddDimension(
        transit_callback_index,
        0,
        1_000_000,
        True,
        "Distance",
    )
    distance_dimension = routing.GetDimensionOrDie("Distance")
    distance_dimension.SetGlobalSpanCostCoefficient(100)

    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    solution = routing.SolveWithParameters(search_parameters)
    if solution is None:
        raise RuntimeError("OR-Tools could not find a dispatch solution")

What does the routing model optimize?

• The routing manager translates internal solver indices into depot and plume nodes.
• The transit callback returns the integer distance between each pair of nodes.
• The distance dimension records route length for each van.
• The global-span coefficient of 100 gives the longest van route extra weight.

• Continue the function by extracting each used van route from the solution:

    assignment_rows = []
    routes = []

    for vehicle_index in range(num_vans):
        if not routing.IsVehicleUsed(solution, vehicle_index):
            continue

        index = routing.Start(vehicle_index)
        node_indices = []
        route_distance = 0
        while not routing.IsEnd(index):
            node_indices.append(manager.IndexToNode(index))
            previous_index = index
            index = solution.Value(routing.NextVar(index))
            route_distance += routing.GetArcCostForVehicle(
                previous_index, index, vehicle_index
            )
        node_indices.append(manager.IndexToNode(index))

        plume_ids = []
        stop_order = 1
        for node_index in node_indices:
            if node_index == 0:
                continue
            plume_id = int(plumes_metric.iloc[node_index - 1]["plume_id"])
            plume_ids.append(plume_id)

How are solved routes decoded?

• Each used vehicle starts at the depot index.
• The solver's next-index values reveal the ordered route.
• Depot nodes are excluded from the assignment rows.
• Each remaining node maps back to its original plume_id.

• Finish optimize_dispatch with assignment rows and map-ready route coordinates:

            assignment_rows.append(
                {
                    "van_id": f"VAN-{vehicle_index + 1:02d}",
                    "stop_order": stop_order,
                    "plume_id": plume_id,
                }
            )
            stop_order += 1

        route_points = [node_geometries[node_index] for node_index in node_indices]
        route_wgs = gpd.GeoDataFrame(
            geometry=route_points, crs=metric_crs
        ).to_crs(DATA_CRS)
        coordinates = [[point.y, point.x] for point in route_wgs.geometry]
        routes.append(
            {
                "van_id": f"VAN-{vehicle_index + 1:02d}",
                "plume_ids": plume_ids,
                "distance_m": route_distance,
                "coordinates": coordinates,
            }
        )

    assignments = pd.DataFrame(assignment_rows, columns=assignment_columns)
    return assignments, routes

What does the function return?

• The assignments table records the van, stop order, and plume for every visit.
• Each route records its plume IDs and total planned distance.
• The route coordinates return to WGS 84 so Folium can draw them.

• Add run_pipeline below optimize_dispatch to connect the complete workflow:

def run_pipeline(
    data_dir: str | Path = DEFAULT_DATA_DIR,
    methane_threshold: float = 2.8,
    eps_m: int = 180,
):
    raw, pipelines = load_data(data_dir)
    clean, rejected = validate_telemetry(raw)
    snapped, pipelines_wgs, metric_crs = snap_readings(clean, pipelines)
    labeled, plumes_metric = detect_plumes(
        snapped,
        methane_threshold=methane_threshold,
        eps_m=eps_m,
    )
    assignments, routes = optimize_dispatch(plumes_metric, metric_crs)
    plumes_wgs = (
        plumes_metric.to_crs(DATA_CRS) if not plumes_metric.empty else plumes_metric
    )
    return {
        "raw": raw,
        "accepted": labeled,
        "rejected": rejected,
        "pipelines": pipelines_wgs,
        "plumes": plumes_wgs,
        "assignments": assignments,
        "routes": routes,
    }

What does the pipeline orchestrator do?

The run_pipeline function executes ingestion, validation, snapping, clustering, and dispatch in one reproducible sequence.

Its result dictionary gives the dashboard and tests one shared source of truth.

• Save geomethane_ops.py.

Seeing an indentation or import error?

• Check that every continuation block remains inside optimize_dispatch with the indentation shown.
• Confirm the OR-Tools import sits with the other imports at the top of geomethane_ops.py.
• Confirm run_pipeline starts at the left edge after optimize_dispatch ends.

Help me compare my dispatch function with the expected indentation.

✔️ Awesome, I've got everything!

Great. Your pipeline module now turns validated plume candidates into two depot-to-depot inspection routes.

ⓧ I'd like to double check the full code

from math import hypot
from pathlib import Path

import geopandas as gpd
import pandas as pd
from ortools.constraint_solver import pywrapcp, routing_enums_pb2
from shapely import Point, shortest_line
from sklearn.cluster import DBSCAN

DATA_CRS = "EPSG:4326"
DEPOT_LON_LAT = (-122.2600, 37.7900)
DEFAULT_DATA_DIR = Path(__file__).resolve().parent / "data"
REQUIRED_COLUMNS = {
    "reading_id",
    "timestamp",
    "vehicle_id",
    "latitude",
    "longitude",
    "methane_ppm",
    "speed_kph",
}


def load_data(data_dir: str | Path = DEFAULT_DATA_DIR):
    data_dir = Path(data_dir)
    telemetry = pd.read_csv(data_dir / "telemetry.csv")
    pipelines = gpd.read_file(data_dir / "pipelines.geojson")
    return telemetry, pipelines


def validate_telemetry(frame: pd.DataFrame):
    missing_columns = REQUIRED_COLUMNS.difference(frame.columns)
    if missing_columns:
        raise ValueError(f"Missing required columns: {sorted(missing_columns)}")

    work = frame.copy()
    work["timestamp"] = pd.to_datetime(work["timestamp"], errors="coerce", utc=True)
    for column in ["latitude", "longitude", "methane_ppm", "speed_kph"]:
        work[column] = pd.to_numeric(work[column], errors="coerce")

    seen_ids = set()
    rejection_reasons = []

    for _, row in work.iterrows():
        issues = []
        reading_id = "" if pd.isna(row["reading_id"]) else str(row["reading_id"]).strip()
        vehicle_id = "" if pd.isna(row["vehicle_id"]) else str(row["vehicle_id"]).strip()

        if not reading_id:
            issues.append("missing_reading_id")
        elif reading_id in seen_ids:
            issues.append("duplicate_reading_id")
        else:
            seen_ids.add(reading_id)

        if not vehicle_id:
            issues.append("missing_vehicle_id")
        if pd.isna(row["timestamp"]):
            issues.append("invalid_timestamp")
        if (
            pd.isna(row["latitude"])
            or pd.isna(row["longitude"])
            or not -90 <= row["latitude"] <= 90
            or not -180 <= row["longitude"] <= 180
        ):
            issues.append("invalid_coordinates")
        if pd.isna(row["methane_ppm"]) or row["methane_ppm"] <= 0:
            issues.append("invalid_methane")
        if pd.isna(row["speed_kph"]) or row["speed_kph"] < 0:
            issues.append("invalid_speed")

        rejection_reasons.append("; ".join(issues))

    work["rejection_reason"] = rejection_reasons
    clean = work.loc[work["rejection_reason"] == ""].drop(columns="rejection_reason")
    rejected = work.loc[work["rejection_reason"] != ""]
    return clean.reset_index(drop=True), rejected.reset_index(drop=True)


def snap_readings(clean: pd.DataFrame, pipelines: gpd.GeoDataFrame):
    points = gpd.GeoDataFrame(
        clean.copy(),
        geometry=gpd.points_from_xy(
            clean["longitude"], clean["latitude"], crs=DATA_CRS
        ),
        crs=DATA_CRS,
    )
    pipelines_wgs = pipelines.to_crs(DATA_CRS).reset_index(drop=True)
    metric_crs = pipelines_wgs.estimate_utm_crs()
    points_metric = points.to_crs(metric_crs)
    pipelines_metric = pipelines_wgs.to_crs(metric_crs)

    joined = gpd.sjoin_nearest(
        points_metric,
        pipelines_metric[["pipeline_id", "geometry"]],
        how="left",
        distance_col="snap_distance_m",
    )
    joined = joined.loc[~joined.index.duplicated(keep="first")].copy()

    snapped_points = []
    for _, row in joined.iterrows():
        pipeline_geometry = pipelines_metric.loc[int(row["index_right"]), "geometry"]
        connector = shortest_line(row.geometry, pipeline_geometry)
        snapped_points.append(Point(*connector.coords[-1]))

    snapped_wgs = gpd.GeoDataFrame(
        geometry=snapped_points, crs=metric_crs
    ).to_crs(DATA_CRS)
    joined["snapped_longitude"] = snapped_wgs.geometry.x.to_numpy()
    joined["snapped_latitude"] = snapped_wgs.geometry.y.to_numpy()
    joined["snap_distance_m"] = joined["snap_distance_m"].round(1)
    return joined.reset_index(drop=True), pipelines_wgs, metric_crs


def detect_plumes(
    snapped: gpd.GeoDataFrame,
    methane_threshold: float = 2.8,
    eps_m: int = 180,
    min_samples: int = 3,
):
    labeled = snapped.copy()
    labeled["plume_id"] = -1
    candidate_indices = labeled.index[labeled["methane_ppm"] >= methane_threshold]

    if len(candidate_indices) >= min_samples:
        coordinates = [
            [labeled.at[index, "geometry"].x, labeled.at[index, "geometry"].y]
            for index in candidate_indices
        ]
        raw_labels = DBSCAN(eps=eps_m, min_samples=min_samples).fit_predict(coordinates)
        plume_labels = [label + 1 if label >= 0 else -1 for label in raw_labels]
        labeled.loc[candidate_indices, "plume_id"] = plume_labels

    clustered = labeled.loc[labeled["plume_id"] >= 0]
    if clustered.empty:
        empty = gpd.GeoDataFrame(
            columns=[
                "plume_id",
                "reading_count",
                "mean_methane_ppm",
                "peak_methane_ppm",
                "pipeline_id",
                "geometry",
            ],
            geometry="geometry",
            crs=labeled.crs,
        )
        return labeled, empty

    plume_records = []
    for plume_id, group in clustered.groupby("plume_id"):
        plume_records.append(
            {
                "plume_id": int(plume_id),
                "reading_count": len(group),
                "mean_methane_ppm": round(group["methane_ppm"].mean(), 2),
                "peak_methane_ppm": round(group["methane_ppm"].max(), 2),
                "pipeline_id": group["pipeline_id"].mode().iat[0],
                "geometry": Point(group.geometry.x.mean(), group.geometry.y.mean()),
            }
        )

    plumes = gpd.GeoDataFrame(plume_records, geometry="geometry", crs=labeled.crs)
    return labeled, plumes


def optimize_dispatch(
    plumes_metric: gpd.GeoDataFrame,
    metric_crs,
    num_vans: int = 2,
):
    assignment_columns = ["van_id", "stop_order", "plume_id"]
    if plumes_metric.empty:
        return pd.DataFrame(columns=assignment_columns), []

    depot_metric = gpd.GeoDataFrame(
        geometry=[Point(*DEPOT_LON_LAT)], crs=DATA_CRS
    ).to_crs(metric_crs).geometry.iloc[0]
    node_geometries = [depot_metric, *plumes_metric.geometry.tolist()]

    distance_matrix = [
        [int(round(hypot(a.x - b.x, a.y - b.y))) for b in node_geometries]
        for a in node_geometries
    ]

    manager = pywrapcp.RoutingIndexManager(len(distance_matrix), num_vans, 0)
    routing = pywrapcp.RoutingModel(manager)

    def distance_callback(from_index, to_index):
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        return distance_matrix[from_node][to_node]

    transit_callback_index = routing.RegisterTransitCallback(distance_callback)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
    routing.AddDimension(
        transit_callback_index,
        0,
        1_000_000,
        True,
        "Distance",
    )
    distance_dimension = routing.GetDimensionOrDie("Distance")
    distance_dimension.SetGlobalSpanCostCoefficient(100)

    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    solution = routing.SolveWithParameters(search_parameters)
    if solution is None:
        raise RuntimeError("OR-Tools could not find a dispatch solution")

    assignment_rows = []
    routes = []

    for vehicle_index in range(num_vans):
        if not routing.IsVehicleUsed(solution, vehicle_index):
            continue

        index = routing.Start(vehicle_index)
        node_indices = []
        route_distance = 0
        while not routing.IsEnd(index):
            node_indices.append(manager.IndexToNode(index))
            previous_index = index
            index = solution.Value(routing.NextVar(index))
            route_distance += routing.GetArcCostForVehicle(
                previous_index, index, vehicle_index
            )
        node_indices.append(manager.IndexToNode(index))

        plume_ids = []
        stop_order = 1
        for node_index in node_indices:
            if node_index == 0:
                continue
            plume_id = int(plumes_metric.iloc[node_index - 1]["plume_id"])
            plume_ids.append(plume_id)
            assignment_rows.append(
                {
                    "van_id": f"VAN-{vehicle_index + 1:02d}",
                    "stop_order": stop_order,
                    "plume_id": plume_id,
                }
            )
            stop_order += 1

        route_points = [node_geometries[node_index] for node_index in node_indices]
        route_wgs = gpd.GeoDataFrame(
            geometry=route_points, crs=metric_crs
        ).to_crs(DATA_CRS)
        coordinates = [[point.y, point.x] for point in route_wgs.geometry]
        routes.append(
            {
                "van_id": f"VAN-{vehicle_index + 1:02d}",
                "plume_ids": plume_ids,
                "distance_m": route_distance,
                "coordinates": coordinates,
            }
        )

    assignments = pd.DataFrame(assignment_rows, columns=assignment_columns)
    return assignments, routes


def run_pipeline(
    data_dir: str | Path = DEFAULT_DATA_DIR,
    methane_threshold: float = 2.8,
    eps_m: int = 180,
):
    raw, pipelines = load_data(data_dir)
    clean, rejected = validate_telemetry(raw)
    snapped, pipelines_wgs, metric_crs = snap_readings(clean, pipelines)
    labeled, plumes_metric = detect_plumes(
        snapped,
        methane_threshold=methane_threshold,
        eps_m=eps_m,
    )
    assignments, routes = optimize_dispatch(plumes_metric, metric_crs)
    plumes_wgs = (
        plumes_metric.to_crs(DATA_CRS) if not plumes_metric.empty else plumes_metric
    )
    return {
        "raw": raw,
        "accepted": labeled,
        "rejected": rejected,
        "pipelines": pipelines_wgs,
        "plumes": plumes_wgs,
        "assignments": assignments,
        "routes": routes,
    }

How should I use this reference?

Compare the function order, indentation, identifiers, and values with your saved file. The completed module should match this reference exactly.

Add routes and assignments to the dashboard

The optimizer returns a visit order and projected route distance for each van. The map uses straight segments between the depot and plume centroids as planning lines.

What do the route lines represent?

These lines visualize the dispatch sequence calculated from projected centroid distances. They do not represent turn-by-turn road directions.

• In streamlit_app.py, replace the current imports, pipeline loading, heading, caption, and metrics with this completed section:

import folium
import streamlit as st
from streamlit_folium import st_folium

from geomethane_ops import DEPOT_LON_LAT, run_pipeline

st.set_page_config(page_title="GeoMethane-Ops", layout="wide")

results = run_pipeline()
accepted = results["accepted"]
rejected = results["rejected"]
pipelines = results["pipelines"]
plumes = results["plumes"]
assignments = results["assignments"]
routes = results["routes"]

st.title("GeoMethane-Ops")
st.caption(
    "Synthetic operations demo. Plume candidates are analytical priorities, "
    "not confirmed leaks or safety thresholds."
)

metric_columns = st.columns(4)
metric_columns[0].metric("Accepted readings", len(accepted))
metric_columns[1].metric("Rejected readings", len(rejected))
metric_columns[2].metric("Plume candidates", len(plumes))
metric_columns[3].metric("Vans dispatched", assignments["van_id"].nunique())

What changes in the dashboard data flow?

• The dashboard now calls run_pipeline once for every result it needs.
• The assignments and routes become available beside the accepted, rejected, pipeline, and plume results.
• The fourth metric reports the number of vans used by the solution.

• Save streamlit_app.py.

Before you run the dashboard, how many vans do you expect the fourth metric to report?

• Launch the updated dashboard from the active PowerShell environment by running:

streamlit run streamlit_app.py

What does this command do?

Streamlit starts the local dashboard server and opens the application in your browser. The script runs the full pipeline before rendering the page.

You should see 20 accepted readings, 3 rejected readings, 3 plume candidates, and 2 dispatched vans. The new van metric proves that the routing model produced assignments.

• Return to PowerShell.
• Press Ctrl+C to stop the Streamlit server before continuing.

Dashboard fails before the map loads?

• Check that run_pipeline is aligned at the left edge in geomethane_ops.py.
• Check that the result keys in streamlit_app.py match the keys returned by run_pipeline.
• Confirm the .venv environment remains active in PowerShell.

Help me diagnose why run_pipeline does not load in my Streamlit dashboard.

• Replace the current plume-marker section in streamlit_app.py with this depot and assignment-aware section:

folium.Marker(
    location=[DEPOT_LON_LAT[1], DEPOT_LON_LAT[0]],
    tooltip="Inspection depot",
).add_to(operations_map)

assignment_lookup = (
    assignments.set_index("plume_id")["van_id"].to_dict()
    if not assignments.empty
    else {}
)
for row in plumes.itertuples():
    van_id = assignment_lookup.get(int(row.plume_id), "Unassigned")
    folium.Marker(
        location=[row.geometry.y, row.geometry.x],
        tooltip=(
            f"Plume {row.plume_id} | peak {row.peak_methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {van_id}"
        ),
    ).add_to(operations_map)

How do plume markers gain dispatch context?

• The depot marker identifies the common start and end location.
• The assignment lookup maps each plume_id to its assigned van.
• Each plume tooltip now includes the van responsible for that visit.

• Replace the current map-rendering section with this route and rendering section:

route_colors = ["#16a34a", "#7c3aed"]
for route_index, route in enumerate(routes):
    folium.PolyLine(
        route["coordinates"],
        color=route_colors[route_index % len(route_colors)],
        weight=5,
        tooltip=f"{route['van_id']} | {route['distance_m']} m",
    ).add_to(operations_map)

folium.LayerControl().add_to(operations_map)
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

How are the routes drawn?

• Each solved route becomes a Folium polyline.
• Green and purple distinguish the two van assignments.
• Each route tooltip shows the van ID and planned meter distance.

• Replace the current dashboard table section with these completed accepted, rejected, and assignment tables:

st.caption("Accepted readings and nearest synthetic pipelines")
st.dataframe(
    accepted[
        [
            "reading_id",
            "vehicle_id",
            "methane_ppm",
            "pipeline_id",
            "snap_distance_m",
            "plume_id",
        ]
    ],
    hide_index=True,
)

st.caption("Rejected records")
st.dataframe(
    rejected[["reading_id", "rejection_reason"]],
    hide_index=True,
)

st.caption("Optimized inspection assignments")
st.dataframe(assignments, hide_index=True)

What does the assignment table prove?

The final table exposes each van visit as a row containing van_id, stop_order, and plume_id.

You can compare those rows with the plume markers and route colors on the map.

• Save streamlit_app.py.

Before you relaunch the dashboard, do you expect every plume marker to belong to one route?

• Launch the completed operations dashboard by running:

streamlit run streamlit_app.py

What does this check reveal?

The command reruns the complete pipeline and renders its dispatch outputs. The browser provides a visual check that the solver result reaches the map and tables.

You should see a depot marker and two color-coded routes. Every numbered plume marker should identify one assigned van.

The assignment table should contain each of the three plume IDs once. Each route should start at the depot and return to it.

• Return to PowerShell after checking the routes.
• Press Ctrl+C to stop the Streamlit server before creating the tests.

Routes or assignments missing?

• Check that routes comes from results["routes"] near the top of streamlit_app.py.
• Confirm the route loop appears before st_folium renders the map.
• Confirm the final dataframe receives assignments.

Help me find why the optimized routes are missing from my Folium map.

✔️ Awesome, I've got everything!

Your dashboard now shows the operational handoff from plume candidate to assigned inspection route.

ⓧ I'd like to double check the full code

import folium
import streamlit as st
from streamlit_folium import st_folium

from geomethane_ops import DEPOT_LON_LAT, run_pipeline

st.set_page_config(page_title="GeoMethane-Ops", layout="wide")

results = run_pipeline()
accepted = results["accepted"]
rejected = results["rejected"]
pipelines = results["pipelines"]
plumes = results["plumes"]
assignments = results["assignments"]
routes = results["routes"]

st.title("GeoMethane-Ops")
st.caption(
    "Synthetic operations demo. Plume candidates are analytical priorities, "
    "not confirmed leaks or safety thresholds."
)

metric_columns = st.columns(4)
metric_columns[0].metric("Accepted readings", len(accepted))
metric_columns[1].metric("Rejected readings", len(rejected))
metric_columns[2].metric("Plume candidates", len(plumes))
metric_columns[3].metric("Vans dispatched", assignments["van_id"].nunique())

operations_map = folium.Map(
    location=[37.802, -122.240],
    zoom_start=14,
    tiles="CartoDB Positron",
)

folium.GeoJson(
    pipelines,
    name="Synthetic utility pipelines",
    style_function=lambda feature: {"color": "#475569", "weight": 5},
    tooltip=folium.GeoJsonTooltip(
        fields=["pipeline_id", "pressure_class"],
        aliases=["Pipeline", "Pressure class"],
    ),
).add_to(operations_map)

plume_colors = {-1: "#2563eb", 1: "#dc2626", 2: "#ea580c", 3: "#a21caf"}
for row in accepted.itertuples():
    plume_id = int(row.plume_id)
    color = plume_colors.get(plume_id, "#dc2626")
    folium.PolyLine(
        [
            [row.latitude, row.longitude],
            [row.snapped_latitude, row.snapped_longitude],
        ],
        color="#94a3b8",
        weight=1,
        dash_array="4",
    ).add_to(operations_map)
    folium.CircleMarker(
        location=[row.latitude, row.longitude],
        radius=7 if plume_id >= 0 else 4,
        color=color,
        fill=True,
        fill_color=color,
        fill_opacity=0.8,
        tooltip=(
            f"{row.reading_id} | {row.methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {row.snap_distance_m:.1f} m"
        ),
    ).add_to(operations_map)

folium.Marker(
    location=[DEPOT_LON_LAT[1], DEPOT_LON_LAT[0]],
    tooltip="Inspection depot",
).add_to(operations_map)

assignment_lookup = (
    assignments.set_index("plume_id")["van_id"].to_dict()
    if not assignments.empty
    else {}
)
for row in plumes.itertuples():
    van_id = assignment_lookup.get(int(row.plume_id), "Unassigned")
    folium.Marker(
        location=[row.geometry.y, row.geometry.x],
        tooltip=(
            f"Plume {row.plume_id} | peak {row.peak_methane_ppm:.2f} ppm | "
            f"{row.pipeline_id} | {van_id}"
        ),
    ).add_to(operations_map)

route_colors = ["#16a34a", "#7c3aed"]
for route_index, route in enumerate(routes):
    folium.PolyLine(
        route["coordinates"],
        color=route_colors[route_index % len(route_colors)],
        weight=5,
        tooltip=f"{route['van_id']} | {route['distance_m']} m",
    ).add_to(operations_map)

folium.LayerControl().add_to(operations_map)
st_folium(
    operations_map,
    height=620,
    use_container_width=True,
    returned_objects=[],
)

st.caption("Accepted readings and nearest synthetic pipelines")
st.dataframe(
    accepted[
        [
            "reading_id",
            "vehicle_id",
            "methane_ppm",
            "pipeline_id",
            "snap_distance_m",
            "plume_id",
        ]
    ],
    hide_index=True,
)

st.caption("Rejected records")
st.dataframe(
    rejected[["reading_id", "rejection_reason"]],
    hide_index=True,
)

st.caption("Optimized inspection assignments")
st.dataframe(assignments, hide_index=True)

How should I use this reference?

Compare every dashboard section with your saved file. The data loading, map layers, metrics, and tables should appear in this order.

Test the complete operations pipeline

A successful dashboard run proves the current fixture works once. pytest turns the expected data-quality and routing behavior into checks you can rerun after future changes.

• In the Visual Studio Code Explorer sidebar, right-click the project folder.
• Choose the option for creating a new file.
• Enter tests/test_pipeline.py as the file path.
• Add the validation test to the new file:

from geomethane_ops import load_data, run_pipeline, validate_telemetry


def test_validation_separates_bad_records():
    raw, _ = load_data()
    clean, rejected = validate_telemetry(raw)

    assert len(raw) == 23
    assert len(clean) == 20
    assert len(rejected) == 3
    assert set(rejected["rejection_reason"]) == {
        "duplicate_reading_id",
        "invalid_timestamp",
        "invalid_methane",
    }

What does the validation test protect?

This test locks in the fixture's 23 raw rows and its 20 accepted rows. It also checks the three explicit rejection reasons.

• Save tests/test_pipeline.py.
• Run the first test from PowerShell with:

pytest

What does pytest do here?

Pytest discovers the test file and runs each function whose name begins with test_.

The command should complete with one passing test. Your rejection contract is now protected.

First test failing?

• Confirm PowerShell is running from the project folder that contains geomethane_ops.py.
• Check that the test file path is exactly tests/test_pipeline.py.
• Compare the three expected rejection-reason strings with validate_telemetry.

Help me troubleshoot why the telemetry validation test fails.

• Add the spatial snapping and plume-detection test below the first test:

def test_spatial_snap_and_plume_detection():
    results = run_pipeline()

    assert results["accepted"]["pipeline_id"].notna().all()
    assert results["accepted"]["snap_distance_m"].max() < 1000
    assert len(results["plumes"]) == 3
    assert set(results["plumes"]["plume_id"]) == {1, 2, 3}

What does the spatial test protect?

• Every accepted reading must receive a pipeline assignment.
• The fixture's largest snap distance must remain below 1000 meters.
• The default clustering settings must continue producing plume IDs 1, 2, and 3.

• Save tests/test_pipeline.py.
• Run the expanded suite from PowerShell with:

pytest

What does this run confirm?

The suite now checks data validation and spatial processing together. Both tests should complete successfully.

Spatial test failing?

• Confirm run_pipeline passes the snapped projected points into detect_plumes.
• Check that the default methane threshold remains 2.8.
• Check that the default DBSCAN radius remains 180 meters.

Help me identify why the snapping or plume assertions fail.

• Add the exactly-once dispatch test below the spatial test:

def test_dispatch_visits_each_plume_once():
    results = run_pipeline()
    expected = sorted(results["plumes"]["plume_id"].tolist())
    visited = sorted(
        plume_id
        for route in results["routes"]
        for plume_id in route["plume_ids"]
    )

    assert visited == expected
    assert len(results["assignments"]) == len(expected)

What does the dispatch test protect?

The test flattens every route's plume list into one sorted collection. That collection must match the plume candidates exactly.

The assignment row count must also equal the candidate count. A duplicated or missed visit fails the suite.

• Save tests/test_pipeline.py.

Before you run the completed suite, do you expect the validation, spatial, and dispatch contracts to pass together?

• Run all three automated tests from PowerShell with:

pytest

What does the complete suite verify?

Pytest executes all three transformations against the same synthetic fixture. This catches regressions from ingestion through dispatch.

You should see 3 passed. That result proves the validation split, spatial outputs, plume count, and route coverage all match the project's contract.

Dispatch test failing?

• Check that run_pipeline passes plumes_metric into optimize_dispatch.
• Confirm every non-depot node appends one plume_id to the route.
• Confirm each assignment row is created inside the non-depot branch.

Help me trace a missing or duplicated plume assignment.

✔️ Awesome, I've got everything!

All three tests pass. Your pipeline now has repeatable checks from raw telemetry through van dispatch.

ⓧ I'd like to double check the full code

from geomethane_ops import load_data, run_pipeline, validate_telemetry


def test_validation_separates_bad_records():
    raw, _ = load_data()
    clean, rejected = validate_telemetry(raw)

    assert len(raw) == 23
    assert len(clean) == 20
    assert len(rejected) == 3
    assert set(rejected["rejection_reason"]) == {
        "duplicate_reading_id",
        "invalid_timestamp",
        "invalid_methane",
    }


def test_spatial_snap_and_plume_detection():
    results = run_pipeline()

    assert results["accepted"]["pipeline_id"].notna().all()
    assert results["accepted"]["snap_distance_m"].max() < 1000
    assert len(results["plumes"]) == 3
    assert set(results["plumes"]["plume_id"]) == {1, 2, 3}


def test_dispatch_visits_each_plume_once():
    results = run_pipeline()
    expected = sorted(results["plumes"]["plume_id"].tolist())
    visited = sorted(
        plume_id
        for route in results["routes"]
        for plume_id in route["plume_ids"]
    )

    assert visited == expected
    assert len(results["assignments"]) == len(expected)

How should I use this reference?

Compare the imports and all three test functions with your saved file. The assertions should match this reference exactly.

Your tests prove the internal transformations. One final dashboard run proves those results still reach the operator-facing map.

Before you launch it, do you expect every plume to appear once in the assignment table and on one color-coded route?

• Launch the completed default dashboard from PowerShell by running:

streamlit run streamlit_app.py

What should the final dashboard show?

• The metrics should report 20 accepted readings and 3 rejected readings.
• The map should show the pipeline lines, snap connectors, three plume markers, depot, and two color-coded routes.
• The assignment table should list every plume exactly once.

Tests pass but the final dashboard differs?

• Confirm streamlit_app.py calls run_pipeline() without overriding the default settings.
• Confirm the browser refreshed after the latest saved changes.
• Compare the assignment table with the three plume IDs returned by the pipeline.

Help me reconcile passing tests with an incomplete dashboard.

That is the full operations loop working. GeoMethane-Ops now turns imperfect telemetry into tested plume priorities and a two-van inspection plan.

Secret mission

Make Plume Sensitivity Interactive

Your dashboard currently shows one fixed inspection scenario. Add two sidebar controls that recalculate plume candidates and dispatch routes from different operating assumptions.

Clean Up Your Resources

Clean Up Your Resources

Your GeoMethane-Ops resources can stay in place, pause for later, or be deleted entirely. Keeping these local files creates no ongoing service costs.

Resources you used:

• The local virtual environment at .venv.
• The synthetic telemetry file at data/telemetry.csv.
• The synthetic pipeline file at data/pipelines.geojson.
• The pipeline module at geomethane_ops.py.
• The interactive Streamlit dashboard at streamlit_app.py.
• The pinned package list at requirements.txt.
• The automated test suite at tests/test_pipeline.py.

Keep everything running

No action is needed. Choose this option if you want to keep testing different plume-detection scenarios.

• Leave the complete local project folder in place.

Your source files stay ready for another session. The project creates no ongoing service charges.

Pause - I'll come back to this later

Pause the local dashboard before stepping away. Your files remain ready for another session.

• Stop the Streamlit process in the PowerShell terminal from earlier if it is running.
• Run deactivate in the active PowerShell terminal.

The complete project stays in its current folder. Its local files use disk space without creating service charges.

Delete - I don't want to use this again

Removing the folder is permanent. This action affects only the local GeoMethane-Ops files.

• Stop the Streamlit process in the PowerShell terminal from earlier if it is running.
• Run deactivate in the active PowerShell terminal.
• Close the PowerShell window from earlier.
• Close the Visual Studio Code window from earlier.
• Use normal Windows file management to locate the project folder that was open in Visual Studio Code.
• Select that project folder.
• Delete the selected project folder.
• Check the former location to confirm that the project folder no longer appears.

Your virtual environment, synthetic data, dashboard, pipeline module, package list, and tests are now removed. No cloud resources or external service data remain to clean up.

Nice Work!

Nice Work!

Outstanding work! You built a tested local methane-inspection workflow that turns flawed mobile telemetry into defensible inspection priorities. The finished dashboard recomputes every operational result as you adjust its sensitivity controls.

You've learned how to:

• Enforce a data contract that separates 20 accepted readings from 3 rejected records with explicit reasons.
• Use a projected CRS with GeoPandas spatial indexing to link every accepted reading to its nearest fictional pipeline in meters.
• Cluster elevated readings into three DBSCAN plume candidates while preserving noise. Convert those candidates into balanced two-van routes with Google OR-Tools. Verify exactly-once dispatch coverage through pytest.
• Secret Mission: Add Streamlit sensitivity controls for the methane threshold plus the DBSCAN radius. Watch the plume count plus van assignments recompute for each scenario.

Ready to quiz yourself?