# -*- coding: utf-8 -*-”””无人机照片 -> 点位图 + 飞行台账依赖:pip install pillow piexif openpyxl (可选 shp:pip install pyshp)用法: 真实:python drone_log.py -i ”照片文件夹” -o ”成果文件夹” 自测:python drone_log.py --demo”””import os, csv, json, math, argparse, datetimefrom PIL import Imageimport piexifimport openpyxldef _ratio_to_float(rational): ”””piexif 读出来是 (分子, 分母) 元组,转浮点。””” num, den = rational return 0.0 if den == 0 else num / float(den)def dms_to_decimal(dms, ref): ”””度分秒 + 参考字母 -> 十进制;南纬/西经取负。””” degrees = _ratio_to_float(dms[0]) minutes = _ratio_to_float(dms[1]) seconds = _ratio_to_float(dms[2]) decimal = degrees + minutes / 60.0 + seconds / 3600.0 ref = ref.decode(”ascii”).upper() if isinstance(ref, bytes) else str(ref).upper() return -decimal if ref in (”S”, ”W”) else decimaldef _parse_exif_datetime(exif_ifd): raw = exif_ifd.get(piexif.ExifIFD.DateTimeOriginal) or exif_ifd.get(piexif.ExifIFD.DateTime) if not raw: return None text = raw.decode(”ascii”) if isinstance(raw, bytes) else str(raw) try: return datetime.datetime.strptime(text, ”%Y:%m:%d %H:%M:%S”) except ValueError: return Nonedef read_photo(path): ”””读单张照片 GPS + 拍摄时间,返回 dict。””” rec = {”file”: os.path.basename(path), ”lat”: None, ”lon”: None, ”alt”: None, ”dt”: None, ”has_gps”: False} try: exif = piexif.load(path) except Exception: return rec gps = exif.get(”GPS”, {}) exif_ifd = exif.get(”Exif”, {}) if piexif.GPSIFD.GPSLatitude in gps and piexif.GPSIFD.GPSLongitude in gps: try: rec[”lat”] = dms_to_decimal(gps[piexif.GPSIFD.GPSLatitude], gps.get(piexif.GPSIFD.GPSLatitudeRef, b”N”)) rec[”lon”] = dms_to_decimal(gps[piexif.GPSIFD.GPSLongitude], gps.get(piexif.GPSIFD.GPSLongitudeRef, b”E”)) rec[”has_gps”] = True alt = gps.get(piexif.GPSIFD.GPSAltitude) if alt: rec[”alt”] = _ratio_to_float(alt) except Exception: rec[”has_gps”] = False rec[”dt”] = _parse_exif_datetime(exif_ifd) return recdef cluster_flights(photos, gap_minutes=30): ”””按拍摄时间聚类成架次。””” valid = [p for p in photos if p[”has_gps”] and p[”dt”]] valid.sort(key=lambda p: p[”dt”]) flights, current, last_dt = [], [], None for p in valid: if last_dt is None or (p[”dt”] - last_dt).total_seconds() <= gap_minutes * 60: current.append(p) else: flights.append(current); current = [p] last_dt = p[”dt”] if current: flights.append(current) return flightsdef _safe(v): return 0.0 if v is None else vdef write_kml(photos, out): L = ['<?xml version=”1.0” encoding=”UTF-8”?>', '无人机照片点位'] for p in photos: if not p[”has_gps”]: continue L.append(' %s%f,%f,%f' % (p[”file”], _safe(p[”lon”]), _safe(p[”lat”]), _safe(p[”alt”]))) L.append(””) open(out, ”w”, encoding=”utf-8”).write(”\n”.join(L))def write_geojson(photos, out): feats = [] for p in photos: if not p[”has_gps”]: continue feats.append({”type”: ”Feature”, ”properties”: {”file”: p[”file”], ”alt”: p[”alt”], ”time”: p[”dt”].isoformat() if p[”dt”] else None}, ”geometry”: {”type”: ”Point”, ”coordinates”: [_safe(p[”lon”]), _safe(p[”lat”]), _safe(p[”alt”])]}}) json.dump({”type”: ”FeatureCollection”, ”features”: feats}, open(out, ”w”, encoding=”utf-8”), ensure_ascii=False, indent=2)def write_csv(photos, out): with open(out, ”w”, encoding=”utf-8-sig”, newline=””) as f: w = csv.writer(f) w.writerow([”文件名”, ”经度”, ”纬度”, ”海拔(米)”, ”拍摄时间”]) for p in photos: if not p[”has_gps”]: continue w.writerow([p[”file”], ”%.6f” % p[”lon”], ”%.6f” % p[”lat”], (”%.1f” % p[”alt”]) if p[”alt”] is not None else ””, p[”dt”].strftime(”%Y-%m-%d %H:%M:%S”) if p[”dt”] else ””])def flights_summary(flights): rows = [] for i, fl in enumerate(flights, 1): lats = [p[”lat”] for p in fl]; lons = [p[”lon”] for p in fl] alts = [p[”alt”] for p in fl if p[”alt”] is not None] dts = [p[”dt”] for p in fl if p[”dt”]] rows.append({”flight”: i, ”count”: len(fl), ”start”: min(dts).strftime(”%H:%M:%S”) if dts else ””, ”end”: max(dts).strftime(”%H:%M:%S”) if dts else ””, ”center_lat”: sum(lats)/len(lats), ”center_lon”: sum(lons)/len(lons), ”span_lat”: max(lats)-min(lats), ”span_lon”: max(lons)-min(lons), ”alt_min”: min(alts) if alts else None, ”alt_max”: max(alts) if alts else None}) return rowsdef write_excel(flights, out): wb = openpyxl.Workbook(); ws = wb.active; ws.title = ”飞行台账” ws.append([”架次”, ”照片张数”, ”开始时间”, ”结束时间”, ”中心纬度”, ”中心经度”, ”纬度跨度(度)”, ”经度跨度(度)”, ”最低海拔(米)”, ”最高海拔(米)”]) for r in flights_summary(flights): ws.append([r[”flight”], r[”count”], r[”start”], r[”end”], round(r[”center_lat”], 6), round(r[”center_lon”], 6), round(r[”span_lat”], 6), round(r[”span_lon”], 6), round(r[”alt_min”], 1) if r[”alt_min”] is not None else ””, round(r[”alt_max”], 1) if r[”alt_max”] is not None else ””]) ws2 = wb.create_sheet(”全部点位”) ws2.append([”文件名”, ”经度”, ”纬度”, ”海拔(米)”, ”拍摄时间”]) for fl in flights: for p in fl: ws2.append([p[”file”], round(p[”lon”], 6), round(p[”lat”], 6), round(p[”alt”], 1) if p[”alt”] is not None else ””, p[”dt”].strftime(”%Y-%m-%d %H:%M:%S”) if p[”dt”] else ””]) wb.save(out)def process_folder(input_dir, out_dir, gap_minutes=30): os.makedirs(out_dir, exist_ok=True) exts = (”.jpg”, ”.jpeg”, ”.JPG”, ”.JPEG”) files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.lower().endswith(exts)] print(”扫描到 %d 张照片(按扩展名筛选)” % len(files)) photos = [read_photo(f) for f in files] flights = cluster_flights(photos, gap_minutes=gap_minutes) anomalies = [p[”file”] for p in photos if (not p[”has_gps”]) or not (-90 <= _safe(p[”lat”]) <= 90) or not (-180 <= _safe(p[”lon”]) <= 180)] with open(os.path.join(out_dir, ”异常清单.txt”), ”w”, encoding=”utf-8”) as f: if anomalies: f.write(”以下 %d 张照片无 GPS 或坐标异常,未进入点位图:\n” % len(anomalies)) for a in anomalies: f.write(” - %s\n” % a) else: f.write(”未发现无 GPS / 坐标异常照片。\n”) valid = sum(1 for p in photos if p[”has_gps”]) write_kml(photos, os.path.join(out_dir, ”points.kml”)) write_geojson(photos, os.path.join(out_dir, ”points.geojson”)) write_csv(photos, os.path.join(out_dir, ”points.csv”)) write_excel(flights, os.path.join(out_dir, ”飞行台账.xlsx”)) print(”有效带 GPS 照片:%d 张,聚成 %d 个架次” % (valid, len(flights))) for r in flights_summary(flights): print(” 架次%d:%d 张,%s~%s,中心 (%.5f, %.5f)” % (r[”flight”], r[”count”], r[”start”], r[”end”], r[”center_lat”], r[”center_lon”])) print(”成果已写入:%s” % out_dir)def _to_dms(coord): deg = int(coord); mf = (coord - deg) * 60.0; minute = int(mf) sec = (mf - minute) * 60.0 return ((deg, 1), (minute, 1), (int(round(sec * 100)), 100))def _make_demo_photo(path, lat, lon, alt, dt): img = Image.new(”RGB”, (16, 16), (120, 160, 200)) gps_ifd = {piexif.GPSIFD.GPSLatitudeRef: b”N”, piexif.GPSIFD.GPSLatitude: _to_dms(abs(lat)), piexif.GPSIFD.GPSLongitudeRef: b”E”, piexif.GPSIFD.GPSLongitude: _to_dms(abs(lon)), piexif.GPSIFD.GPSAltitude: (int(alt*100), 100), piexif.GPSIFD.GPSAltitudeRef: 0} exif_ifd = {piexif.ExifIFD.DateTimeOriginal: dt.strftime(”%Y:%m:%d %H:%M:%S”).encode()} img.save(path, exif=piexif.dump({”GPS”: gps_ifd, ”Exif”: exif_ifd}))def run_demo(): dd = os.path.join(os.getcwd(), ”demo_photos”); od = os.path.join(os.getcwd(), ”demo_output”) os.makedirs(dd, exist_ok=True) base = datetime.datetime(2026, 8, 16, 10, 0, 0) for i in range(7): _make_demo_photo(os.path.join(dd, ”DJI_%03d.JPG” % i), 30.60+i*0.0002, 114.30+i*0.0003, 120+i, base+datetime.timedelta(minutes=i)) for i in range(5): _make_demo_photo(os.path.join(dd, ”DJI_%03d.JPG” % (i+10)), 30.61+i*0.0002, 114.31+i*0.0003, 130+i, base+datetime.timedelta(hours=1, minutes=i)) process_folder(dd, od, gap_minutes=30)if __name__ == ”__main__”: ap = argparse.ArgumentParser() ap.add_argument(”-i”, ”--input”); ap.add_argument(”-o”, ”--output”) ap.add_argument(”-g”, ”--gap”, type=int, default=30); ap.add_argument(”--demo”, action=”store_true”) a = ap.parse_args() if a.demo: run_demo() elif a.input and a.output: process_folder(a.input, a.output, gap_minutes=a.gap) else: print('真实:python drone_log.py -i ”照片文件夹” -o ”成果文件夹”') print(”演示:python drone_log.py --demo”)