Corrosion

Organizes, checks and normalizes corrosion survey data of underground gas pipelines:

• CIPS (Close Interval Potential Survey) and PCM (Pipeline Current Mapping): one Excel file per survey, listed in a master index.
• ACVG/DCVG: anomaly points, one workbook per year.

Start here: main.py. It is the entry point of the project and runs the whole pipeline: copy the files, check and clean them, normalize them to Excel/JSON, put each segment's CIPS and PCM in the same direction, and split the ACVG/DCVG anomalies per segment, then clean and normalize them on the CIPS line. After installing, run:

uv run main.py --n-jobs 8

Runtime dependencies (from pyproject.toml, locked in uv.lock): pandas, openpyxl, xlsxwriter, joblib, loguru, python-dotenv, python-slugify. Development tools: pytest, ruff, ty, notebook.

Rules the CIPS/PCM index must follow (checked when it loads, with the Excel row numbers of every problem):

• Area is filled.
• Segment is filled, or Sub Segment is (an empty Segment takes the Sub Segment).
• Segment + Diameter is unique (the same route with two diameters is two pipes and is allowed).

Each ACVG/DCVG area sheet needs Segmen, Lokasi Anomali, Kondisi Permukaan, Dia (inch), Latitude, Longitude, On Potential (volt), Off Potential (volt), IR Drop (%), Hasil ACVG (dB), Kedalaman Pipa (m), %drop PCM, Tgl DCVG, Tgl ACVG.

Other paths can be given on the command line (-i, -s, --acvg-index, --acvg-dir).

git clone https://github.com/capella-cgi/corrosions.git
cd corrosions
uv sync                 # creates .venv with Python 3.11 and all dependencies
uv run pytest           # optional: check the installation (all tests pass)

Optional: logging to logs/ is off by default. Turn it on with a .env file in the repo root:

ENABLE_LOG=true

1. Prepare the input files listed above (index workbooks in the repo root, data folders on D:\).

2. Run the pipeline:

   uv run main.py --n-jobs 8

   It loads the index (skipping 2021 by default), copies the referenced files into output/raw_data, checks, cleans and normalizes every CIPS and PCM file, writes file_index.json, syncs the survey direction, and finally splits the ACVG/DCVG anomalies per segment, cleans them and normalizes them (each anomaly gets the position and condition of the nearest reading of its segment's synced CIPS).

3. Check the reports in output/ (see Outputs):

   • checked-cips.xlsx / checked-pcm.xlsx: rows with is_valid = False and their reason;
   • file_index_excluded.json: segments missing a normalized CIPS or PCM file;
   • sync-report.xlsx: segments whose CIPS and PCM start more than 200 m apart (usually the wrong file in the index);
   • acvg-dcvg-report.xlsx: ACVG/DCVG groups with method = none (not linked to a segment), the load issues sheet, and the segments without ACVG-DCVG sheet, which explains every empty ACVG_DCVG cell (e.g. no anomaly within 500 m: no ACVG/DCVG survey on that segment that year).

4. Fix the index or the data and run again. Every step can be repeated; the sync never reverses a file twice.

Common options:

uv run main.py --type cips      # CIPS only (pcm, acvg also possible)
uv run main.py --type acvg      # ACVG/DCVG only (needs the normalized CIPS of an earlier run)
uv run main.py --skip-years     # process every year (default skips 2021)
uv run main.py -y 2021 2022     # skip 2021 and 2022
uv run main.py --no-sync        # keep the normalized files in survey order
uv run main.py -o D:\out        # another output root
uv run main.py --help           # all options

With --type all, a missing ACVG/DCVG index or workbook is skipped with a message so the CIPS/PCM run still finishes; --type acvg stops with the error instead, and never rewrites file_index.json.

Everything goes to output/ (or -o):

How the sync works: each segment's CIPS starts at its west end (west-east lines) or north end (north-south lines), and its PCM starts at the end closer to the CIPS start. Reversed files get their distances (and PCM loss rate / condition) recalculated; JSON and Excel stay identical.

How ACVG/DCVG is matched: a group whose segment name equals an index segment goes to that segment; every other anomaly goes, one by one, to the same-year CIPS track it lies on (up to 500 m away). So anomalies named after a whole pipeline are split over the index sections they lie on.

main.py is the normal way in. The classes below are what it calls, for notebooks or one-off checks:

from corrosions.data.file_index import FileIndex
from corrosions.data.cips import CIPS
from corrosions.data.pcm import PCM
from corrosions.data.acvg_dcvg import AcvgDcvg

# one file
cips = CIPS("output/raw_data/2025/CIPS/<file>.xlsx", year=2025).clean().check().save().normalize()
pcm = PCM("output/raw_data/2025/PCM/<file>.xlsx", year=2025).clean().check().save().normalize()

# the whole index (what main.py does)
fi = FileIndex("IDDA - PCM CIPS File List.xlsx", drop_columns=["Nomor Segment"], skip_years=[2021])
fi.rebuild(source_dir=r"D:\Data\Data IDDA")
fi.check_cips_file("output/raw_data", n_jobs=8)
fi.check_pcm_file("output/raw_data", n_jobs=8)
fi.to_json(n_jobs=8)                       # also syncs the CIPS/PCM direction
fi.sync_report

# ACVG/DCVG
acvg = AcvgDcvg("IDDA - ACVG FIle List.xlsx", skip_years=[2021])
acvg.load().match("output/file_index_idda-pcm-cips-file-list.csv").rebuild()
acvg.clean().normalize()                  # cleaned/<year>/ACVG_DCVG, normalize/acvg_dcvg
acvg.assign_index()
fi.assign_acvg_dcvg(acvg.normalized_files(), counts=acvg.anomaly_counts())  # acvg_dcvg_normalized_file + total_anomaly in file_index.json

Without an index, sync_files normalizes one segment's CIPS and PCM, puts them in the same direction (like the pipeline's sync) and, if an ACVG/DCVG file is given, normalizes its anomalies on the synced CIPS line:

from corrosions.sync import sync_files

result = sync_files(
    "CIPS - ICCP Demo Segment.xlsx",
    "PCM Demo Segment Reversed.xlsx",
    2024,
    acvg_dcvg="acvg-dcvg-demo-segment-8-jakarta.xlsx",   # optional
    output_dir="output",
)
# {
#     "cips_json": "output/normalize/cips/json/2024-cips-iccp-demo-segment.json",
#     "pcm_json": "output/normalize/pcm/json/2024-pcm-demo-segment-reversed.json",
#     "acvg_dcvg_json": "output/normalize/acvg_dcvg/json/2024-acvg-dcvg-demo-segment-8-jakarta.json",
#     "cips_axis": "north-south", "cips_reversed": False, "pcm_reversed": True, "start_gap_m": 50.34,
# }

The three JSON files are the normalized CIPS and PCM after the sync and the ACVG/DCVG anomalies with real_distance and closest_cips_condition (acvg_dcvg_json is None without an ACVG/DCVG file). The ACVG/DCVG file is one segment's anomalies with the columns listed in Input files (like the raw_data/<year>/ACVG_DCVG/ files). Example output: wiki/Normalizing-Data.md.

CIPS, PCM and ACVG/DCVG use the same chain:

Data(path, year=...).clean().check().save().normalize(...)

clean() drops empty rows, empty or 0 coordinates, rows missing a required value and duplicate coordinates; check() fills report; save() writes the cleaned copy to output/cleaned/<year>/<KIND>/; normalize() adds distances and a condition and writes output/normalize/<cips|pcm|acvg_dcvg>/<excel|json>/<year>-<slug>.* (<slug> = slugified source filename). save() and check() are optional.

Example: a 4-reading ICCP CIPS file CIPS - ICCP Demo Segment.xlsx (2024) gives output/normalize/cips/json/2024-cips-iccp-demo-segment.json:

[
  {
    "voltage": -1.1,
    "off_voltage": -0.9,
    "latitude": -6.2,
    "longitude": 106.8,
    "real_distance": 0.0,
    "condition": "PROTECTED",
    "comment": null,
    "dcp_feature_dcvg_anomaly": "Test Post TP-01"
  },
  {
    "voltage": -0.95,
    "off_voltage": -0.86,
    "latitude": -6.2009,
    "longitude": 106.8001,
    "real_distance": 100.6841260516,
    "condition": "PROTECTED",
    "comment": null,
    "dcp_feature_dcvg_anomaly": null
  }
]

and in cips.report: protection = "ICCP", n_normalized = 4, length_km = 0.302, protected_percentage = 75.0, unprotected_percentage = 25.0. The PCM and ACVG/DCVG versions of this example (input, Excel, JSON and report) are in wiki/Normalizing-Data.md.

check() (on CIPS, PCM and AcvgDcvgFile) never raises: it stores a quality summary on report (a dict, empty until check() runs) and returns the object, so it chains like the other steps. It checks whatever df holds when it is called, so its place in the chain matters: every batch step checks the cleaned data (clean().check().save() in check_cips_file, check_pcm_file and AcvgDcvg.clean()), so n_duplicates is 0 there. Call check() before clean() to inspect a raw file.

cips = CIPS("output/raw_data/2025/CIPS/<file>.xlsx", year=2025).check()
cips.report["is_valid"]          # False -> look at missing_columns / has_voltage
cips.report["missing_columns"]

normalize() then adds its results to the same report, keeping the check() keys (check() replaces the whole report, so run it first):

cips = cips.clean().normalize()
cips.report["protected_percentage"], cips.report["length_km"]

Every normalize() (CIPS, PCM, AcvgDcvgFile, and AcvgDcvg, which passes it on to each file) takes normalize_dir=None. It replaces the default output/normalize/<cips|pcm|acvg_dcvg>: the files go to <normalize_dir>/excel and <normalize_dir>/json.

pcm.clean().normalize(normalize_dir="D:/tmp/pcm")   # D:/tmp/pcm/json/<year>-<slug>.json

The sync and the ACVG/DCVG matching still read the normalized CIPS/PCM from output/normalize/, so main.py keeps the default.

check_cips_file / check_pcm_file put one report per file into checked-cips.xlsx / checked-pcm.xlsx, next to the clean/normalize results and a reason for any failure; the CIPS report leaves out the duplicates list (too long for an Excel cell).

Full API: wiki/API-Reference.md.

uv run pytest                  # tests (tests/)
uv run ruff check --fix src/   # lint
uv run ruff format .           # format
uvx ty check src/              # type check

Steps used after this pipeline; the notebooks below are not part of this repository:

1. Extract indirects > run data-2025.ipynb
2. Check /tests/files.xlsx
3. Run all.ipynb
4. Run sync.ipynb
5. Run calculate-stats.ipynb
6. Run seeder-builder.ipynb