Build a Real-Data EV Swap Pipeline

Introduction

30 Second Summary

A public dataset can change as soon as someone cleans or reshapes it. Without a clear trail, nobody can tell which problems came from the source.

In this project, you will build a traceable local data pipeline from real Town of Cary EV charging sessions. Its audit trail separates source issues from controlled defects before PostgreSQL proves the repaired model.

What You'll Build

Your final demo shows a real public dataset moving from preserved source files to a repaired database with passing quality checks.

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

• A preserved source trail where the original CSV and metadata remain untouched beside their SHA-256 checksum values.
• A transparent failure-to-fix demo that separates natural source issues from deliberately planted defects.
• A queryable quality result with zero mismatched payments. It also shows zero orphan payments plus the five busiest stations through SQL checks.
• Secret Mission: Build a compact quality summary that groups natural issues, planted defects, and cleaning actions beside the raw-file checksum.

Are there any prerequisites?

You'll need a Mac with Python 3.11 or newer. This project also assumes that Docker Desktop and Visual Studio Code are installed.

Everything runs locally at no cost.

Before We Start

Before the hands-on work begins, commit to building a traceable pipeline that preserves the Town of Cary source records. You will track natural data-quality issues separately from the defects planted for cleaning practice.

Set Up the Local Pipeline

A reproducible data pipeline needs isolated dependencies. It also needs a database that can be rebuilt without changing your Mac.

In this step, you create a Python virtual environment for the pinned packages. You also define a local PostgreSQL service with Docker.

In this step, get ready to:

• Create the project folders and supplied configuration files.
• Create a virtual environment with the pinned Python packages.
• Start the local PostgreSQL service with Docker Compose.

Create the project structure and files

The project uses separate folders for untouched source files, generated tables, cleaned tables, and audit records. This boundary protects the original download from later transformations.

• Press Cmd+Space to open Spotlight.
• Type Visual Studio Code into Spotlight.
• Press Enter to open Visual Studio Code.

Why Open a Folder as a Workspace?

A workspace gives the editor and integrated terminal the same project context. Every file you create stays under one visible project folder.

• Click File in the top menu.
• Click Open Folder....
• Select your Desktop in the folder picker.
• Click New Folder.
• Enter swapgrid-real-data as the folder name.
• Click Open.
• Select Yes, I trust the authors if the Workspace Trust dialog appears.

You should see swapgrid-real-data at the top of the Explorer sidebar.

• Create src inside swapgrid-real-data with the New Folder icon.
• Create data inside swapgrid-real-data with the New Folder icon.
• Create raw inside data with the New Folder icon.
• Create landing inside data with the New Folder icon.
• Create clean inside data with the New Folder icon.
• Create audit inside data with the New Folder icon.

What Do These Folders Protect?

• The raw layer preserves the downloaded HTTP response bodies.
• The landing layer holds real-derived records with controlled defects.
• The clean layer holds repaired tables for PostgreSQL.
• The audit layer explains detected issues and recorded corrections.

The README.md records the dataset source before the pipeline touches any data. Its source details make every future table traceable.

• Create README.md in swapgrid-real-data with the New File... icon.
• Paste the project description and source details into README.md with this first block:

# SwapGrid Real-Data Pipeline

This local project treats public EV charging sessions as fictional SwapGrid battery swaps. The source download is preserved byte-for-byte in `data/raw/`; generated and modified tables live in separate folders.

## Data source

- Dataset: Electric Vehicle Charging Stations
- Publisher: Town of Cary, North Carolina
- Source page: https://data.townofcary.org/explore/dataset/electric-vehicle-charging-stations/information/
- Direct CSV export: https://data.townofcary.org/api/v2/catalog/datasets/electric-vehicle-charging-stations/exports/csv
- Metadata API: https://data.townofcary.org/api/v2/catalog/datasets/electric-vehicle-charging-stations
- Dataset-specific license: CC0 1.0 Universal
- License URL: https://creativecommons.org/publicdomain/zero/1.0/
- Attribution: Town of Cary, North Carolina

What Does This Documentation Establish?

• The opening explains how public charging sessions become fictional battery swaps.
• The source URLs identify the exact public dataset and metadata response.
• The license details record CC0 1.0 Universal for this dataset.
• The attribution credits the Town of Cary as the source.

• Save README.md with Cmd+S.
• Press Cmd+Shift+V to open the Markdown preview.

You should see the project title above a Data source section containing the publisher, source links, license, and attribution.

Does the Markdown Preview Look Incomplete?

Confirm that every source detail starts on its own line with a hyphen. Check that README.md is saved in the top level of swapgrid-real-data.

Help me compare the first section of my README with the supplied Markdown.

• Return to the README.md editor tab.
• Append the raw-data statement and local run commands below the attribution line:

The public records remain unchanged in `data/raw/`. SwapGrid station IDs, driver records, payment records, and controlled defects are project-generated learning artifacts.

## Run locally on macOS

```bash
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install -r requirements.txt
docker compose up -d
python src/download_data.py
python src/build_landing.py
python src/clean_data.py
python src/load_postgres.py
```

Why Record the Run Order?

The raw-data statement separates public records from generated learning artifacts. The command list captures the reproducible order for running the finished pipeline.

• Save README.md with Cmd+S.
• Return to the Markdown preview tab.

You should see a Run locally on macOS section with one formatted command block.

Are the Commands Rendering as Plain Text?

Check that the opening and closing triple backticks remain on separate lines. The word bash belongs directly after the opening backticks.

Help me fix the command block in my Markdown README.

• Return to the README.md editor tab.
• Append the layer descriptions and cleanup commands below the local run block:

## Data layers

- `data/raw/`: untouched HTTP response bodies from the official source
- `data/landing/`: real-derived and synthetic tables with controlled defects
- `data/clean/`: repaired tables ready for PostgreSQL
- `data/audit/`: natural-versus-planted problem manifest and row-level cleaning log

## Cleanup

Pause PostgreSQL but keep its data:

```bash
docker compose down
```

Delete the container, network, and named database volume:

```bash
docker compose down --volumes
```

What Do the Final Sections Add?

The layer list defines where each stage belongs. The cleanup section distinguishes pausing PostgreSQL from removing its named volume.

• Save README.md with Cmd+S.
• Return to the Markdown preview tab.

You should see four data-layer entries followed by separate pause and delete commands.

Are the Cleanup Blocks Merged Together?

Confirm that each cleanup command has its own opening and closing triple backticks. Keep a blank line around each fenced block.

Help me repair the cleanup section in my Markdown README.

✔️ Awesome, I've got everything!

Your README.md now documents the source, license, attribution, run order, data layers, and cleanup paths.

ⓧ I'd like to double check the full code

Compare your complete README.md with this reference.

# SwapGrid Real-Data Pipeline

This local project treats public EV charging sessions as fictional SwapGrid battery swaps. The source download is preserved byte-for-byte in `data/raw/`; generated and modified tables live in separate folders.

## Data source

- Dataset: Electric Vehicle Charging Stations
- Publisher: Town of Cary, North Carolina
- Source page: https://data.townofcary.org/explore/dataset/electric-vehicle-charging-stations/information/
- Direct CSV export: https://data.townofcary.org/api/v2/catalog/datasets/electric-vehicle-charging-stations/exports/csv
- Metadata API: https://data.townofcary.org/api/v2/catalog/datasets/electric-vehicle-charging-stations
- Dataset-specific license: CC0 1.0 Universal
- License URL: https://creativecommons.org/publicdomain/zero/1.0/
- Attribution: Town of Cary, North Carolina

The public records remain unchanged in `data/raw/`. SwapGrid station IDs, driver records, payment records, and controlled defects are project-generated learning artifacts.

## Run locally on macOS

```bash
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install -r requirements.txt
docker compose up -d
python src/download_data.py
python src/build_landing.py
python src/clean_data.py
python src/load_postgres.py
```

## Data layers

- `data/raw/`: untouched HTTP response bodies from the official source
- `data/landing/`: real-derived and synthetic tables with controlled defects
- `data/clean/`: repaired tables ready for PostgreSQL
- `data/audit/`: natural-versus-planted problem manifest and row-level cleaning log

## Cleanup

Pause PostgreSQL but keep its data:

```bash
docker compose down
```

Delete the container, network, and named database volume:

```bash
docker compose down --volumes
```

What Should Match?

This reference is the complete project README. Its source details and command order should match your file exactly.

The dependency file pins every library to a known release. Those pins keep later data transformations consistent across project runs.

• Create requirements.txt in swapgrid-real-data with the New File... icon.
• Paste the four pinned dependencies into requirements.txt with this block:

pandas==3.0.6
requests==2.34.2
SQLAlchemy==2.1.5
psycopg[binary]==3.3.6

What Are These Packages?

• pandas 3.0.6 reads, profiles, transforms, and writes the tabular data.
• Requests 2.34.2 downloads the public CSV and metadata responses.
• SQLAlchemy 2.1.5 creates the database connection used by pandas.
• Psycopg 3.3.6 provides the PostgreSQL driver in a self-contained binary installation.

• Save requirements.txt with Cmd+S.

You should see exactly four dependency lines in the editor.

Do the Version Pins Look Different?

Check every equals sign and version number. Confirm that binary remains inside the Psycopg brackets.

Help me find a typo in my pinned Python requirements.

✔️ Awesome, I've got everything!

Your dependency file contains the four releases required by the pipeline.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt with this reference.

pandas==3.0.6
requests==2.34.2
SQLAlchemy==2.1.5
psycopg[binary]==3.3.6

What Should Match?

The package names and version pins should match this four-line reference exactly.

The Compose file describes a repeatable local database. It fixes the PostgreSQL image version, connection values, localhost port, and persistent volume.

• Create compose.yaml in swapgrid-real-data with the New File... icon.
• Paste the PostgreSQL service definition into compose.yaml with this block:

services:
  postgres:
    image: postgres:18.6
    environment:
      POSTGRES_DB: swapgrid
      POSTGRES_USER: swapgrid
      POSTGRES_PASSWORD: swapgrid
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - swapgrid_pgdata:/var/lib/postgresql

volumes:
  swapgrid_pgdata:

What Does This Configuration Define?

• YAML indentation places one postgres service under services.
• postgres:18.6 pins the official PostgreSQL image release.
• POSTGRES_DB, POSTGRES_USER, and POSTGRES_PASSWORD initialize the local database with swapgrid.
• 127.0.0.1:5432:5432 exposes PostgreSQL only through your Mac's localhost interface.
• swapgrid_pgdata preserves database files outside the container.

• Save compose.yaml with Cmd+S.

You should see a postgres service and a top-level swapgrid_pgdata volume in the editor.

Does the YAML Show Indentation Warnings?

Use spaces for every indentation level. Keep volumes at the far left when it declares the named volume.

Help me fix the indentation in my Compose file.

✔️ Awesome, I've got everything!

Your Compose file now defines the local PostgreSQL service and its persistent named volume.

ⓧ I'd like to double check the full code

Compare your complete compose.yaml with this reference.

services:
  postgres:
    image: postgres:18.6
    environment:
      POSTGRES_DB: swapgrid
      POSTGRES_USER: swapgrid
      POSTGRES_PASSWORD: swapgrid
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - swapgrid_pgdata:/var/lib/postgresql

volumes:
  swapgrid_pgdata:

What Should Match?

The service name, image pin, initialization values, port binding, and volume path should match this reference exactly.

Create the virtual environment and install packages

A virtual environment keeps this project's libraries separate from packages used by other Python projects. Activating it directs the installation into .venv.

• Click View in the Visual Studio Code top menu.
• Click Terminal.

You should see an integrated terminal at the bottom of Visual Studio Code with swapgrid-real-data as its current folder.

• Create the isolated environment by running this command:

python3 -m venv .venv

What Does This Command Do?

The venv module creates a private Python environment in .venv. Project installations can now stay separate from your system Python.

• Confirm that .venv appears in the Explorer sidebar.

You now have a dedicated Python environment inside the project folder.

Did the Virtual Environment Fail to Appear?

Confirm that the integrated terminal is inside swapgrid-real-data. Check that your Python installation is version 3.11 or newer.

Help me create the virtual environment inside my project folder.

• Activate the virtual environment by running this command:

source .venv/bin/activate

What Does Activation Change?

Activation updates the current terminal session to use the Python executable inside .venv. Commands in this terminal now target the isolated environment.

Your terminal prompt should now show that the virtual environment is active.

Does the Prompt Look Unchanged?

Confirm that .venv exists in the project folder. Run the activation command from the integrated terminal inside swapgrid-real-data.

Help me activate my macOS Python virtual environment.

The installation may take a little time while Python downloads the pinned packages. Keep the terminal open until the command finishes.

• Install the project dependencies by running this command:

python3 -m pip install -r requirements.txt

How Does the Requirements File Control Installation?

The -r option tells pip to read package specifications from requirements.txt. The exact pins make each installation resolve to the same four direct dependency releases.

Wait for the installation command to return to the terminal prompt. You should see no unresolved installation error at the end of the output.

Did a Package Fail to Install?

Confirm that the virtual environment remains active. Compare every package pin in requirements.txt with the supplied reference.

Help me troubleshoot the pinned package installation.

Start the local PostgreSQL service

Docker Desktop runs the database inside a container. The Compose definition makes the same PostgreSQL service reproducible whenever you restart the project.

• Press Cmd+Space to open Spotlight.
• Type Docker into Spotlight.
• Press Enter to start Docker Desktop.
• Wait for Docker Desktop to finish starting.
• Return to the integrated terminal in Visual Studio Code.

Before you start the service, what evidence do you expect Terminal to show when PostgreSQL is ready?

• Start the PostgreSQL service in detached mode by running this command:

docker compose up -d

What Does This Command Start?

Docker Compose reads compose.yaml and creates the postgres service. Detached mode leaves the database running after the command returns.

The service uses PostgreSQL 18.6 on 127.0.0.1:5432. Its database name, user, and password are all swapgrid.

Terminal should report that the postgres service started without an error. That confirms your local database is running from the supplied Compose configuration.

Did the PostgreSQL Service Fail to Start?

Confirm that Docker Desktop has finished starting. Check that compose.yaml is saved in swapgrid-real-data.

Help me start the local PostgreSQL service.

That is the local foundation complete. Next, you download the untouched Town of Cary CSV and metadata into the protected raw layer.

Download the Real Charging Data

Your isolated Python environment is ready. Your local PostgreSQL service is running.

Before modeling or cleaning can alter the records, the pipeline needs a genuine public CSV dataset on disk. Preserving the original response creates evidence you can always return to.

Requests saves each HTTP response as unchanged bytes. A SHA-256 checksum gives each download a precise fingerprint.

In this step, get ready to:

• Create the downloader for the official CSV export and metadata response.
• Run the downloader to preserve both source responses in data/raw.
• Confirm the source attribution and dataset-specific license in README.md.

Create and review the downloader

The downloader keeps source locations separate from download behavior. Its helper rejects unsuccessful responses before writing any bytes.

• Select the src folder in the Explorer sidebar.
• Click the New File icon at the top of the Explorer sidebar.
• Enter download_data.py and press Enter.
• Add the source configuration and reusable download helper by pasting this code into src/download_data.py.

from hashlib import sha256
from pathlib import Path

import requests

RAW_DIR = Path("data/raw")
CSV_URL = (
    "https://data.townofcary.org/api/v2/catalog/datasets/"
    "electric-vehicle-charging-stations/exports/csv"
)
METADATA_URL = (
    "https://data.townofcary.org/api/v2/catalog/datasets/"
    "electric-vehicle-charging-stations"
)


def download(url: str, destination: Path) -> None:
    response = requests.get(url, timeout=60)
    response.raise_for_status()
    destination.write_bytes(response.content)
    digest = sha256(response.content).hexdigest()
    print(f"Saved {destination} | {len(response.content)} bytes | SHA-256 {digest}")

What Does This Code Do?

• RAW_DIR keeps both source responses inside the untouched raw layer.
• CSV_URL identifies the official charging-session export.
• METADATA_URL identifies the official dataset metadata response.
• requests.get(url, timeout=60) retrieves one response with a finite timeout.
• response.raise_for_status() stops the write when the server reports an unsuccessful request.
• destination.write_bytes(response.content) preserves the response body without converting its contents.
• sha256(response.content).hexdigest() calculates the fingerprint printed beside the byte count.

• Save src/download_data.py with Cmd+S (macOS) or Ctrl+S (Windows).
• Confirm download_data.py appears under src in the Explorer sidebar.

File Not Appearing Under src?

Check that you selected the existing src folder before creating the file. Move download_data.py into that folder if it appears beside src.

Help me place the downloader in the correct project folder.

The helper handles one response at a time. The main() function creates the raw folder before sending both destinations through that helper.

• Complete the downloader by pasting this code below download() in src/download_data.py.

def main() -> None:
    RAW_DIR.mkdir(parents=True, exist_ok=True)
    download(CSV_URL, RAW_DIR / "cary_ev_sessions.csv")
    download(METADATA_URL, RAW_DIR / "cary_ev_metadata.json")


if __name__ == "__main__":
    main()

How Does the Script Start?

• main() creates data/raw if the folder is missing.
• The first download() call writes the official export to cary_ev_sessions.csv.
• The second download() call writes the metadata response to cary_ev_metadata.json.
• if __name__ == "__main__": starts the workflow when you run this file directly.

• Save src/download_data.py with Cmd+S (macOS) or Ctrl+S (Windows).
• Confirm the unsaved indicator disappears from the download_data.py tab.

Seeing a Python Syntax Warning?

Check that main() starts at the far left of the file. Keep the two download() calls indented inside that function.

Help me compare my downloader with the expected Python structure.

✔️ Awesome, I've got everything!

Your saved downloader now contains the source URLs, binary writer, checksum calculation, and two output destinations.

ⓧ I'd like to double check the full code

Compare your complete src/download_data.py file with this reference.

from hashlib import sha256
from pathlib import Path

import requests

RAW_DIR = Path("data/raw")
CSV_URL = (
    "https://data.townofcary.org/api/v2/catalog/datasets/"
    "electric-vehicle-charging-stations/exports/csv"
)
METADATA_URL = (
    "https://data.townofcary.org/api/v2/catalog/datasets/"
    "electric-vehicle-charging-stations"
)


def download(url: str, destination: Path) -> None:
    response = requests.get(url, timeout=60)
    response.raise_for_status()
    destination.write_bytes(response.content)
    digest = sha256(response.content).hexdigest()
    print(f"Saved {destination} | {len(response.content)} bytes | SHA-256 {digest}")


def main() -> None:
    RAW_DIR.mkdir(parents=True, exist_ok=True)
    download(CSV_URL, RAW_DIR / "cary_ev_sessions.csv")
    download(METADATA_URL, RAW_DIR / "cary_ev_metadata.json")


if __name__ == "__main__":
    main()

Run the downloader and inspect the CSV

The raw layer stores the response bodies before any SwapGrid modeling begins. Every later transformation works from these preserved files.

Before you run the downloader, consider whether the CSV response and metadata response will produce the same byte count.

• Return to the Visual Studio Code terminal from the previous step.
• Download both official responses by running this command:

python src/download_data.py

What Does This Command Do?

• python src/download_data.py starts the script through its main() function.
• Each download() call saves the received bytes without parsing them.
• Each completed write prints its file path, byte count, and SHA-256 checksum.

You should see two lines beginning with Saved. Each line ends with a byte count plus a SHA-256 checksum.

That is your first real-data win. Both official responses are now preserved inside the raw layer.

• Expand data in the Explorer sidebar.
• Expand raw under data.
• Select cary_ev_sessions.csv.
• Read the first line of the CSV file.

You should see nine field names separated by semicolon characters. From left to right, the fields are start_date, station_name, charging_time_hh_mm_ss, energy_kwh, address_1, address_2, city, state_province, and zip_postal_code.

Downloads Missing or Incomplete?

Confirm your Mac has an active internet connection. Confirm the terminal path ends with swapgrid-real-data before using the same downloader command again.

If the request still fails, the source may be temporarily unavailable. Keep the raw files untouched while you retry the download.

Help me troubleshoot the downloader without modifying the raw responses.

Confirm the source and license

Source attribution explains where the preserved bytes came from. The dataset-specific license records how the public data can be reused.

Before you check, predict which source details should connect the raw files to their publisher.

• Select README.md in the Explorer sidebar.
• Find the Data source section.
• Confirm the dataset name is Electric Vehicle Charging Stations.
• Confirm the source page points to https://data.townofcary.org/explore/dataset/electric-vehicle-charging-stations/information/.
• Confirm the dataset-specific license is CC0 1.0 Universal.
• Confirm the attribution names the Town of Cary in North Carolina.

You should see the dataset name, publisher, source page, direct export, metadata API, license, and attribution recorded together. Those details make your preserved raw files traceable to their public source.

Your untouched source files now have checksums plus documented attribution. Next, you'll turn the real charging sessions into a deliberately dirty SwapGrid landing zone.

Build the Dirty Landing Zone

Your untouched Town of Cary files now give the pipeline a real foundation. This step turns those public sessions into a SwapGrid landing model while keeping the raw responses unchanged.

Real data alone may not contain enough cleaning work for a useful exercise. You will use pandas to profile natural issues before creating a visible failure state with disclosed planted defects.

In this step, get ready to:

• Profile the untouched public sessions for naturally occurring quality issues.
• Derive stations plus swaps from the public records.
• Generate deterministic drivers plus payments with controlled defects.

Profile the untouched source

A landing zone gives the pipeline a separate place for modeled data. The builder only reads from data/raw/cary_ev_sessions.csv.

• Select the existing src folder in the Explorer sidebar.
• Click the New File control.
• Enter build_landing.py as the file name.
• Add the imports plus source schema by pasting this code:

from datetime import date, datetime, timedelta
from hashlib import sha256
from pathlib import Path
import random

import pandas as pd

RAW_FILE = Path("data/raw/cary_ev_sessions.csv")
LANDING_DIR = Path("data/landing")
AUDIT_DIR = Path("data/audit")
RNG = random.Random(42)

SOURCE_COLUMNS = [
    "start_date",
    "station_name",
    "charging_time_hh_mm_ss",
    "energy_kwh",
    "address_1",
    "address_2",
    "city",
    "state_province",
    "zip_postal_code",
]
REQUIRED_COLUMNS = [column for column in SOURCE_COLUMNS if column != "address_2"]

What does this foundation define?

• The path constants keep the preserved source separate from generated landing files.
• The seeded random generator makes every driver assignment repeatable.
• The source schema lists the nine fields expected from the public CSV.
• The required-field list keeps address_2 optional.

• Save src/build_landing.py.
• Confirm that build_landing.py appears under src in the Explorer sidebar.

File missing from the Explorer sidebar?

Check that you selected the existing src folder before creating the file. Move the file into src if it was created beside that folder.

Ask for help if the file still appears in the wrong location: help me place build_landing.py inside the src folder

• Add the station schema plus helper functions below REQUIRED_COLUMNS by pasting this code:

STATION_COLUMNS = [
    "station_name",
    "address_1",
    "address_2",
    "city",
    "state_province",
    "zip_postal_code",
]


def stable_key(values) -> str:
    text = "\x1f".join("<NULL>" if pd.isna(value) else str(value) for value in values)
    return sha256(text.encode("utf-8")).hexdigest()


def alternate_date(value: str, pattern: str) -> str:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    return parsed.strftime(pattern)

Why use stable keys?

The public records do not contain SwapGrid identifiers. stable_key() converts the original values into a repeatable source-session key.

The same values always produce the same key. alternate_date() preserves a timestamp while changing only its display format.

• Save src/build_landing.py.
• Confirm that stable_key() appears above alternate_date() in the file.

Seeing red syntax markers?

Check that both function definitions begin at the far left. Keep each function body indented with four spaces.

Ask for help with the highlighted lines if the markers remain: help me fix the helper syntax in build_landing.py

• Add the start of main() below alternate_date() by pasting this code:

def main() -> None:
    LANDING_DIR.mkdir(parents=True, exist_ok=True)
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)

    sessions = pd.read_csv(RAW_FILE, sep=";")
    missing_columns = sorted(set(SOURCE_COLUMNS) - set(sessions.columns))
    if missing_columns:
        raise ValueError(f"Source schema changed; missing columns: {missing_columns}")

    sessions = sessions[SOURCE_COLUMNS].copy()
    sessions["source_session_key"] = sessions.apply(
        lambda row: stable_key(row[SOURCE_COLUMNS]), axis=1
    )

    print(f"Public source rows: {len(sessions)}")


if __name__ == "__main__":
    main()

How is the raw file protected?

• The explicit separator reads the semicolon-delimited source correctly.
• The schema check stops the script when an expected public field is absent.
• The copied DataFrame holds every later transformation in memory.
• The script never writes to data/raw/cary_ev_sessions.csv.

• Save src/build_landing.py.
• Switch back to the activated terminal from the previous step.
• Read the preserved source by running:

python src/build_landing.py

What does this first run prove?

Terminal prints the number of public source rows. The successful count proves that the expected schema loaded without rewriting the CSV.

• Confirm that Terminal shows a line beginning with Public source rows:.
• Confirm that data/landing now appears in the Explorer sidebar.
• Confirm that data/audit now appears in the Explorer sidebar.

Source file not loading?

Confirm that data/raw/cary_ev_sessions.csv still exists inside swapgrid-real-data. Check that the terminal remains inside the same folder.

Ask for help with the traceback without editing the raw file: help me diagnose why build_landing.py cannot read the preserved CSV

• Insert the duplicate plus missing-field profile immediately above the first print( line by pasting this code:

    manifest = []
    duplicate_count = int(sessions.duplicated(subset=SOURCE_COLUMNS).sum())
    if duplicate_count:
        manifest.append(
            {
                "origin": "natural",
                "table_name": "source_sessions",
                "issue_type": "duplicate_rows",
                "affected_rows": duplicate_count,
                "record_ids": "",
                "details": "Exact duplicate rows detected in the public CSV.",
            }
        )

    for column in REQUIRED_COLUMNS:
        count = int(sessions[column].isna().sum())
        if count:
            manifest.append(
                {
                    "origin": "natural",
                    "table_name": "source_sessions",
                    "issue_type": f"missing_{column}",
                    "affected_rows": count,
                    "record_ids": "",
                    "details": f"Missing required values observed in {column}.",
                }
            )

How are observed issues classified?

The profile runs before any controlled defects exist. Every condition found here receives the natural origin.

Natural entries keep record_ids empty because the source does not use project-generated SwapGrid IDs.

• Save src/build_landing.py.
• Run the updated profile with:

python src/build_landing.py

What does the successful rerun show?

Terminal prints the public row count again without a traceback. The duplicate plus missing-value checks can process the preserved source schema.

Seeing an indentation error?

Place the complete profile block inside main(). Every line in that block starts with at least four spaces.

Ask for help with the first reported line: help me fix the natural-profile indentation

• Replace the existing public-row summary with the date profile plus energy profile below:

    parsed_dates = pd.to_datetime(sessions["start_date"], errors="coerce", utc=True)
    invalid_dates = int(parsed_dates.isna().sum())
    if invalid_dates:
        manifest.append(
            {
                "origin": "natural",
                "table_name": "source_sessions",
                "issue_type": "invalid_start_date",
                "affected_rows": invalid_dates,
                "record_ids": "",
                "details": "Start dates that pandas could not parse.",
            }
        )

    non_positive = int((sessions["energy_kwh"].fillna(0) <= 0).sum())
    manifest.append(
        {
            "origin": "natural",
            "table_name": "source_sessions",
            "issue_type": "non_positive_energy",
            "affected_rows": non_positive,
            "record_ids": "",
            "details": "Public sessions with energy_kwh less than or equal to zero.",
        }
    )

    print(f"Public source rows: {len(sessions)}")
    print(f"Natural non-positive-energy sessions: {non_positive}")

What completes the source profile?

• Date coercion makes unparseable timestamps countable as missing values.
• The energy check includes missing values plus values at or below zero.
• The manifest records the observed count even when the count is zero.

Before you run the profile, predict whether the public source contains a non-positive energy session.

• Save src/build_landing.py.
• Reveal the observed energy result by running:

python src/build_landing.py

What should you observe?

Terminal prints a nonzero value after Natural non-positive-energy sessions:. That condition comes from the untouched public records.

You have separated source observations from every defect the project adds next.

Natural count missing?

Check that both summary lines remain inside main(). Confirm that the energy column is spelled energy_kwh.

Ask for help with the output if the count still does not print: help me inspect the natural energy profile

Build the landing tables

The public location fields become stations. Each public session becomes a fictional swap linked to a deterministic synthetic driver.

• Insert the station derivation immediately above the summary lines by pasting this code:

    stations = sessions[STATION_COLUMNS].fillna("").drop_duplicates(ignore_index=True)
    stations.insert(0, "station_id", [f"ST{i:03d}" for i in range(1, len(stations) + 1)])
    stations["station_key"] = stations.apply(
        lambda row: stable_key(row[STATION_COLUMNS]), axis=1
    )
    station_lookup = dict(zip(stations["station_key"], stations["station_id"]))

    session_station_keys = sessions[STATION_COLUMNS].fillna("").apply(
        lambda row: stable_key(row[STATION_COLUMNS]), axis=1
    )

How do sessions find their stations?

Unique public locations receive sequential station IDs such as ST001. The stable location key maps each session back to its derived station.

• Save src/build_landing.py.
• Run the station derivation with:

python src/build_landing.py

What does this rerun confirm?

The source profile still prints successfully. The station mapping can process every public session without changing the observed natural count.

Seeing a station-key error?

Check that STATION_COLUMNS includes the six location fields. Confirm that the temporary station_key column is created before station_lookup.

Ask for help with the traceback if the mapping still fails: help me fix the station-key mapping

• Insert the deterministic drivers plus real-derived swaps immediately above the summary lines by pasting:

    driver_ids = [f"DR{i:04d}" for i in range(1, 301)]
    drivers = pd.DataFrame(
        {
            "driver_id": driver_ids,
            "city": ["Cary"] * len(driver_ids),
            "signup_date": [
                (date(2021, 1, 1) + timedelta(days=index % 730)).isoformat()
                for index in range(len(driver_ids))
            ],
        }
    )

    swaps = pd.DataFrame(
        {
            "swap_id": [f"SW{i:06d}" for i in range(1, len(sessions) + 1)],
            "source_session_key": sessions["source_session_key"],
            "station_id": [station_lookup[key] for key in session_station_keys],
            "driver_id": [RNG.choice(driver_ids) for _ in range(len(sessions))],
            "swap_started_at": sessions["start_date"],
            "duration_hh_mm_ss": sessions["charging_time_hh_mm_ss"],
            "energy_kwh": sessions["energy_kwh"],
        }
    )

Why are the generated records repeatable?

The script always creates driver IDs from DR0001 through DR0300. The seeded generator repeats the same driver assignment on each run.

Each swap retains its source-session key. That link keeps every fictional swap traceable to the preserved source.

• Save src/build_landing.py.
• Run the growing landing builder with:

python src/build_landing.py

What does the model prove now?

Terminal reaches both profile summaries without a traceback. The script can create all 300 drivers plus one real-derived swap for every source session.

Driver or swap construction failing?

Check that driver_ids is defined before both DataFrames. Confirm that session_station_keys has the same length as sessions.

Ask for help with the failing DataFrame: help me fix the driver or swap construction

• Insert the payment model plus the first controlled defects immediately above the summary lines by pasting:

    payments = pd.DataFrame(
        {
            "payment_id": [f"PY{i:06d}" for i in range(1, len(swaps) + 1)],
            "swap_id": swaps["swap_id"],
            "amount": [round(2.50 + float(value) * 0.32, 2) for value in swaps["energy_kwh"]],
            "expected_amount": [
                round(2.50 + float(value) * 0.32, 2) for value in swaps["energy_kwh"]
            ],
            "status": ["paid"] * len(swaps),
            "paid_at": swaps["swap_started_at"],
        }
    )

    duplicate_ids = swaps.head(5)["swap_id"].tolist()
    for _, row in swaps.head(5).copy().iterrows():
        swaps.loc[len(swaps)] = row

    station_variant_ids = stations.head(3)["station_id"].tolist()
    for index, value in zip(stations.index[:3], ["CARY", "cary", "Cary "]):
        stations.loc[index, "city"] = value

    print(f"Planted duplicate swaps: {len(duplicate_ids)}")

What enters the dirty model first?

Every original swap receives one payment with an amount derived from its energy. The first five swaps are then copied to create five controlled duplicate rows.

The first three station cities receive known capitalization plus whitespace variants. Their exact station IDs remain available for the manifest.

Before you run the new model, predict the planted duplicate count that Terminal will add.

• Move the new planted-duplicate summary below the natural-energy summary.
• Save src/build_landing.py.
• Reveal the controlled duplicate count by running:

python src/build_landing.py

What should the latest run add?

Terminal now reports Planted duplicate swaps: 5. This is the intended duplicate failure state.

Duplicate count missing?

Check that duplicate_ids is captured before the five copied rows are appended. Keep the planted-duplicate summary inside main().

Ask for help if the count differs from five: help me inspect the planted swap duplicates

• Insert the driver variants plus mixed-date defects immediately above the summary lines by pasting:

    driver_variant_ids = drivers.head(3)["driver_id"].tolist()
    for index, value in zip(drivers.index[:3], ["CARY", "cary", "Cary "]):
        drivers.loc[index, "city"] = value

    missing_driver_id = drivers.loc[3, "driver_id"]
    drivers.loc[3, "city"] = None

    mixed_swap_ids = swaps.head(2)["swap_id"].tolist()
    swaps.loc[0, "swap_started_at"] = alternate_date(
        str(swaps.loc[0, "swap_started_at"]), "%m/%d/%Y %H:%M:%S"
    )
    swaps.loc[1, "swap_started_at"] = alternate_date(
        str(swaps.loc[1, "swap_started_at"]), "%d-%b-%Y %H:%M:%S"
    )

    mixed_driver_ids = drivers.head(2)["driver_id"].tolist()
    drivers.loc[0, "signup_date"] = alternate_date(
        f"{drivers.loc[0, 'signup_date']}T00:00:00+00:00", "%m/%d/%Y"
    )
    drivers.loc[1, "signup_date"] = alternate_date(
        f"{drivers.loc[1, 'signup_date']}T00:00:00+00:00", "%d-%b-%Y"
    )

Why track exact defect IDs?

Each controlled change targets known driver or swap IDs. The manifest can disclose precisely which records the project altered.

The alternate formats preserve the original moments. Only their string representations become inconsistent.

• Save src/build_landing.py.
• Run the builder with the city plus date defects by running:

python src/build_landing.py

What does this rerun establish?

Terminal still reports five planted duplicate swaps. The targeted city plus date changes complete without altering that disclosed count.

Date conversion failing?

Check that both calls use alternate_date(). Confirm that each original value is converted to a string before the helper receives it.

Ask for help with the failing record: help me fix the controlled date formats

• Insert the payment defects plus planted-issue definitions immediately above the summary lines by pasting:

    missing_payment_id = payments.loc[3, "payment_id"]
    payments.loc[3, "paid_at"] = None

    mismatch_payment_ids = payments.head(5)["payment_id"].tolist()
    payments.loc[payments.index[:5], "amount"] = (
        payments.loc[payments.index[:5], "amount"] + 7.00
    )

    planted = [
        ("swaps", "duplicate_rows", duplicate_ids, "Copied five swap rows."),
        ("stations", "city_variants", station_variant_ids, "Cary has three spellings."),
        ("drivers", "city_variants", driver_variant_ids, "Cary has three spellings."),
        ("drivers", "missing_city", [missing_driver_id], "One driver city is missing."),
        ("swaps", "mixed_date_formats", mixed_swap_ids, "Two swap dates use alternate formats."),
        ("drivers", "mixed_date_formats", mixed_driver_ids, "Two signup dates use alternate formats."),
        ("payments", "missing_paid_at", [missing_payment_id], "One payment timestamp is missing."),
        ("payments", "payment_mismatch", mismatch_payment_ids, "Five amounts exceed the expected amount by 7.00."),
    ]

    print(f"Planted payment mismatches: {len(mismatch_payment_ids)}")

What completes the failure state?

One payment loses its timestamp. Five payment amounts increase while their expected amounts remain unchanged.

The planted list connects every controlled issue to its table. It also keeps the exact affected record IDs.

Before you run the script, predict the planted payment-mismatch count that Terminal will add.

• Move the new payment-mismatch summary below the planted-duplicate summary.
• Save src/build_landing.py.
• Reveal the payment failure count by running:

python src/build_landing.py

What should Terminal report now?

Terminal adds Planted payment mismatches: 5. The landing model now contains both designed failures required by the cleaning step.

Payment mismatch count missing?

Check that mismatch_payment_ids is created before the first five amounts increase. Keep expected_amount unchanged.

Ask for help if the count differs from five: help me inspect the planted payment mismatches

Write the landing zone and manifest

Controlled defects become useful when their origins are disclosed. The final block writes the four dirty tables plus a manifest that separates observed source conditions from planted problems.

• Insert the manifest writer plus table writers immediately above the four summary lines by pasting:

    for table_name, issue_type, record_ids, details in planted:
        manifest.append(
            {
                "origin": "planted",
                "table_name": table_name,
                "issue_type": issue_type,
                "affected_rows": len(record_ids),
                "record_ids": "|".join(record_ids),
                "details": details,
            }
        )

    stations.drop(columns=["station_key"]).to_csv(
        LANDING_DIR / "stations.csv", index=False
    )
    drivers.to_csv(LANDING_DIR / "drivers.csv", index=False)
    swaps.to_csv(LANDING_DIR / "swaps.csv", index=False)
    payments.to_csv(LANDING_DIR / "payments.csv", index=False)
    pd.DataFrame(manifest).to_csv(AUDIT_DIR / "problem_manifest.csv", index=False)

What does the final block write?

• The four model tables are written under data/landing.
• The problem manifest is written under data/audit.
• Natural observations retain the natural origin.
• Controlled failures retain the planted origin plus exact record IDs.

Before you run the completed builder, predict the two planted counts that its summary will show.

• Save src/build_landing.py.
• Build the complete dirty landing zone by running:

python src/build_landing.py

What does the completed run prove?

Terminal reports Planted duplicate swaps: 5.

Terminal also reports Planted payment mismatches: 5 plus a nonzero natural non-positive-energy count.

That is the intended failure state. Your real-derived landing model now contains transparent problems that the cleaner can repair.

• Confirm that data/landing contains stations.csv.
• Confirm that data/landing contains drivers.csv.
• Confirm that data/landing contains swaps.csv.
• Confirm that data/landing contains payments.csv.
• Confirm that data/audit contains problem_manifest.csv.

✔️ Awesome, I've got everything!

Strong result. Your landing zone now exposes natural source issues plus known controlled defects without changing the raw files.

• Keep src/build_landing.py saved.
• Leave the successful Terminal summary visible for your checkpoint.

ⓧ I'd like to double check the full code

Compare your complete src/build_landing.py with this reference.

from datetime import date, datetime, timedelta
from hashlib import sha256
from pathlib import Path
import random

import pandas as pd

RAW_FILE = Path("data/raw/cary_ev_sessions.csv")
LANDING_DIR = Path("data/landing")
AUDIT_DIR = Path("data/audit")
RNG = random.Random(42)

SOURCE_COLUMNS = [
    "start_date",
    "station_name",
    "charging_time_hh_mm_ss",
    "energy_kwh",
    "address_1",
    "address_2",
    "city",
    "state_province",
    "zip_postal_code",
]
REQUIRED_COLUMNS = [column for column in SOURCE_COLUMNS if column != "address_2"]
STATION_COLUMNS = [
    "station_name",
    "address_1",
    "address_2",
    "city",
    "state_province",
    "zip_postal_code",
]


def stable_key(values) -> str:
    text = "\x1f".join("<NULL>" if pd.isna(value) else str(value) for value in values)
    return sha256(text.encode("utf-8")).hexdigest()


def alternate_date(value: str, pattern: str) -> str:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    return parsed.strftime(pattern)


def main() -> None:
    LANDING_DIR.mkdir(parents=True, exist_ok=True)
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)

    sessions = pd.read_csv(RAW_FILE, sep=";")
    missing_columns = sorted(set(SOURCE_COLUMNS) - set(sessions.columns))
    if missing_columns:
        raise ValueError(f"Source schema changed; missing columns: {missing_columns}")

    sessions = sessions[SOURCE_COLUMNS].copy()
    sessions["source_session_key"] = sessions.apply(
        lambda row: stable_key(row[SOURCE_COLUMNS]), axis=1
    )

    manifest = []
    duplicate_count = int(sessions.duplicated(subset=SOURCE_COLUMNS).sum())
    if duplicate_count:
        manifest.append(
            {
                "origin": "natural",
                "table_name": "source_sessions",
                "issue_type": "duplicate_rows",
                "affected_rows": duplicate_count,
                "record_ids": "",
                "details": "Exact duplicate rows detected in the public CSV.",
            }
        )

    for column in REQUIRED_COLUMNS:
        count = int(sessions[column].isna().sum())
        if count:
            manifest.append(
                {
                    "origin": "natural",
                    "table_name": "source_sessions",
                    "issue_type": f"missing_{column}",
                    "affected_rows": count,
                    "record_ids": "",
                    "details": f"Missing required values observed in {column}.",
                }
            )

    parsed_dates = pd.to_datetime(sessions["start_date"], errors="coerce", utc=True)
    invalid_dates = int(parsed_dates.isna().sum())
    if invalid_dates:
        manifest.append(
            {
                "origin": "natural",
                "table_name": "source_sessions",
                "issue_type": "invalid_start_date",
                "affected_rows": invalid_dates,
                "record_ids": "",
                "details": "Start dates that pandas could not parse.",
            }
        )

    non_positive = int((sessions["energy_kwh"].fillna(0) <= 0).sum())
    manifest.append(
        {
            "origin": "natural",
            "table_name": "source_sessions",
            "issue_type": "non_positive_energy",
            "affected_rows": non_positive,
            "record_ids": "",
            "details": "Public sessions with energy_kwh less than or equal to zero.",
        }
    )

    stations = sessions[STATION_COLUMNS].fillna("").drop_duplicates(ignore_index=True)
    stations.insert(0, "station_id", [f"ST{i:03d}" for i in range(1, len(stations) + 1)])
    stations["station_key"] = stations.apply(
        lambda row: stable_key(row[STATION_COLUMNS]), axis=1
    )
    station_lookup = dict(zip(stations["station_key"], stations["station_id"]))

    session_station_keys = sessions[STATION_COLUMNS].fillna("").apply(
        lambda row: stable_key(row[STATION_COLUMNS]), axis=1
    )

    driver_ids = [f"DR{i:04d}" for i in range(1, 301)]
    drivers = pd.DataFrame(
        {
            "driver_id": driver_ids,
            "city": ["Cary"] * len(driver_ids),
            "signup_date": [
                (date(2021, 1, 1) + timedelta(days=index % 730)).isoformat()
                for index in range(len(driver_ids))
            ],
        }
    )

    swaps = pd.DataFrame(
        {
            "swap_id": [f"SW{i:06d}" for i in range(1, len(sessions) + 1)],
            "source_session_key": sessions["source_session_key"],
            "station_id": [station_lookup[key] for key in session_station_keys],
            "driver_id": [RNG.choice(driver_ids) for _ in range(len(sessions))],
            "swap_started_at": sessions["start_date"],
            "duration_hh_mm_ss": sessions["charging_time_hh_mm_ss"],
            "energy_kwh": sessions["energy_kwh"],
        }
    )

    payments = pd.DataFrame(
        {
            "payment_id": [f"PY{i:06d}" for i in range(1, len(swaps) + 1)],
            "swap_id": swaps["swap_id"],
            "amount": [round(2.50 + float(value) * 0.32, 2) for value in swaps["energy_kwh"]],
            "expected_amount": [
                round(2.50 + float(value) * 0.32, 2) for value in swaps["energy_kwh"]
            ],
            "status": ["paid"] * len(swaps),
            "paid_at": swaps["swap_started_at"],
        }
    )

    duplicate_ids = swaps.head(5)["swap_id"].tolist()
    for _, row in swaps.head(5).copy().iterrows():
        swaps.loc[len(swaps)] = row

    station_variant_ids = stations.head(3)["station_id"].tolist()
    for index, value in zip(stations.index[:3], ["CARY", "cary", "Cary "]):
        stations.loc[index, "city"] = value

    driver_variant_ids = drivers.head(3)["driver_id"].tolist()
    for index, value in zip(drivers.index[:3], ["CARY", "cary", "Cary "]):
        drivers.loc[index, "city"] = value

    missing_driver_id = drivers.loc[3, "driver_id"]
    drivers.loc[3, "city"] = None

    mixed_swap_ids = swaps.head(2)["swap_id"].tolist()
    swaps.loc[0, "swap_started_at"] = alternate_date(
        str(swaps.loc[0, "swap_started_at"]), "%m/%d/%Y %H:%M:%S"
    )
    swaps.loc[1, "swap_started_at"] = alternate_date(
        str(swaps.loc[1, "swap_started_at"]), "%d-%b-%Y %H:%M:%S"
    )

    mixed_driver_ids = drivers.head(2)["driver_id"].tolist()
    drivers.loc[0, "signup_date"] = alternate_date(
        f"{drivers.loc[0, 'signup_date']}T00:00:00+00:00", "%m/%d/%Y"
    )
    drivers.loc[1, "signup_date"] = alternate_date(
        f"{drivers.loc[1, 'signup_date']}T00:00:00+00:00", "%d-%b-%Y"
    )

    missing_payment_id = payments.loc[3, "payment_id"]
    payments.loc[3, "paid_at"] = None

    mismatch_payment_ids = payments.head(5)["payment_id"].tolist()
    payments.loc[payments.index[:5], "amount"] = (
        payments.loc[payments.index[:5], "amount"] + 7.00
    )

    planted = [
        ("swaps", "duplicate_rows", duplicate_ids, "Copied five swap rows."),
        ("stations", "city_variants", station_variant_ids, "Cary has three spellings."),
        ("drivers", "city_variants", driver_variant_ids, "Cary has three spellings."),
        ("drivers", "missing_city", [missing_driver_id], "One driver city is missing."),
        ("swaps", "mixed_date_formats", mixed_swap_ids, "Two swap dates use alternate formats."),
        ("drivers", "mixed_date_formats", mixed_driver_ids, "Two signup dates use alternate formats."),
        ("payments", "missing_paid_at", [missing_payment_id], "One payment timestamp is missing."),
        ("payments", "payment_mismatch", mismatch_payment_ids, "Five amounts exceed the expected amount by 7.00."),
    ]
    for table_name, issue_type, record_ids, details in planted:
        manifest.append(
            {
                "origin": "planted",
                "table_name": table_name,
                "issue_type": issue_type,
                "affected_rows": len(record_ids),
                "record_ids": "|".join(record_ids),
                "details": details,
            }
        )

    stations.drop(columns=["station_key"]).to_csv(
        LANDING_DIR / "stations.csv", index=False
    )
    drivers.to_csv(LANDING_DIR / "drivers.csv", index=False)
    swaps.to_csv(LANDING_DIR / "swaps.csv", index=False)
    payments.to_csv(LANDING_DIR / "payments.csv", index=False)
    pd.DataFrame(manifest).to_csv(AUDIT_DIR / "problem_manifest.csv", index=False)

    print(f"Public source rows: {len(sessions)}")
    print(f"Natural non-positive-energy sessions: {non_positive}")
    print(f"Planted duplicate swaps: {len(duplicate_ids)}")
    print(f"Planted payment mismatches: {len(mismatch_payment_ids)}")


if __name__ == "__main__":
    main()

How to use this reference

Use the full file to compare ordering plus indentation. Keep your editable copy in src/build_landing.py.

Summary missing or files not created?

Check that the completed file contains four adjacent summary lines near the bottom. Confirm that the table writers appear before those lines.

Keep every new main-body line indented with four spaces. A misplaced line can end main() too early.

Ask for help with the traceback plus terminal output: help me troubleshoot my SwapGrid landing builder

You built the intended dirty landing zone without touching the preserved source. Next up, you will clean the four tables while recording every removal plus repair in a row-level audit log.

Clean the Tables and Log Changes

The landing tables in Visual Studio Code now contain the defects you documented. Duplicate swaps plus mismatched payments make the current model fail reconciliation.

A clean model needs every removal or repair to stay explainable. In this step, a Python cleaner fixes the tables while a row-level audit log captures each change.

In this step, get ready to:

• Remove duplicate swaps plus unusable natural-source records.
• Standardize cities plus dates before repairing the payment records.
• Write four clean tables plus a row-level log of every correction.

Remove unusable swap records

Each source_session_key traces a swap back to one public charging session. The cleaner keeps one swap per key before removing records with unusable energy or dates.

• Select the src folder in the Visual Studio Code file tree.
• Use the new-file icon at the top of the file tree.
• Name the new file clean_data.py.
• Start the cleaner with its imports plus project paths by adding:

from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

LANDING_DIR = Path("data/landing")
CLEAN_DIR = Path("data/clean")
AUDIT_DIR = Path("data/audit")
PLANTED_DUPLICATE_IDS = {f"SW{i:06d}" for i in range(1, 6)}
PLANTED_SWAP_DATE_IDS = {"SW000001", "SW000002"}
PLANTED_DRIVER_DATE_IDS = {"DR0001", "DR0002"}

What sets the cleaning boundaries?

• Pandas reads each landing CSV into a table that the cleaner can filter or update.
• The directory constants keep landing inputs separate from clean outputs plus audit artifacts.
• The three planted-ID sets let the audit log label controlled defects without misclassifying them as natural issues.

• Add the cleaner setup plus its reusable audit recorder below the constants:

def main() -> None:
    CLEAN_DIR.mkdir(parents=True, exist_ok=True)
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)

    stations = pd.read_csv(LANDING_DIR / "stations.csv")
    drivers = pd.read_csv(LANDING_DIR / "drivers.csv")
    swaps = pd.read_csv(LANDING_DIR / "swaps.csv")
    payments = pd.read_csv(LANDING_DIR / "payments.csv")
    audit = []

    def record(table, record_id, field, old_value, new_value, action, origin):
        audit.append(
            {
                "changed_at_utc": datetime.now(timezone.utc).isoformat(),
                "table_name": table,
                "record_id": record_id,
                "field_name": field,
                "old_value": old_value,
                "new_value": new_value,
                "action": action,
                "origin": origin,
            }
        )

How does the audit recorder work?

• The four calls to pd.read_csv() load copies of the landing tables into memory.
• The audit list collects one dictionary for every removal or repair.
• The nested record() function stores the affected record plus the old value.
• The origin field preserves the distinction between natural issues plus planted defects.

• Add the duplicate-removal logic inside main() below record():

    duplicate_mask = swaps.duplicated(subset=["source_session_key"], keep="first")
    for _, row in swaps[duplicate_mask].iterrows():
        origin = "planted" if row["swap_id"] in PLANTED_DUPLICATE_IDS else "natural"
        record(
            "swaps",
            row["swap_id"],
            "source_session_key",
            row["source_session_key"],
            "<row removed>",
            "drop_duplicate",
            origin,
        )
    swaps = swaps.drop_duplicates(
        subset=["source_session_key"], keep="first", ignore_index=True
    )

How are duplicates handled?

• The duplicate mask finds repeated source_session_key values after the first occurrence.
• Each removed row receives either a planted or natural origin in the audit log.
• The final call keeps one canonical swap for each source session.

• Add the natural energy-quality filter below the duplicate-removal logic:

    usable_indices = []
    for index, row in swaps.iterrows():
        if pd.isna(row["energy_kwh"]) or float(row["energy_kwh"]) <= 0:
            record(
                "swaps",
                row["swap_id"],
                "energy_kwh",
                row["energy_kwh"],
                "<row removed>",
                "drop_non_positive_energy",
                "natural",
            )
        else:
            usable_indices.append(index)
    swaps = swaps.loc[usable_indices].copy().reset_index(drop=True)

Why remove these sessions?

A fictional battery swap needs a positive energy transfer. The cleaner records these removals as natural because the values came from the untouched public source.

• Add the date parser below the energy filter:

    parsed_swap_dates = pd.to_datetime(
        swaps["swap_started_at"], errors="coerce", utc=True, format="mixed"
    )
    valid_swap_dates = []
    for index, parsed in parsed_swap_dates.items():
        swap_id = swaps.loc[index, "swap_id"]
        if pd.isna(parsed):
            record(
                "swaps",
                swap_id,
                "swap_started_at",
                swaps.loc[index, "swap_started_at"],
                "<row removed>",
                "drop_invalid_date",
                "natural",
            )
        else:
            valid_swap_dates.append(index)

How does date validation work?

• The mixed-format parser accepts the valid source representation plus the two planted alternatives.
• The coercion setting converts unparseable values into missing timestamps that the cleaner can detect.
• An invalid source date causes the whole swap to be removed because the event cannot be placed reliably in time.

• Continue the same date loop with planted-date normalization plus the final valid-row filter:

            if swap_id in PLANTED_SWAP_DATE_IDS:
                old_value = swaps.loc[index, "swap_started_at"]
                new_value = parsed.strftime("%Y-%m-%dT%H:%M:%SZ")
                swaps.loc[index, "swap_started_at"] = new_value
                record(
                    "swaps",
                    swap_id,
                    "swap_started_at",
                    old_value,
                    new_value,
                    "normalize_date",
                    "planted",
                )
    swaps = swaps.loc[valid_swap_dates].copy().reset_index(drop=True)

Why normalize only two valid dates?

The two identified swap dates are valid planted variants. The cleaner converts them to one UTC format before recording each repair as planted.

• Add the temporary output plus verification footer below the swap-date logic:

    stations.to_csv(CLEAN_DIR / "stations.csv", index=False)
    drivers.to_csv(CLEAN_DIR / "drivers.csv", index=False)
    swaps.to_csv(CLEAN_DIR / "swaps.csv", index=False)
    payments.to_csv(CLEAN_DIR / "payments.csv", index=False)
    pd.DataFrame(audit).to_csv(AUDIT_DIR / "cleaning_audit_log.csv", index=False)

    duplicate_count = int(swaps.duplicated(subset=["source_session_key"]).sum())
    mismatch_count = int(
        (payments["amount"].round(2) != payments["expected_amount"].round(2)).sum()
    )
    orphan_count = int((~payments["swap_id"].isin(set(swaps["swap_id"]))).sum())
    print(f"Remaining duplicate source-session keys: {duplicate_count}")
    print(f"Remaining payment mismatches: {mismatch_count}")
    print(f"Remaining orphan payments: {orphan_count}")
    print(f"Audit entries: {len(audit)}")


if __name__ == "__main__":
    main()

What does the footer prove?

• The five writes create the first clean-table snapshot plus its audit log.
• The three quality counts expose any duplicate keys plus payment reconciliation problems.
• The final count confirms that the cleaner recorded at least one action.

• Save src/clean_data.py.

Before you run this first pass, do you expect swap cleanup alone to make every payment check reach zero?

• Run the first cleaning pass from the activated terminal with:

python src/clean_data.py

What does this first run test?

This run tests the row-level swap cleanup before payment repair exists. It also writes a checkpoint that proves the cleaner is reading the landing files.

You will see Remaining duplicate source-session keys: 0. The payment mismatch plus orphan checks still show nonzero values.

That shortfall is expected. Removing swaps can leave payments pointing at discarded rows, while the five planted amount mismatches still need explicit repairs.

Cleaner stopped before printing counts?

Check that every line added after def main() uses four spaces. Keep the planted-date block nested beneath the date loop.

Confirm that data/landing still contains all four CSV files created in the previous step.

Ask for help by sharing the traceback without changing the landing files: help me debug my clean_data.py traceback.

Repair standardized fields and payments

The first pass proved that row cleanup cannot reconcile related tables by itself. The next additions standardize valid values before removing orphan payments or recalculating amounts.

• Find the line that filters valid_swap_dates inside main().
• Insert the city standardization block directly below that line:

    for frame, table_name, id_column in [
        (stations, "stations", "station_id"),
        (drivers, "drivers", "driver_id"),
    ]:
        for index, row in frame.iterrows():
            old_value = row["city"]
            new_value = "Cary" if pd.isna(old_value) else str(old_value).strip().title()
            if old_value != new_value:
                frame.loc[index, "city"] = new_value
                record(
                    table_name,
                    row[id_column],
                    "city",
                    old_value,
                    new_value,
                    "standardize_city",
                    "planted",
                )

How are city values standardized?

• The same loop handles station cities plus driver cities.
• Missing driver cities become Cary.
• Whitespace removal plus title casing converts every planted spelling to the same value.
• Only changed values receive an audit entry.

• Add the driver-date normalization block below the city loop:

    parsed_signup_dates = pd.to_datetime(
        drivers["signup_date"], errors="coerce", utc=True, format="mixed"
    )
    for index, parsed in parsed_signup_dates.items():
        driver_id = drivers.loc[index, "driver_id"]
        if driver_id in PLANTED_DRIVER_DATE_IDS and not pd.isna(parsed):
            old_value = drivers.loc[index, "signup_date"]
            new_value = parsed.strftime("%Y-%m-%d")
            drivers.loc[index, "signup_date"] = new_value
            record(
                "drivers",
                driver_id,
                "signup_date",
                old_value,
                new_value,
                "normalize_date",
                "planted",
            )

What happens to driver dates?

The cleaner targets the two driver IDs recorded in the planted manifest. Their valid alternative formats become one ISO date representation with a matching audit entry.

• Add orphan-payment removal below the driver-date loop:

    valid_swap_ids = set(swaps["swap_id"])
    payment_indices = []
    for index, row in payments.iterrows():
        if row["swap_id"] not in valid_swap_ids:
            record(
                "payments",
                row["payment_id"],
                "swap_id",
                row["swap_id"],
                "<row removed>",
                "drop_orphan_payment",
                "natural",
            )
        else:
            payment_indices.append(index)
    payments = payments.loc[payment_indices].copy().reset_index(drop=True)

Why remove orphan payments?

A payment must reference a swap that survives cleaning. Payments linked to discarded natural-source sessions are removed before reconciliation so the clean tables retain valid relationships.

• Add the timestamp repair loop below orphan-payment removal:

    started_at_by_swap = dict(zip(swaps["swap_id"], swaps["swap_started_at"]))
    energy_by_swap = dict(zip(swaps["swap_id"], swaps["energy_kwh"]))
    for index, row in payments.iterrows():
        payment_id = row["payment_id"]
        swap_id = row["swap_id"]
        if pd.isna(row["paid_at"]):
            new_value = started_at_by_swap[swap_id]
            payments.loc[index, "paid_at"] = new_value
            record(
                "payments",
                payment_id,
                "paid_at",
                row["paid_at"],
                new_value,
                "fill_missing_timestamp",
                "planted",
            )

How is the missing timestamp repaired?

• The two dictionaries connect each surviving swap to its cleaned start time plus energy value.
• A missing paid_at receives the corresponding swap start time.
• The repair remains planted because the missing timestamp was deliberately introduced.

• Continue inside the payment loop with amount reconciliation:

        expected = round(2.50 + float(energy_by_swap[swap_id]) * 0.32, 2)
        payments.loc[index, "expected_amount"] = expected
        if round(float(row["amount"]), 2) != expected:
            record(
                "payments",
                payment_id,
                "amount",
                row["amount"],
                expected,
                "reconcile_payment",
                "planted",
            )
            payments.loc[index, "amount"] = expected

How are payment amounts reconciled?

Each expected amount is recalculated from the cleaned swap energy. A mismatched amount is logged before both payment columns receive the reconciled value.

• Save src/clean_data.py.

Verify the clean model and audit log

The cleaner now covers removals plus repairs before reaching its existing output footer. The final run tests the complete model across swap uniqueness plus payment relationships.

Before you run the completed cleaner, do you expect all three quality counts to reach zero?

• Run the completed cleaner from the activated terminal with:

python src/clean_data.py

What does the final run prove?

• A zero duplicate count proves that every cleaned swap has a unique source-session key.
• A zero mismatch count proves that payment amounts reconcile with cleaned swap energy.
• A zero orphan count proves that every payment points to a surviving swap.
• A nonzero audit count proves that the cleaner disclosed its removals plus repairs.

You will see Remaining duplicate source-session keys: 0. You will also see zero remaining payment mismatches plus zero remaining orphan payments.

That completes the repair loop. Your dirty landing data now produces a reconciled clean model without hiding what changed.

Still seeing a nonzero quality count?

Confirm that the city plus payment blocks appear before the five CSV writes. Code placed after those writes cannot affect the files produced by that run.

Check that the payment amount formula uses the cleaned value from energy_by_swap.

Compare the file with the complete version below. You can also ask for help with a remaining quality count.

• Expand data/clean in the Visual Studio Code file tree.
• Confirm that stations.csv is listed.
• Confirm that drivers.csv is listed.
• Confirm that swaps.csv is listed.
• Confirm that payments.csv is listed.
• Select data/audit/cleaning_audit_log.csv in the file tree.
• Confirm that its header contains changed_at_utc through origin.

✔️ My cleaner works

Great work. Your clean tables now reconcile, while cleaning_audit_log.csv preserves the reason plus origin for every correction.

ⓧ I'd like to double check the full code

Compare src/clean_data.py with this complete version. Pay close attention to indentation inside main() plus the nested payment loop.

from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

LANDING_DIR = Path("data/landing")
CLEAN_DIR = Path("data/clean")
AUDIT_DIR = Path("data/audit")
PLANTED_DUPLICATE_IDS = {f"SW{i:06d}" for i in range(1, 6)}
PLANTED_SWAP_DATE_IDS = {"SW000001", "SW000002"}
PLANTED_DRIVER_DATE_IDS = {"DR0001", "DR0002"}


def main() -> None:
    CLEAN_DIR.mkdir(parents=True, exist_ok=True)
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)

    stations = pd.read_csv(LANDING_DIR / "stations.csv")
    drivers = pd.read_csv(LANDING_DIR / "drivers.csv")
    swaps = pd.read_csv(LANDING_DIR / "swaps.csv")
    payments = pd.read_csv(LANDING_DIR / "payments.csv")
    audit = []

    def record(table, record_id, field, old_value, new_value, action, origin):
        audit.append(
            {
                "changed_at_utc": datetime.now(timezone.utc).isoformat(),
                "table_name": table,
                "record_id": record_id,
                "field_name": field,
                "old_value": old_value,
                "new_value": new_value,
                "action": action,
                "origin": origin,
            }
        )

    duplicate_mask = swaps.duplicated(subset=["source_session_key"], keep="first")
    for _, row in swaps[duplicate_mask].iterrows():
        origin = "planted" if row["swap_id"] in PLANTED_DUPLICATE_IDS else "natural"
        record(
            "swaps",
            row["swap_id"],
            "source_session_key",
            row["source_session_key"],
            "<row removed>",
            "drop_duplicate",
            origin,
        )
    swaps = swaps.drop_duplicates(
        subset=["source_session_key"], keep="first", ignore_index=True
    )

    usable_indices = []
    for index, row in swaps.iterrows():
        if pd.isna(row["energy_kwh"]) or float(row["energy_kwh"]) <= 0:
            record(
                "swaps",
                row["swap_id"],
                "energy_kwh",
                row["energy_kwh"],
                "<row removed>",
                "drop_non_positive_energy",
                "natural",
            )
        else:
            usable_indices.append(index)
    swaps = swaps.loc[usable_indices].copy().reset_index(drop=True)

    parsed_swap_dates = pd.to_datetime(
        swaps["swap_started_at"], errors="coerce", utc=True, format="mixed"
    )
    valid_swap_dates = []
    for index, parsed in parsed_swap_dates.items():
        swap_id = swaps.loc[index, "swap_id"]
        if pd.isna(parsed):
            record(
                "swaps",
                swap_id,
                "swap_started_at",
                swaps.loc[index, "swap_started_at"],
                "<row removed>",
                "drop_invalid_date",
                "natural",
            )
        else:
            valid_swap_dates.append(index)
            if swap_id in PLANTED_SWAP_DATE_IDS:
                old_value = swaps.loc[index, "swap_started_at"]
                new_value = parsed.strftime("%Y-%m-%dT%H:%M:%SZ")
                swaps.loc[index, "swap_started_at"] = new_value
                record(
                    "swaps",
                    swap_id,
                    "swap_started_at",
                    old_value,
                    new_value,
                    "normalize_date",
                    "planted",
                )
    swaps = swaps.loc[valid_swap_dates].copy().reset_index(drop=True)

    for frame, table_name, id_column in [
        (stations, "stations", "station_id"),
        (drivers, "drivers", "driver_id"),
    ]:
        for index, row in frame.iterrows():
            old_value = row["city"]
            new_value = "Cary" if pd.isna(old_value) else str(old_value).strip().title()
            if old_value != new_value:
                frame.loc[index, "city"] = new_value
                record(
                    table_name,
                    row[id_column],
                    "city",
                    old_value,
                    new_value,
                    "standardize_city",
                    "planted",
                )

    parsed_signup_dates = pd.to_datetime(
        drivers["signup_date"], errors="coerce", utc=True, format="mixed"
    )
    for index, parsed in parsed_signup_dates.items():
        driver_id = drivers.loc[index, "driver_id"]
        if driver_id in PLANTED_DRIVER_DATE_IDS and not pd.isna(parsed):
            old_value = drivers.loc[index, "signup_date"]
            new_value = parsed.strftime("%Y-%m-%d")
            drivers.loc[index, "signup_date"] = new_value
            record(
                "drivers",
                driver_id,
                "signup_date",
                old_value,
                new_value,
                "normalize_date",
                "planted",
            )

    valid_swap_ids = set(swaps["swap_id"])
    payment_indices = []
    for index, row in payments.iterrows():
        if row["swap_id"] not in valid_swap_ids:
            record(
                "payments",
                row["payment_id"],
                "swap_id",
                row["swap_id"],
                "<row removed>",
                "drop_orphan_payment",
                "natural",
            )
        else:
            payment_indices.append(index)
    payments = payments.loc[payment_indices].copy().reset_index(drop=True)

    started_at_by_swap = dict(zip(swaps["swap_id"], swaps["swap_started_at"]))
    energy_by_swap = dict(zip(swaps["swap_id"], swaps["energy_kwh"]))
    for index, row in payments.iterrows():
        payment_id = row["payment_id"]
        swap_id = row["swap_id"]
        if pd.isna(row["paid_at"]):
            new_value = started_at_by_swap[swap_id]
            payments.loc[index, "paid_at"] = new_value
            record(
                "payments",
                payment_id,
                "paid_at",
                row["paid_at"],
                new_value,
                "fill_missing_timestamp",
                "planted",
            )

        expected = round(2.50 + float(energy_by_swap[swap_id]) * 0.32, 2)
        payments.loc[index, "expected_amount"] = expected
        if round(float(row["amount"]), 2) != expected:
            record(
                "payments",
                payment_id,
                "amount",
                row["amount"],
                expected,
                "reconcile_payment",
                "planted",
            )
            payments.loc[index, "amount"] = expected

    stations.to_csv(CLEAN_DIR / "stations.csv", index=False)
    drivers.to_csv(CLEAN_DIR / "drivers.csv", index=False)
    swaps.to_csv(CLEAN_DIR / "swaps.csv", index=False)
    payments.to_csv(CLEAN_DIR / "payments.csv", index=False)
    pd.DataFrame(audit).to_csv(AUDIT_DIR / "cleaning_audit_log.csv", index=False)

    duplicate_count = int(swaps.duplicated(subset=["source_session_key"]).sum())
    mismatch_count = int(
        (payments["amount"].round(2) != payments["expected_amount"].round(2)).sum()
    )
    orphan_count = int((~payments["swap_id"].isin(set(swaps["swap_id"]))).sum())
    print(f"Remaining duplicate source-session keys: {duplicate_count}")
    print(f"Remaining payment mismatches: {mismatch_count}")
    print(f"Remaining orphan payments: {orphan_count}")
    print(f"Audit entries: {len(audit)}")


if __name__ == "__main__":
    main()

A matching file produces the four cleaned CSVs plus data/audit/cleaning_audit_log.csv when you run it.

Your cleaned files now reconcile with a traceable history of every correction. Next, you will load the model into PostgreSQL to prove the same relationships with SQL queries.

Load PostgreSQL and Validate Quality

Your clean CSV files now agree on swaps and payments. They still cannot prove cross-table relationships as clearly as a relational database.

PostgreSQL will hold the clean model alongside its quality records. SQL joins can then expose payment mismatches or orphan records across tables.

The loader uses pandas DataFrame.to_sql to write each file into the database. SQLAlchemy uses Psycopg to connect that loading process to PostgreSQL.

In this step, get ready to:

• Map the clean tables and audit artifacts to six PostgreSQL tables.
• Run SQL checks for table counts and payment integrity.
• Compare the PostgreSQL swap count with the cleaned CSV row count.

Map the CSV files to PostgreSQL tables

A table map gives every clean or audit file a destination inside PostgreSQL. The database URL points to the local service that remains running from Step 1.

• In the Explorer sidebar in Visual Studio Code, select the src folder.
• Click the New File icon at the top of the Explorer sidebar.
• Type load_postgres.py as the file name.
• Press Enter.
• Add the connection settings and table map by pasting this code into src/load_postgres.py:

from pathlib import Path

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = "postgresql+psycopg://swapgrid:swapgrid@localhost:5432/swapgrid"
CLEAN_DIR = Path("data/clean")
AUDIT_DIR = Path("data/audit")

TABLE_FILES = {
    "stations": CLEAN_DIR / "stations.csv",
    "drivers": CLEAN_DIR / "drivers.csv",
    "swaps": CLEAN_DIR / "swaps.csv",
    "payments": CLEAN_DIR / "payments.csv",
    "problem_manifest": AUDIT_DIR / "problem_manifest.csv",
    "cleaning_audit_log": AUDIT_DIR / "cleaning_audit_log.csv",
}

What does this code set up?

• DATABASE_URL identifies the local swapgrid database and its PostgreSQL connection details.
• CLEAN_DIR points to the four cleaned model files.
• AUDIT_DIR points to the issue manifest and row-level cleaning log.
• TABLE_FILES pairs each PostgreSQL table name with the CSV file that supplies its rows.

• Save src/load_postgres.py with Cmd+S (macOS) or Ctrl+S (Windows).
• Select src/load_postgres.py in the Explorer sidebar.

You should see the new file beneath src. Its editor tab should contain the database URL and all six file mappings.

Missing a clean or audit file?

Confirm that the data/clean folder contains all four clean CSV files. Confirm that data/audit contains both audit CSV files.

If one is missing, return to the cleaner from the previous step before continuing.

help me identify which SwapGrid CSV file is missing from the loader map.

Define the SQL quality checks

The query dictionary turns the database load into a quality checkpoint. Each query answers one specific question about row totals or relationships between tables.

• Add the first portion of QUERIES below the TABLE_FILES dictionary by pasting this code:

QUERIES = {
    "table_counts": """
        SELECT 'stations' AS table_name, COUNT(*) AS row_count FROM stations
        UNION ALL
        SELECT 'drivers', COUNT(*) FROM drivers
        UNION ALL
        SELECT 'swaps', COUNT(*) FROM swaps
        UNION ALL
        SELECT 'payments', COUNT(*) FROM payments
    """,
    "mismatched_payments": """
        SELECT COUNT(*) AS mismatched_payments
        FROM payments
        WHERE amount <> expected_amount
    """,

What do these checks prove?

• table_counts reports the number of rows in each clean model table.
• mismatched_payments counts payments whose actual amount differs from the expected amount.
• A zero mismatch count proves that the payment repairs survived the database load.

• Complete the same QUERIES dictionary by pasting this code directly after the first portion:

    "orphan_payments": """
        SELECT COUNT(*) AS orphan_payments
        FROM payments AS p
        LEFT JOIN swaps AS s ON p.swap_id = s.swap_id
        WHERE s.swap_id IS NULL
    """,
    "busiest_stations": """
        SELECT st.station_name, COUNT(*) AS swap_count
        FROM swaps AS sw
        JOIN stations AS st ON sw.station_id = st.station_id
        GROUP BY st.station_name
        ORDER BY swap_count DESC
        LIMIT 5
    """,
}

How do the relationship checks work?

• orphan_payments uses a left join to find payments without a matching swap.
• busiest_stations joins swaps to station names before counting activity.
• LIMIT 5 keeps the final ranking focused on the five busiest stations.

• Save src/load_postgres.py.
• Press Cmd+F (macOS) or Ctrl+F (Windows) to open the editor search field.
• Enter QUERIES in the search field.

You should find one QUERIES definition. It should contain the labels table_counts, mismatched_payments, orphan_payments, and busiest_stations.

Seeing an unfinished query dictionary?

Check that the first portion ends after the mismatched_payments entry. Check that the second portion ends with the closing brace for QUERIES.

Keep every triple quote in place because each SQL query spans multiple lines.

help me find a missing quote or brace in my QUERIES dictionary.

Load the database and inspect the results

The main function loads all six files inside one database transaction. It then runs every quality query through the same connection.

• Add the loader function below the closing brace of QUERIES by pasting this code:

def main() -> None:
    engine = create_engine(DATABASE_URL)
    with engine.begin() as connection:
        for table_name, path in TABLE_FILES.items():
            frame = pd.read_csv(path)
            rows = frame.to_sql(
                name=table_name,
                con=connection,
                if_exists="replace",
                index=False,
                chunksize=1000,
                method="multi",
            )
            print(f"Loaded {table_name}: {rows} rows")

        for label, query in QUERIES.items():
            print(f"\n{label}")
            result = connection.execute(text(query))
            print(result.keys())
            for row in result.fetchall():
                print(tuple(row))

    engine.dispose()


if __name__ == "__main__":
    main()

What does the loader do?

• create_engine() prepares the PostgreSQL connection from DATABASE_URL.
• engine.begin() keeps the table loads and checks inside a managed transaction.
• frame.to_sql() replaces each destination table with the rows from its mapped CSV file.
• connection.execute() runs every query in QUERIES.
• engine.dispose() releases the database connection resources after the checks finish.

• Save src/load_postgres.py.

Before you run the loader, do you expect the database checks to preserve the clean files' zero mismatch and zero orphan results?

• Switch back to the terminal panel from earlier.
• Load the six tables and run the SQL checks by running this command:

python src/load_postgres.py

What should you see?

• The first six messages confirm that stations, drivers, swaps, payments, problem_manifest, and cleaning_audit_log were loaded.
• The table_counts section prints row totals for the four clean model tables.
• The mismatched_payments result equals 0.
• The orphan_payments result equals 0.
• The busiest_stations section prints five station rows with their swap counts.
• The number after Loaded swaps: matches the swaps row in the table_counts section.

Strong finish. Your cleaned model now survives relational checks inside PostgreSQL without losing its audit trail.

✔️ Awesome, I've got everything!

Your six PostgreSQL tables are loaded. Double check that mismatched_payments and orphan_payments both equal 0.

ⓧ I'd like to double check the full code

Compare your complete src/load_postgres.py file with this reference.

from pathlib import Path

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = "postgresql+psycopg://swapgrid:swapgrid@localhost:5432/swapgrid"
CLEAN_DIR = Path("data/clean")
AUDIT_DIR = Path("data/audit")

TABLE_FILES = {
    "stations": CLEAN_DIR / "stations.csv",
    "drivers": CLEAN_DIR / "drivers.csv",
    "swaps": CLEAN_DIR / "swaps.csv",
    "payments": CLEAN_DIR / "payments.csv",
    "problem_manifest": AUDIT_DIR / "problem_manifest.csv",
    "cleaning_audit_log": AUDIT_DIR / "cleaning_audit_log.csv",
}

QUERIES = {
    "table_counts": """
        SELECT 'stations' AS table_name, COUNT(*) AS row_count FROM stations
        UNION ALL
        SELECT 'drivers', COUNT(*) FROM drivers
        UNION ALL
        SELECT 'swaps', COUNT(*) FROM swaps
        UNION ALL
        SELECT 'payments', COUNT(*) FROM payments
    """,
    "mismatched_payments": """
        SELECT COUNT(*) AS mismatched_payments
        FROM payments
        WHERE amount <> expected_amount
    """,
    "orphan_payments": """
        SELECT COUNT(*) AS orphan_payments
        FROM payments AS p
        LEFT JOIN swaps AS s ON p.swap_id = s.swap_id
        WHERE s.swap_id IS NULL
    """,
    "busiest_stations": """
        SELECT st.station_name, COUNT(*) AS swap_count
        FROM swaps AS sw
        JOIN stations AS st ON sw.station_id = st.station_id
        GROUP BY st.station_name
        ORDER BY swap_count DESC
        LIMIT 5
    """,
}


def main() -> None:
    engine = create_engine(DATABASE_URL)
    with engine.begin() as connection:
        for table_name, path in TABLE_FILES.items():
            frame = pd.read_csv(path)
            rows = frame.to_sql(
                name=table_name,
                con=connection,
                if_exists="replace",
                index=False,
                chunksize=1000,
                method="multi",
            )
            print(f"Loaded {table_name}: {rows} rows")

        for label, query in QUERIES.items():
            print(f"\n{label}")
            result = connection.execute(text(query))
            print(result.keys())
            for row in result.fetchall():
                print(tuple(row))

    engine.dispose()


if __name__ == "__main__":
    main()

How to use this reference

Compare the connection settings and file map first. Then check the four query definitions and the complete main() function.

Loader failed or checks are nonzero?

• Confirm that the local PostgreSQL service from Step 1 is still running.
• Confirm that the terminal prompt still shows the activated .venv environment.
• Confirm that every path in TABLE_FILES points to an existing CSV file.
• Rerun the cleaner from the previous step if either payment check is nonzero.

help me diagnose why the SwapGrid PostgreSQL loader failed or returned a nonzero quality check.

Secret mission

Build a Quality Summary Report

Turn the detailed quality records into a reusable report. The output separates natural issues from planted defects. It also connects each report run to the preserved raw CSV through a fresh checksum.

Clean Up Your Resources

Clean Up Your Resources

Everything here runs on your Mac at no ongoing cost. Choose whether to keep the PostgreSQL service running, pause it for later, or delete the database volume and project files.

Resources you used:

• Local PostgreSQL 18.6 service defined in compose.yaml.
• Named Docker volume swapgrid_pgdata containing the local swapgrid database.
• Project folder swapgrid-real-data containing .venv, src, data/raw, data/landing, data/clean, and data/audit.

Keep everything running

No action is needed. Choose this option if you want to run more SQL checks or regenerate the quality report.

• Leave the PostgreSQL service running for additional database queries.
• Keep swapgrid_pgdata available with all six loaded tables.
• Keep swapgrid-real-data available with its raw, landing, clean, audit, and source files.

Pause - I'll come back to this later

Shut down the running database to free local resources. The named volume and every project file stay available for your return.

• Switch back to the terminal panel from earlier.
• Stop PostgreSQL while preserving swapgrid_pgdata by running this command:

docker compose down

What does this command do?

This stops the PostgreSQL container. It removes the Compose network.

The named volume swapgrid_pgdata remains available for your next start.

• Confirm Terminal returns to the prompt without an error.
• Restart PostgreSQL later from the swapgrid-real-data folder by running this command:

docker compose up -d

What happens when you restart?

Detached mode starts PostgreSQL in the background. The existing named volume reconnects to the service.

Your six database tables become available again without reloading the CSV files.

Delete - I don't want to use this again

Remove every local resource when you want a clean slate. Deleting the named volume permanently removes the database.

• Switch back to the terminal panel from earlier.
• Remove the PostgreSQL container, network, and swapgrid_pgdata volume by running this command:

docker compose down --volumes

What does this remove?

This stops the PostgreSQL container. It removes the Compose network.

It also deletes swapgrid_pgdata with the six database tables stored inside it.

• Confirm Terminal returns to the prompt without an error.
• Close the open swapgrid-real-data folder in Visual Studio Code.
• Delete the swapgrid-real-data project folder using Finder.
• Confirm swapgrid-real-data no longer appears in the location where you created it.

Nice Work!

Nice Work!

Mission accomplished! Your local SwapGrid pipeline now turns preserved Town of Cary charging data into a validated PostgreSQL model with traceable repairs.

You've learned how to:

• Preserve a real public dataset byte for byte. Connect it to its source metadata. Verify its identity with a SHA-256 checksum.
• Build real-derived stations and swaps. Generate deterministic drivers and payments. Separate natural issues from planted defects in a problem manifest.
• Load six related tables into PostgreSQL. Run SQL quality checks that prove payment reconciliation. Confirm zero orphan payments. Rank the five busiest stations.
• Secret Mission: Built a compact quality summary for natural issues and planted defects. Counted cleaning actions. Linked every report run to the current raw-file checksum.

Ready to quiz yourself?