mayihua
e9f87fc693
TXT / Excel(CSV) / SHP 三种格式任意互转
支持 WGS84 / CGCS2000 / Xian80 / Beijing54 / Web Mercator 坐标系转换
由 Mapo 🗺️ 自动生成
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""
|
||
Conversion workflow orchestrator.
|
||
Detects input format, reads, transforms CRS, writes output.
|
||
"""
|
||
import os
|
||
from pathlib import Path
|
||
from core.data_model import CoordData
|
||
from core.coord_transform import CoordTransformer
|
||
from file_io.txt_handler import TxtHandler
|
||
from file_io.excel_handler import ExcelHandler
|
||
from file_io.shp_handler import ShpHandler
|
||
|
||
# Registry: {extension_lowercase: handler_class}
|
||
HANDLERS = {}
|
||
for h in (TxtHandler, ExcelHandler, ShpHandler):
|
||
for ext in h.EXTENSIONS:
|
||
HANDLERS[ext] = h
|
||
|
||
|
||
def get_handler(path: str):
|
||
ext = Path(path).suffix.lower()
|
||
h = HANDLERS.get(ext)
|
||
if not h:
|
||
raise ValueError(f"不支持的文件格式: {ext}（支持: {', '.join(sorted(HANDLERS))}）")
|
||
return h
|
||
|
||
|
||
def resolve_format(path: str, fmt_hint: str) -> str:
|
||
"""Determine output format. fmt_hint='auto' → infer from path ext."""
|
||
if fmt_hint and fmt_hint.lower() != "auto":
|
||
return fmt_hint.lower()
|
||
ext = Path(path).suffix.lower()
|
||
ext_map = {".txt": "txt", ".csv": "csv", ".xlsx": "xlsx", ".xls": "xlsx", ".shp": "shp"}
|
||
f = ext_map.get(ext)
|
||
if not f:
|
||
raise ValueError(f"无法从扩展名推断输出格式: {ext}")
|
||
return f
|
||
|
||
|
||
def format_to_ext(fmt: str) -> str:
|
||
mapping = {"txt": ".txt", "csv": ".csv", "xlsx": ".xlsx", "shp": ".shp"}
|
||
return mapping.get(fmt, f".{fmt}")
|
||
|
||
|
||
class ConversionWorkflow:
|
||
"""Orchestrate read → transform → write."""
|
||
|
||
def run(self, input_path: str, output_path: str,
|
||
from_crs: str | None = None, to_crs: str | None = None,
|
||
output_format: str = "auto") -> dict:
|
||
# 1. Read
|
||
in_handler = get_handler(input_path)
|
||
data = in_handler.read(input_path)
|
||
|
||
# 2. Determine source CRS
|
||
src_crs = from_crs or data.crs
|
||
tgt_crs = to_crs
|
||
|
||
# 3. Transform if needed
|
||
if src_crs and tgt_crs and src_crs != tgt_crs:
|
||
transformer = CoordTransformer(src_crs, tgt_crs)
|
||
xs = [p.x for p in data.points]
|
||
ys = [p.y for p in data.points]
|
||
new_xs, new_ys = transformer.transform_batch(xs, ys)
|
||
for i, p in enumerate(data.points):
|
||
p.x = round(new_xs[i], 6)
|
||
p.y = round(new_ys[i], 6)
|
||
data.crs = tgt_crs
|
||
crs_note = transformer.description
|
||
else:
|
||
crs_note = src_crs or "未指定"
|
||
|
||
# 4. Write
|
||
out_fmt = resolve_format(output_path, output_format)
|
||
out_ext = format_to_ext(out_fmt)
|
||
# Ensure output path has the right extension
|
||
final_path = str(Path(output_path).with_suffix(out_ext))
|
||
|
||
out_handler = get_handler(final_path)
|
||
out_handler.write(data, final_path)
|
||
|
||
return {
|
||
"success": True,
|
||
"input": input_path,
|
||
"output": final_path,
|
||
"input_format": Path(input_path).suffix.lower(),
|
||
"output_format": out_fmt,
|
||
"crs": crs_note,
|
||
"feature_count": data.count,
|
||
}
|