pin.py

import os
import io

import matplotlib
matplotlib.use("Agg")  # 백엔드 Agg로 고정

import geopandas as gpd
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import contextily as ctx
from shapely.geometry import Point
from shapely import wkt as shp_wkt
from PIL import Image

import gradio as gr

# -------------------- 경로 설정 --------------------
SHP_PATH = r"C:\Users\USER\iCloudDrive\인하대학교\5학년 1학기\표본론 및 실습\TeamProject\Data\inha_boundary.shp"
CSV_PATH = r"C:\Users\USER\iCloudDrive\인하대학교\5학년 1학기\표본론 및 실습\TeamProject\데이터 수집\측정결과.csv"

# 이미지 해상도(픽셀)
IMG_W, IMG_H = 1200, 900

# -------------------- 캠퍼스 경계 준비 --------------------
gdf_boundary_4326 = gpd.read_file(SHP_PATH).to_crs(epsg=4326)
gdf_boundary_3857 = gdf_boundary_4326.to_crs(epsg=3857)

# DeprecationWarning 피하기 위해 union_all 우선 사용
try:
    poly_3857 = gdf_boundary_3857.geometry.union_all()
except AttributeError:
    poly_3857 = gdf_boundary_3857.geometry.unary_union

if poly_3857.geom_type == "MultiPolygon":
    poly_3857 = max(poly_3857.geoms, key=lambda g: g.area)

BOUND_MINX, BOUND_MINY, BOUND_MAXX, BOUND_MAXY = poly_3857.bounds

# -------------------- 초기 CSV 로딩 --------------------
# -------------------- 초기 CSV 로딩 --------------------
if os.path.exists(CSV_PATH):
    df_u = pd.read_csv(CSV_PATH, encoding="cp949")

    # 컬럼 정리
    df_u.columns = [c.strip() for c in df_u.columns]
    df_u = df_u.loc[:, ~df_u.columns.str.contains("^Unnamed")]

    if {"lat", "lon", "dB"}.issubset(df_u.columns):
        # 이미 lat/lon/dB 형식이면 그대로 사용
        df_init = df_u[["lat", "lon", "dB"]].copy()
    elif "위치" in df_u.columns and "dB" in df_u.columns:
        # 옛날 형식: 위치(WKT), dB → lat/lon/dB로 변환
        pos_series = df_u["위치"].astype(str).str.strip()

        def safe_load_wkt(s):
            s_up = s.upper()
            if "POINT" not in s_up:
                return None
            try:
                return shp_wkt.loads(s)
            except Exception:
                return None

        geom = pos_series.apply(safe_load_wkt)
        mask = geom.notnull()

        if mask.any():
            geom_valid = geom[mask]
            lons = geom_valid.apply(lambda g: round(g.x, 7))
            lats = geom_valid.apply(lambda g: round(g.y, 7))
            dB_vals = pd.to_numeric(df_u.loc[mask, "dB"], errors="coerce")

            df_init = pd.DataFrame(
                {
                    "lat": lats.values,
                    "lon": lons.values,
                    "dB": dB_vals.values,
                }
            ).dropna(subset=["lat", "lon"])
        else:
            df_init = pd.DataFrame(columns=["lat", "lon", "dB"])
    else:
        # 형식을 알아볼 수 없으면 빈 DF
        df_init = pd.DataFrame(columns=["lat", "lon", "dB"])
else:
    df_init = pd.DataFrame(columns=["lat", "lon", "dB"])

# 타입 보정
for col in ["lat", "lon", "dB"]:
    if col not in df_init.columns:
        df_init[col] = pd.Series(dtype=float)



# -------------------- 지도 이미지 생성 --------------------
def make_map_image(df_points: pd.DataFrame):
    """캠퍼스 경계 + 저장된 포인트들을 그린 PNG 이미지를 numpy 배열로 반환."""
    fig, ax = plt.subplots(
        figsize=(IMG_W / 100, IMG_H / 100), dpi=100
    )  # → 정확히 IMG_W x IMG_H 픽셀

    # 캠퍼스 폴리곤 (테두리만)
    gdf_boundary_3857.plot(
        ax=ax,
        facecolor="none",
        edgecolor="darkblue",
        linewidth=2,
        alpha=1.0,
    )

    # 베이스맵
    ctx.add_basemap(ax, source=ctx.providers.OpenStreetMap.Mapnik, alpha=0.7)

    # 저장된 포인트 찍기
    df_points = df_points.dropna(subset=["lat", "lon"])
    if len(df_points) > 0:
        gdf_pts = gpd.GeoDataFrame(
            df_points.copy(),
            geometry=gpd.points_from_xy(df_points["lon"], df_points["lat"]),
            crs=4326,
        ).to_crs(epsg=3857)

        gdf_pts.plot(
            ax=ax,
            column="dB",
            cmap="viridis",
            markersize=40,
            edgecolor="black",
            linewidth=0.3,
            alpha=0.9,
            legend=False,
            zorder=5,
        )

    # 폴리곤 bounds에 딱 맞게
    ax.set_xlim(BOUND_MINX, BOUND_MAXX)
    ax.set_ylim(BOUND_MINY, BOUND_MAXY)
    ax.axis("off")

    # 여백 제거해서 전체 캔버스를 지도에 맞춤
    plt.subplots_adjust(0, 0, 1, 1)

    # Figure -> PNG bytes -> PIL -> numpy
    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=100)
    buf.seek(0)
    img = np.array(Image.open(buf))

    plt.close(fig)
    return img


# -------------------- 픽셀 → 위경도 변환 --------------------
def pixel_to_latlon(x_px: int, y_px: int):
    """

    Gradio 이미지 클릭 좌표(픽셀, (x, y))를

    EPSG:3857 → EPSG:4326(lat, lon)으로 변환.

    """
    # x: [0, IMG_W] -> [BOUND_MINX, BOUND_MAXX]
    x_3857 = BOUND_MINX + (BOUND_MAXX - BOUND_MINX) * (x_px / IMG_W)
    # y: 위가 0, 아래가 IMG_H → [BOUND_MAXY, BOUND_MINY]로 매핑
    y_3857 = BOUND_MAXY - (BOUND_MAXY - BOUND_MINY) * (y_px / IMG_H)

    g = gpd.GeoSeries([Point(x_3857, y_3857)], crs=3857).to_crs(epsg=4326)
    pt = g.iloc[0]
    lat, lon = pt.y, pt.x

    # 여기서 소수점 7자리로 반올림
    return round(float(lat), 7), round(float(lon), 7)


# -------------------- Gradio 콜백 --------------------
def on_map_click(img, df_state, evt: gr.SelectData):
    """이미지 클릭 시 위경도 텍스트 박스 갱신."""
    if evt is None or evt.index is None:
        return None, None
    x, y = evt.index  # (x_px, y_px)
    lat, lon = pixel_to_latlon(x, y)
    return lat, lon  # 이미 7자리로 반올림된 값


def add_point(lat, lon, db, df_state):
    """lat/lon/dB 추가하고 CSV 저장 + 지도 갱신."""
    if df_state is None or isinstance(df_state, dict):
        df_state = df_init.copy()

    if lat is None or lon is None or db is None:
        msg = "위도/경도/dB 입력 오류."
        return df_state, make_map_image(df_state), df_state, msg

    try:
        lat = float(lat)
        lon = float(lon)
        db = float(db)
    except ValueError:
        msg = "숫자 오류."
        return df_state, make_map_image(df_state), df_state, msg

    # 안전하게 한 번 더 7자리로 맞춤
    lat = round(lat, 7)
    lon = round(lon, 7)

    # 폴리곤 안인지 체크
    pt = gpd.GeoSeries([Point(lon, lat)], crs=4326).to_crs(epsg=3857).iloc[0]
    if not poly_3857.contains(pt):
        msg = "캠퍼스 경계 밖."
        return df_state, make_map_image(df_state), df_state, msg

    new_row = pd.DataFrame([{"lat": lat, "lon": lon, "dB": db}])
    new_df = pd.concat([df_state, new_row], ignore_index=True)

    os.makedirs(os.path.dirname(CSV_PATH), exist_ok=True)
    new_df[["lat", "lon", "dB"]].to_csv(CSV_PATH, index=False, encoding="cp949")

    # 메시지도 7자리로
    msg = f"{len(new_df)}개 점 저장됨 (마지막: lat={lat:.7f}, lon={lon:.7f}, dB={db:.1f})"
    return new_df, make_map_image(new_df), new_df, msg

def load_uploaded_csv(file, df_state):
    """

    CSV 업로드 시:

    - (케이스 A) '위치', 'dB' 컬럼이 있으면: 위치(WKT) -> lat/lon, dB 그대로

    - (케이스 B) 'lat', 'lon', 'dB' 컬럼이 있으면: 그대로 사용

    - 둘 다 아니면 에러

    """
    if df_state is None or isinstance(df_state, dict):
        df_state = df_init.copy()

    if file is None:
        msg = "업로드된 파일 없음."
        return df_state, make_map_image(df_state), df_state, msg

    # CSV 읽기 (인코딩 시도)
    try:
        df_u = pd.read_csv(file.name, encoding="cp949")
    except UnicodeDecodeError:
        df_u = pd.read_csv(file.name, encoding="utf-8")

    # 컬럼 이름 공백/Unnamed 정리
    df_u.columns = [c.strip() for c in df_u.columns]
    df_u = df_u.loc[:, ~df_u.columns.str.contains("^Unnamed")]

    # ---------- 케이스 B: lat/lon/dB 형식 ----------
    if {"lat", "lon", "dB"}.issubset(df_u.columns):
        new_df = df_u[["lat", "lon", "dB"]].copy()
        # 숫자/반올림 보정
        new_df["lat"] = pd.to_numeric(new_df["lat"], errors="coerce").round(7)
        new_df["lon"] = pd.to_numeric(new_df["lon"], errors="coerce").round(7)
        new_df["dB"] = pd.to_numeric(new_df["dB"], errors="coerce")
        new_df = new_df.dropna(subset=["lat", "lon"])

        df_merged = new_df.copy()

        os.makedirs(os.path.dirname(CSV_PATH), exist_ok=True)
        df_merged[["lat", "lon", "dB"]].to_csv(CSV_PATH, index=False, encoding="cp949")

        msg = f"lat/lon/dB 형식 CSV에서 {len(df_merged)}개 점 읽어옴."
        return df_merged, make_map_image(df_merged), df_merged, msg

    # ---------- 케이스 A: 위치(WKT), dB 형식 ----------
    if "위치" in df_u.columns and "dB" in df_u.columns:
        pos_series = df_u["위치"].astype(str).str.strip()

        def safe_load_wkt(s):
            s_up = s.upper()
            if "POINT" not in s_up:
                return None
            try:
                return shp_wkt.loads(s)
            except Exception:
                return None

        geom = pos_series.apply(safe_load_wkt)
        mask = geom.notnull()

        if not mask.any():
            msg = "'위치' 컬럼에 유효한 POINT WKT가 없다."
            return df_state, make_map_image(df_state), df_state, msg

        geom_valid = geom[mask]
        lons = geom_valid.apply(lambda g: round(g.x, 7))
        lats = geom_valid.apply(lambda g: round(g.y, 7))
        dB_vals = pd.to_numeric(df_u.loc[mask, "dB"], errors="coerce")

        new_df = pd.DataFrame(
            {
                "lat": lats.values,
                "lon": lons.values,
                "dB": dB_vals.values,
            }
        ).dropna(subset=["lat", "lon"])

        df_merged = new_df.copy()

        os.makedirs(os.path.dirname(CSV_PATH), exist_ok=True)
        df_merged[["lat", "lon", "dB"]].to_csv(CSV_PATH, index=False, encoding="cp949")

        msg = f"[위치, dB] 형식 CSV에서 유효한 점 {len(df_merged)}개 읽어옴."
        return df_merged, make_map_image(df_merged), df_merged, msg

    # ---------- 둘 다 아니면 ----------
    msg = "CSV 형식을 알아볼 수 없다. (위치,dB 또는 lat,lon,dB 필요)"
    return df_state, make_map_image(df_state), df_state, msg



# -------------------- Gradio 앱 --------------------
def build_app():
    init_img = make_map_image(df_init)

    with gr.Blocks(title="소음 측정 기록기") as demo:
        gr.Markdown(
            "## 인하대 캠퍼스 소음 측정\n"
            "- 지도 이미지 클릭 → 위/경도 자동 입력 → dB 저장\n"
            "- 또는 [위치, dB] CSV 업로드해서 한 번에 보여주기"
        )

        df_state = gr.State(df_init)

        with gr.Row():
            map_img = gr.Image(
                value=init_img,
                type="numpy",
                label=None,
                interactive=True,
            )

            with gr.Column():
                lat_box = gr.Number(label="위도 (lat)", interactive=False)
                lon_box = gr.Number(label="경도 (lon)", interactive=False)
                db_box = gr.Number(label="데시벨 (dB)")

                add_btn = gr.Button("현재 좌표 추가 + CSV 저장", variant="primary")
                csv_upload = gr.File(
                    label="CSV 업로드 ([위치, dB] 형식)",
                    file_types=[".csv"],
                )

                status = gr.Markdown()

                table = gr.Dataframe(
                    value=df_init,
                    headers=["lat", "lon", "dB"],
                    label="저장된 측정값",
                    interactive=False,
                    wrap=True,
                )

        # 이미지 클릭 시: lat/lon 갱신
        map_img.select(
            fn=on_map_click,
            inputs=[map_img, df_state],
            outputs=[lat_box, lon_box],
        )

        # 버튼 클릭 시: DF/지도/상태 갱신 + CSV 저장
        add_btn.click(
            fn=add_point,
            inputs=[lat_box, lon_box, db_box, df_state],
            outputs=[table, map_img, df_state, status],
        )

        # CSV 업로드 시: 업로드 데이터로 지도/테이블/상태 갱신
        csv_upload.change(
            fn=load_uploaded_csv,
            inputs=[csv_upload, df_state],
            outputs=[table, map_img, df_state, status],
        )

    return demo


if __name__ == "__main__":
    app = build_app()
    app.launch(share=True)