feature: support parsing Norway (Banenor) kmm2 format

Norway files are identified by a "NorKmmToKmm" token in the VER header and
store WGS84 lat/lon in the coordinate columns instead of SWEREF99 TM
northing/easting. read_kmm2 now detects this and converts lat/lon to
northing/easting on read, so the rest of the pipeline (sync + geodetic)
behaves identically to Swedish files.

Also adds revert_banenor_kmm2.py, which reverses the earlier SWEREF
conversion applied to delivered Norway files, restoring the original
lat/lon format (only genuine SWEREF files are reverted).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: FelixAbrahamsson

kmm/positions/read_kmm2.py

@@ -5,10 +5,17 @@
 import numpy as np
 import pandas as pd
 from pydantic import validate_call
+from sweref99 import projections
 
 pattern = re.compile(r".+\[.+\]")
 pattern2 = re.compile(r"CMAST")
 
+# Norway ("Banenor") kmm2 files are identified by this token in the VER header
+# line. They store WGS84 latitude/longitude in the coordinate columns instead of
+# SWEREF99 TM northing/easting.
+norway_header_token = "NorKmmToKmm"
+tm = projections.make_transverse_mercator("SWEREF_99_TM")
+
 expected_columns = [
     "code",
     "centimeter",
@@ -46,6 +53,29 @@
 )
 
 
+def latlon_to_sweref(dataframe):
+    """
+    Norway kmm2 files store WGS84 latitude/longitude in the coordinate columns
+    (parsed here into the ``northing``/``easting`` columns of the shared layout).
+    Convert them to SWEREF99 TM northing/easting so the rest of the pipeline,
+    including the geodetic transform, behaves identically to Swedish files.
+    """
+    if len(dataframe) == 0 or not {"northing", "easting"}.issubset(dataframe.columns):
+        return dataframe
+
+    latitude = dataframe["northing"].to_numpy()
+    longitude = dataframe["easting"].to_numpy()
+    grid = [
+        (np.nan, np.nan)
+        if np.isnan(lat) or np.isnan(lon)
+        else tm.geodetic_to_grid(float(lat), float(lon))
+        for lat, lon in zip(latitude, longitude)
+    ]
+    dataframe["northing"] = np.array([n for n, _ in grid], dtype=np.float32)
+    dataframe["easting"] = np.array([e for _, e in grid], dtype=np.float32)
+    return dataframe
+
+
 @validate_call
 def read_kmm2(
     path: Path, raise_on_malformed_data: bool = True, replace_commas: bool = True
@@ -55,10 +85,12 @@ def read_kmm2(
         for index, line in enumerate(path.read_text(encoding="latin1").splitlines())
         if pattern.match(line) or pattern2.match(line)
     ]
+    norway = False
     with open(path, "r", encoding="latin1") as f:
         line = f.readline()
         if line.startswith("VER"):
             skiprows = [0] + skiprows
+            norway = norway_header_token in line
         elif raise_on_malformed_data and not line.startswith("POS"):
             raise ValueError("Malformed data, first line is not POS or VER")
 
@@ -92,12 +124,15 @@ def read_kmm2(
             else:
                 columns = expected_columns
 
-            return pd.read_csv(
+            dataframe = pd.read_csv(
                 file_obj,
                 **parser_kwargs,
                 names=columns,
                 dtype=expected_dtypes,
             )
+            if norway:
+                dataframe = latlon_to_sweref(dataframe)
+            return dataframe
         except pd.errors.EmptyDataError:
             return pd.DataFrame(columns=expected_columns)
     except Exception as e:
@@ -114,3 +149,20 @@ def test_patterns():
 
 def test_extra_columns():
     read_kmm2(Path("tests/extra_columns.kmm2"))
+
+
+def test_norway_latlon_to_sweref():
+    # Norway ("Banenor") files store WGS84 lat/lon in the coordinate columns;
+    # read_kmm2 must convert them to SWEREF99 TM northing/easting so the rest of
+    # the pipeline (including .geodetic()) recovers the original coordinates.
+    dataframe = read_kmm2(Path("tests/norway.kmm2"))
+    assert len(dataframe) == 4
+    # SWEREF99 TM: northing in the millions, easting in the hundred-thousands.
+    assert (dataframe["northing"] > 6_000_000).all()
+    assert (dataframe["easting"] > 100_000).all()
+
+    latitude, longitude = tm.grid_to_geodetic(
+        dataframe["northing"].iloc[0], dataframe["easting"].iloc[0]
+    )
+    assert abs(latitude - 59.9100025) < 1e-4
+    assert abs(longitude - 10.7545870) < 1e-4

revert_banenor_kmm2.py

@@ -0,0 +1,133 @@
+"""
+Reverse the SWEREF conversion that was applied to Norway ("Banenor") kmm2 files.
+
+The files in ``banenor-wire-positions-20260720`` that carry a
+``VER\\tNorKmmToKmm...`` header were originally delivered in the Norway format
+(WGS84 lat/lon, comma decimals, integer fields, 21 columns) but had their
+coordinate columns rewritten to SWEREF99 TM northing/easting (period decimals,
+float fields, 24 columns) so the old library could parse them.
+
+This script converts those files back to their original Norway format and
+writes them to ``banenor-wire-positions-20260720-converted-to-original``.
+Files without the Norway header are left untouched (not copied).
+"""
+
+from pathlib import Path
+
+from sweref99 import projections
+from tqdm import tqdm
+
+tm = projections.make_transverse_mercator("SWEREF_99_TM")
+
+input_dir = Path("/home/felix/data/banenor-wire-positions-20260720")
+output_dir = Path(
+    "/home/felix/data/banenor-wire-positions-20260720-converted-to-original"
+)
+
+NORWAY_TOKEN = "NorKmmToKmm"
+
+# Column layout of a converted file (kmm.positions.read_kmm2.expected_columns).
+# The original Norway format only keeps the first 21 columns.
+NORTHING = 10
+EASTING = 11
+N_ORIGINAL_COLUMNS = 21
+# Fields that are integers in the original Norway format but were written as
+# floats (e.g. "3.0") or zero-padded strings (e.g. "0010") after conversion.
+INT_COLUMNS = {1, 2, 3, 4, 6, 7, 8, 9, 12, 20}
+
+
+def revert_int(value):
+    value = value.strip()
+    if value == "":
+        return value
+    try:
+        return str(int(float(value)))
+    except ValueError:
+        # Some fields (e.g. alphanumeric track sections like "2d") are not
+        # numeric; leave them untouched.
+        return value
+
+
+def revert_coordinate(value):
+    # WGS84 with comma decimals, mirroring the original delivered precision.
+    return repr(float(value)).replace(".", ",")
+
+
+def revert_line(line):
+    if not line.startswith("POS"):
+        return line
+    fields = line.split("\t")
+
+    northing = float(fields[NORTHING])
+    easting = float(fields[EASTING])
+    latitude, longitude = tm.grid_to_geodetic(northing, easting)
+    fields[NORTHING] = revert_coordinate(latitude)
+    fields[EASTING] = revert_coordinate(longitude)
+
+    for index in INT_COLUMNS:
+        if index < len(fields):
+            fields[index] = revert_int(fields[index])
+
+    return "\t".join(fields[:N_ORIGINAL_COLUMNS])
+
+
+def revert_file(path):
+    # latin1 + CRLF, matching the original delivery.
+    text = path.read_text(encoding="latin1")
+    lines = text.split("\n")
+    out_lines = [revert_line(line.rstrip("\r")) for line in lines if line != ""]
+    return "\r\n".join(out_lines) + "\r\n"
+
+
+def classify(path):
+    """
+    Categorise a kmm2 file:
+      - "non-norway": missing the NorKmmToKmm header (leave as is).
+      - "sweref":     norway header, coordinate columns hold SWEREF99 TM
+                      northing/easting (converted) -> reverse to lat/lon.
+      - "latlon":     norway header, coordinate columns already hold WGS84
+                      lat/lon (never converted) -> no reversion needed.
+      - "unknown":    norway header but no parseable/degenerate coordinates.
+    """
+    with open(path, "r", encoding="latin1") as f:
+        first_line = f.readline()
+        if not (first_line.startswith("VER") and NORWAY_TOKEN in first_line):
+            return "non-norway"
+        for line in f:
+            if line.startswith("POS"):
+                fields = line.rstrip("\r\n").split("\t")
+                try:
+                    northing = float(fields[NORTHING].replace(",", "."))
+                except (ValueError, IndexError):
+                    return "unknown"
+                if abs(northing) > 100_000:
+                    return "sweref"
+                if 3 < abs(northing) < 100:
+                    return "latlon"
+                return "unknown"
+    return "unknown"
+
+
+def main():
+    output_dir.mkdir(exist_ok=True)
+    paths = sorted(input_dir.glob("*.kmm2"))
+
+    categories = {"non-norway": [], "sweref": [], "latlon": [], "unknown": []}
+    for path in tqdm(paths):
+        kind = classify(path)
+        categories[kind].append(path.name)
+        if kind == "sweref":
+            (output_dir / path.name).write_text(revert_file(path), encoding="latin1")
+
+    print(f"reverse-converted (sweref -> original latlon): {len(categories['sweref'])}")
+    print(f"non-norway (left as is):                       {len(categories['non-norway'])}")
+    print(f"already original latlon (left as is):          {len(categories['latlon'])}")
+    for name in categories["latlon"]:
+        print(f"    {name}")
+    print(f"unknown/degenerate (left as is):               {len(categories['unknown'])}")
+    for name in categories["unknown"]:
+        print(f"    {name}")
+
+
+if __name__ == "__main__":
+    main()

tests/norway.kmm2

@@ -0,0 +1,5 @@
+VER	NorKmmToKmm2_1.03	SplSetup NO_NORM
+POS	23060200	10	0	200	?	3	0	0	0	59,9100025118578	10,7545870162215	10						?		0
+POS	23060300	10	0	201	?	3	0	0	0	59,9099990016758	10,7546024063035	10						?		0
+POS	23060400	10	0	202	?	3	0	0	0	59,909994982491	10,754618120436	10						?		0
+POS	23060500	10	0	203	?	3	0	0	0	59,9099912415574	10,7546337465434	10						?		0