Spaces:
Sleeping
Sleeping
Download pin.py from Louppian/Sampling_Proj: direct link, hf CLI and curl.
- Browser
- Download file 13.2 kB
-
https://huggingface.co/spaces/Louppian/Sampling_Proj/resolve/main/pin.py
- Command line
-
hf download hf://spaces/Louppian/Sampling_Proj/pin.py
-
curl -L -o pin.py https://huggingface.co/spaces/Louppian/Sampling_Proj/resolve/main/pin.py
13.2 kB
| 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) | |