Sampling_Proj / pin.py
Louppian's picture
Upload folder using huggingface_hub
6b09cbd verified
Raw History Blame Contribute Delete
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)