feat: CoordConverter 坐标转换套件 v1.0.0

TXT / Excel(CSV) / SHP 三种格式任意互转
支持 WGS84 / CGCS2000 / Xian80 / Beijing54 / Web Mercator 坐标系转换

由 Mapo 🗺️ 自动生成

file_io/__init__.py

@@ -0,0 +1,4 @@
+# file_io/__init__.py
+from .txt_handler import TxtHandler
+from .excel_handler import ExcelHandler
+from .shp_handler import ShpHandler

file_io/excel_handler.py

@@ -0,0 +1,72 @@
+"""
+Excel/CSV file reader/writer using pandas + openpyxl.
+"""
+import os
+from pathlib import Path
+import pandas as pd
+from core.data_model import CoordData, CoordPoint
+
+
+class ExcelHandler:
+    EXTENSIONS = {".xlsx", ".xls", ".csv"}
+
+    @classmethod
+    def read(cls, path: str, sheet: str | None = None) -> CoordData:
+        ext = Path(path).suffix.lower()
+        if ext == ".csv":
+            df = pd.read_csv(path, encoding="utf-8-sig")
+        else:
+            df = pd.read_excel(path, sheet_name=sheet or 0, dtype_backend="numpy_nullable")
+
+        if df.empty:
+            raise ValueError(f"文件无数据: {path}")
+
+        records = df.to_dict(orient="records")
+        # Convert nan to None
+        for r in records:
+            for k, v in r.items():
+                if pd.isna(v):
+                    r[k] = None
+
+        cols = list(df.columns)
+        x_col, y_col, z_col = cls._find_coord_cols(cols)
+        return CoordData.from_records(records, x_col=x_col, y_col=y_col, z_col=z_col)
+
+    @classmethod
+    def write(cls, data: CoordData, path: str, sheet: str = "坐标数据"):
+        records = data.to_records()
+        if not records:
+            raise ValueError("无数据可写入")
+
+        df = pd.DataFrame(records)
+        ext = Path(path).suffix.lower()
+
+        if ext == ".csv":
+            df.to_csv(path, index=False, encoding="utf-8-sig")
+        else:
+            with pd.ExcelWriter(path, engine="openpyxl") as writer:
+                df.to_excel(writer, sheet_name=sheet, index=False)
+
+    @staticmethod
+    def _find_coord_cols(cols: list[str]) -> tuple[str, str, str | None]:
+        """Same logic as TxtHandler for column discovery."""
+        x_patterns = ["x", "lon", "经度", "lng", "easting"]
+        y_patterns = ["y", "lat", "纬度", "northing"]
+        z_patterns = ["z", "h", "高程", "高度", "alt"]
+
+        x_col = y_col = z_col = None
+        lower_cols = {c: c.lower().replace(" ", "").replace("_", "").replace("-","")
+                      for c in cols}
+
+        for c, lc in lower_cols.items():
+            if any(p in lc for p in x_patterns):
+                x_col = c
+            elif any(p in lc for p in y_patterns):
+                y_col = c
+            elif any(p in lc for p in z_patterns):
+                z_col = c
+
+        if not x_col and len(cols) >= 2:
+            x_col, y_col = cols[0], cols[1]
+
+        return x_col, y_col, z_col

file_io/shp_handler.py

@@ -0,0 +1,53 @@
+"""
+SHP (Shapefile) reader/writer using geopandas.
+"""
+import os
+from pathlib import Path
+import geopandas as gpd
+import pandas as pd
+from shapely.geometry import Point
+from core.data_model import CoordData, CoordPoint
+
+
+class ShpHandler:
+    EXTENSIONS = {".shp"}
+
+    @classmethod
+    def read(cls, path: str) -> CoordData:
+        gdf = gpd.read_file(path, encoding="utf-8")
+        if gdf.empty:
+            raise ValueError(f"Shapefile 无数据: {path}")
+
+        crs = gdf.crs.to_string() if gdf.crs else None
+
+        # Extract centroids for non-Point geometries
+        geom_col = gdf.geometry.name
+        points = []
+        for _, row in gdf.iterrows():
+            geom = row[geom_col]
+            if geom is None or geom.is_empty:
+                continue
+            centroid = geom.centroid if geom.geom_type != "Point" else geom
+            pt = CoordPoint(
+                x=centroid.x,
+                y=centroid.y,
+                z=centroid.z if hasattr(centroid, "z") and centroid.z else None,
+                attrs={k: v for k, v in row.items() if k != geom_col and not pd.isna(v)},
+            )
+            points.append(pt)
+
+        cols = [c for c in gdf.columns if c != geom_col]
+        return CoordData(points=points, crs=crs, columns=cols,
+                         geometry_type=gdf.geom_type.iloc[0] if len(gdf) > 0 else "Point")
+
+    @classmethod
+    def write(cls, data: CoordData, path: str, crs: str | None = None):
+        records = data.to_records()
+        if not records:
+            raise ValueError("无数据可写入")
+
+        # Build geometry column
+        geometry = [Point(r.pop("X"), r.pop("Y")) for r in records]
+
+        gdf = gpd.GeoDataFrame(records, geometry=geometry, crs=crs or data.crs)
+        gdf.to_file(path, encoding="utf-8")

file_io/txt_handler.py

@@ -0,0 +1,131 @@
+"""
+TXT file reader/writer.
+Auto-detects delimiters (space, tab, comma) and column names.
+Handles optional header line and comment lines (#, //).
+"""
+import csv
+import io
+from pathlib import Path
+from core.data_model import CoordData, CoordPoint
+
+
+class TxtHandler:
+    EXTENSIONS = {".txt", ".csv"}
+
+    @classmethod
+    def read(cls, path: str) -> CoordData:
+        path = Path(path)
+        raw = path.read_text(encoding="utf-8-sig")
+
+        # Strip comment lines
+        lines = [l for l in raw.splitlines()
+                 if l.strip() and not l.strip().startswith(("#", "//"))]
+
+        if not lines:
+            raise ValueError(f"文件为空或全为注释: {path}")
+
+        # Detect delimiter
+        delim = cls._detect_delimiter(lines)
+        reader = csv.reader(io.StringIO("\n".join(lines)), delimiter=delim)
+
+        all_rows = list(reader)
+        if not all_rows:
+            raise ValueError(f"无法解析文件: {path}")
+
+        # Check if first row looks like a header (contains non-numeric first field
+        # or known column names)
+        first_row = all_rows[0]
+        has_header = cls._looks_like_header(first_row)
+
+        if has_header:
+            col_names = first_row
+            data_rows = all_rows[1:]
+        else:
+            # No header — auto-generate column names
+            num_cols = len(first_row)
+            col_names = [f"COL{i+1}" for i in range(num_cols)]
+            data_rows = all_rows
+
+        # Detect X, Y, Z columns
+        x_col, y_col, z_col = cls._find_coord_cols(col_names)
+
+        records = []
+        for row in data_rows:
+            record = {}
+            for i, name in enumerate(col_names):
+                val = row[i].strip() if i < len(row) else None
+                record[name] = val
+            records.append(record)
+
+        return CoordData.from_records(records, x_col=x_col, y_col=y_col, z_col=z_col)
+
+    @classmethod
+    def write(cls, data: CoordData, path: str):
+        records = data.to_records()
+        if not records:
+            raise ValueError("无数据可写入")
+
+        delim = cls._detect_delimiter_from_ext(path)
+        fieldnames = list(records[0].keys())
+
+        with open(path, "w", newline="", encoding="utf-8-sig") as f:
+            writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=delim)
+            writer.writeheader()
+            writer.writerows(records)
+
+    @staticmethod
+    def _looks_like_header(row: list[str]) -> bool:
+        """Heuristic: header if first few entries are non-numeric strings."""
+        if not row:
+            return False
+        # If any column name matches known coord patterns, it's a header
+        lower = {c.lower().strip() for c in row[:5]}
+        coord_keywords = {"x", "y", "lon", "lat", "经度", "纬度", "点号", "id"}
+        if lower & coord_keywords:
+            return True
+        # If the first entry is not a valid number, it's probably a header
+        try:
+            float(row[0].strip())
+            return False
+        except (ValueError, AttributeError):
+            return True
+
+    @staticmethod
+    def _detect_delimiter(lines: list[str]) -> str:
+        """Auto-detect: tab > comma > space."""
+        sample = "\n".join(lines[:20])
+        tab_count = sample.count("\t")
+        comma_count = sample.count(",")
+        # Check if commas are between spaces (likely CSV) vs within text
+        if comma_count > 3 or comma_count > tab_count:
+            return ","
+        if tab_count > 0:
+            return "\t"
+        return " "  # default: space-separated
+
+    @staticmethod
+    def _detect_delimiter_from_ext(path: str) -> str:
+        return "," if path.lower().endswith(".csv") else "\t"
+
+    @staticmethod
+    def _find_coord_cols(cols: list[str]) -> tuple[str, str, str | None]:
+        """Find X, Y, Z columns."""
+        x_patterns = ["x", "lon", "经度", "lng", "easting"]
+        y_patterns = ["y", "lat", "纬度", "northing"]
+        z_patterns = ["z", "h", "高程", "高度", "alt", "elevation"]
+
+        x_col = y_col = z_col = None
+        for c in cols:
+            lc = c.lower().replace(" ", "").replace("_", "").replace("-", "")
+            if any(p == lc or lc.startswith(p) or lc.endswith(p) for p in x_patterns):
+                x_col = c
+            elif any(p == lc or lc.startswith(p) or lc.endswith(p) for p in y_patterns):
+                y_col = c
+            elif any(p == lc or lc.startswith(p) or lc.endswith(p) for p in z_patterns):
+                z_col = c
+
+        # Fallback: first two numeric-ish columns if no match
+        if not x_col and len(cols) >= 2:
+            x_col, y_col = cols[0], cols[1]
+
+        return x_col, y_col, z_col