구현하고자 하는 시스템 흐름도
1단계: 사용자 요청 (Client Request)
- 프론트엔드(React)에서 사용자가 권역(예: Honam, Seoul)을 선택합니다.
- 앱은 사용자의 현재 위치(GPS 위도/경도)를 감지하여 권역 정보와 함께 서버(FastAPI)로 전송합니다.
2단계: 대상 병원 1차 필터링 (DB Query, Oracle)
- 서버는 전송받은 위치 정보를 이용해 Oracle DB(HOSPITAL_MASTER)를 조회합니다.
- DB 내부의 공간 함수(Oracle Spatial)를 사용하여 내 위치와 병원 간의 직선거리를 계산하고, 가장 가까운 상위 10개 병원만 필터링하여 추출합니다.
3단계: 실시간 경로 탐색 (External API Call)
- 서버는 1차로 선별된 10개의 병원 리스트에 대해서만 네이버 지도(Naver Maps API)에 경로 계산을 요청합니다.
- 입력: 내 위치(출발지) ↔ 병원 위치(목적지)
- 출력: 실제 차량 이동 시 예상되는 최단 거리(km)와 소요 시간(분)을 받아옵니다. (API 호출 횟수 최소화)
4단계: 최종 결과 반환 (Response)
- 병원 기본 정보에 API로 받은 정밀 경로 정보(거리, 시간)를 결합합니다.
- 최종적으로 사용자가 보기 편하도록 소요 시간순으로 정렬된 JSON 데이터를 프론트엔드로 응답합니다.
Naver API 신청
네이버 클라우드 플랫폼(ncloud.com)에 접속합니다. 회원가입 후 로그인을 합니다. 결제 수단(카드)을 등록해야 발급할 수 있습니다.
서비스 > Application Services > Map를 선택하고 해당 페이지로 접속해 이용신청을 클릭합니다.
콘솔에 접속했다면 좌측 메뉴 리스트에서 Maps를 클릭해 Application으로 들어갑니다.
프로젝트에 사용할 Application 이름과 API를 선택합니다.
해당 프로젝트에서는 Directions 5를 사용합니다. 자세한 API 명세서는 홈페이지에서 확인할 수 있습니다.
정보를 잘 입력하고 등록을 한 뒤 Subscription으로 돌아오면 상품이 이용 중이라고 뜹니다.
등록이 완료되면 Application으로 돌아와 API 관리에서 인증정보를 클릭하여 인증키를 확인합니다.
아래의 인증 정보를 .env 파일에 붙여넣으면 됩니다.
# Naver Maps API
NAVER_MAP_CLIENT_ID="Client_ID"
NAVER_MAP_CLIENT_SECRET="Client_Secret"
VScode 코드 작성
기존의 폴더 구조는 다음과 같습니다.
routers/navigation.py 와 utils/region_mapping.py와 services/map_service.py 를 추가하고
main.py와 db_service.py를 수정해줍니다.
siren-backend/
├── .env
├── requirements.txt
├── main.py # 앱 실행 및 라우터 등록, CORS 설정 등 프로젝트의 중앙 제어 장치
├── database.py
├── schemas.py
├── routers/
│ ├── notification.py
│ ├── analysis.py
│ └── navigation.py # 사용자의 GPS 위치를 기반으로 최적의 병원을 추천하는 통로
├── utils/
│ └── region_mapping.py # 프론트엔드 권역 이름을 DB 쿼리용 행정구역 리스트로 변환하는 도구
└── services/
├── ws_manager.py
├── audio_service.py
├── map_service.py # 네이버 지도 API와 통신하여 실시간 주행 거리/시간을 계산하는 모듈
└── db_service.py # Oracle DB에 접근하여 병원 정보를 조회하거나 환자 로그를 저장하는 처리기
db_service.py
가장 먼저 db_service.py 코드입니다.
해당 코드는 Oracle DB의 지리 정보 기능(Spatial)을 활용해 내 위치 주변의 병원을 효율적으로 찾는 코드입니다.
# services/db_service.py
import oracledb
# 권역별 병원 정보 DB에서 조회
def get_hospitals_by_province(provinces_list: list, my_lat: float, my_lng: float):
"""
Oracle Spatial 기능(SDO_GEOM.SDO_DISTANCE)을 사용하여
내 위치와 병원 간의 거리를 계산하고, 가까운 순으로 상위 10개를 반환합니다.
"""
from database import get_db_connection
connection = get_db_connection()
cursor = connection.cursor()
try:
# ---------------------------------------------------------
# [Oracle Spatial Query 설명]
# 1. SDO_GEOMETRY(2001, 4326, SDO_POINT_TYPE(X, Y, NULL), NULL, NULL)
# - 2001: 2차원 점(Point)을 의미
# - 4326: WGS84 좌표계 (GPS 표준)
# - SDO_POINT_TYPE: 좌표값 입력 (순서 중요: 경도 X, 위도 Y)
#
# 2. SDO_GEOM.SDO_DISTANCE(점1, 점2, 오차범위, 단위)
# - 두 지점 사이의 정확한 거리를 계산하여 반환
# ---------------------------------------------------------
sql = f"""
SELECT HPID, HOSPITAL_NM, PROVINCE, X_POS, Y_POS, TELNO,
SDO_GEOM.SDO_DISTANCE(
SDO_GEOMETRY(2001, 4326, SDO_POINT_TYPE(X_POS, Y_POS, NULL), NULL, NULL),
SDO_GEOMETRY(2001, 4326, SDO_POINT_TYPE({my_lng}, {my_lat}, NULL), NULL, NULL),
0.01,
'unit=KM'
) AS DISTANCE_KM
FROM HOSPITAL_MASTER
"""
# 지역 필터링 (리스트가 있을 때만 WHERE 절 추가)
if provinces_list:
province_str = "','".join(provinces_list)
sql += f" WHERE PROVINCE IN ('{province_str}')"
# 거리순 정렬 후 상위 10개 추출
sql += " ORDER BY DISTANCE_KM ASC FETCH FIRST 10 ROWS ONLY"
cursor.execute(sql)
# 결과 매핑 (컬럼명 소문자로 변환)
columns = [col[0].lower() for col in cursor.description]
hospitals = [dict(zip(columns, row)) for row in cursor.fetchall()]
return hospitals
except Exception as e:
print(f"❌ DB 조회 실패 (Oracle Spatial): {e}")
return []
finally:
if cursor: cursor.close()
if connection: connection.close()
oracle 데이터베이스와 통신하기 위한 드라이버를 가져옵니다.
인자로 provinces_list(검색할 지역 목록)과 위도, 경도를 받아오는 get_hospitals_by_province 함수를 선언합니다.
해당 함수는 이후에 navigation.py 에서 병원 리스트를 가져올 때 사용합니다.
import oracledb
def get_hospitals_by_province(provinces_list: list, my_lat: float, my_lng: float):
Oracle Spatial의 기능인 SDO_GEOM.SDO_DISTANCE( ) 함수를 사용하여 내 위치와 병원 간의 두 점 사이의 거리를 구합니다.
- 2001: 2차원 점(Point)이라는 뜻입니다.
- 4326: GPS 좌표계(WGS84)를 쓴다는 국제 표준 코드입니다.
나머지 인자에 자신의 위치와 병원의 위치를 넣고 결과를 KM로 받습니다.
try:
sql = f"""
SELECT HPID, HOSPITAL_NM, PROVINCE, X_POS, Y_POS, TELNO,
SDO_GEOM.SDO_DISTANCE(
SDO_GEOMETRY(2001, 4326, SDO_POINT_TYPE(X_POS, Y_POS, NULL), NULL, NULL),
SDO_GEOMETRY(2001, 4326, SDO_POINT_TYPE({my_lng}, {my_lat}, NULL), NULL, NULL),
0.01,
'unit=KM'
) AS DISTANCE_KM
FROM HOSPITAL_MASTER
"""
동적 쿼리는 SQL 문장이 고정되어 있지 않고, 상황(조건)에 따라 살아 움직이듯 변하는 쿼리를 말합니다.
만약에 provinces_list에 넘어온 권역이 여러 개라면('Honam', 'Seoul') 이를 python의 문자열 합치기 함수 join( )을 사용하여 ' 와 ' 사이에 콤마를 넣어줍니다.
if provinces_list:
province_str = "','".join(provinces_list)
sql += f" WHERE PROVINCE IN ('{province_str}')"
계산된 거리(DISTANCE_KM)로 정렬을 한 뒤 상위 10개만 잘라서 가져옵니다.
sql += " ORDER BY DISTANCE_KM ASC FETCH FIRST 10 ROWS ONLY"
SQL을 실행하고 컬럼명을 소문자로 바꾼 뒤 hospitals 변수에 딕셔너리 형태로 넣습니다.
cursor.fetchall( ) 에서 가져온 값을 columns와 row에 담고 튜플 형태로 짝지어줍니다(zip). 그리고 JSON 형태의 파일로 만들기 위해 dict( )을 사용합니다.
hospitals을 반환해줍니다.
cursor.execute(sql)
columns = [col[0].lower() for col in cursor.description]
hospitals = [dict(zip(columns, row)) for row in cursor.fetchall()]
map_service.py
가장 먼저 map_service.py 코드입니다.
해당 코드는 네이버 지도 서버에서 출발지부터 목적지까지 가는 길을 물어보고, 그 결과를 정리해서 가져오는 역할을 합니다.
import requests
import os
from dotenv import load_dotenv
load_dotenv()
def calculate_route_to_hospital(start_lat, start_lng, goal_lat, goal_lng, goal_name):
client_id = os.getenv("NAVER_MAP_CLIENT_ID")
client_secret = os.getenv("NAVER_MAP_CLIENT_SECRET")
url = "https://maps.apigw.ntruss.com/map-direction/v1/driving"
headers = {
"X-NCP-APIGW-API-KEY-ID": client_id,
"X-NCP-APIGW-API-KEY": client_secret,
}
params = {
"start": f"{start_lng},{start_lat}",
"goal": f"{goal_lng},{goal_lat}",
"option": "trafast"
}
# [로그] 요청 정보 출력
print(f"\n🗺️ [Naver Map] 경로 계산 시작")
print(f" 출발지 : {start_lat}, {start_lng}")
print(f" 목적지: {goal_name} ({goal_lat}, {goal_lng})")
try:
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
data = response.json()
if data['code'] == 0:
summary = data['route']['trafast'][0]['summary']
result = {
"hospital_name": goal_name,
"distance_km": round(summary['distance'] / 1000, 2),
"duration_min": int(summary['duration'] / 1000 / 60),
# 프론트엔드 마커 표시용 좌표 반환
"goal_lat": goal_lat,
"goal_lng": goal_lng
}
# [로그] 성공 결과 출력
print(f"✅ [Naver Map] 계산 성공!")
print(f" 📏 거리: {result['distance_km']}km")
print(f" ⏳ 예상 소요시간: {result['duration_min']}분")
return result
else:
print(f"⚠️ [Naver Map] API 응답 오류 (Code: {data['code']})")
else:
print(f"❌ [Naver Map] HTTP 에러 (Status: {response.status_code})")
print(f" 응답내용: {response.text}") # 에러 원인 파악을 위해 출력
return None
except Exception as e:
print(f"❌ [Naver Map] 예외 발생: {e}")
return None
requests는 파이썬에서 인터넷 브라우저 역할을 하는 도구로 네이버 서버에 URL을 치고 들어가는 역할을 해줍니다.
import requests
내 위치의 위도와 경도, 병원의 위도와 경도 마지막으로 병원 이름을 인자로 주는 calculate_route_to_hospital 함수를 선언합니다.
이후 navigation.py 에서 내 기준과 병원 위치를 naver api로 보내는 역할을 합니다.
def calculate_route_to_hospital(start_lat, start_lng, goal_lat, goal_lng, goal_name):
처음에 발급했던 Naver API 키 정보를 아래와 같이 입력합니다.
client_id = os.getenv("NAVER_MAP_CLIENT_ID")
client_secret = os.getenv("NAVER_MAP_CLIENT_SECRET")
url = "https://maps.apigw.ntruss.com/map-direction/v1/driving"
아래의 코드는 서버에 들어갈 때 사용할 헤더입니다.
API 명세서를 확인하면 헤더에 대한 설명을 확인할 수 있습니다.
headers = {
"X-NCP-APIGW-API-KEY-ID": client_id,
"X-NCP-APIGW-API-KEY": client_secret,
}
다음으로는 GET 요청을 할 때 사용할 파라미터입니다.
파라미터 또한 API 명세서를 확인하여 사용합니다.
params = {
"start": f"{start_lng},{start_lat}",
"goal": f"{goal_lng},{goal_lat}",
"option": "trafast"
}
| 필드 (Field) | 타입 (Type) | 필수 여부 | 설명 (Description) |
| start | String | 필수 | 출발지 좌표 (경도,위도) 예시: 127.12345,37.12345 |
| goal | String | 필수 | 목적지 좌표 (경도,위도) :로 구분하여 최대 10개 입력 가능 (가장 가까운 곳 1개 조회) 예시: 123.45,34.56:124.56,35.67 |
| waypoints | String | 선택 | 경유지 좌표 (경도,위도) |
| option | String | 선택 | 경로 탐색 옵션 (:로 구분 최대 3개) • traoptimal: 실시간 최적 (기본값) • trafast: 실시간 빠른 길 • tracomfort: 실시간 편한 길 • traavoidtoll: 무료 도로 우선 • traavoidcaronly: 자동차 전용 제외 |
| cartype | Integer | 선택 | 차종 (톨게이트비 계산용) • 1: 1종 소형/승용 (기본값) • 2: 2종 중형 • 3: 3종 대형 • 4: 4종 화물 • 5: 5종 특수 • 6: 1종 경차 |
| fueltype | String | 선택 | 연료 (유류비 계산용) • gasoline: 휘발유 (기본값) • diesel: 경유 • lpg: LPG • highgradegasoline: 고급 휘발유 |
| mileage | Double | 선택 | 연비 (km/L) 차량의 연비 직접 설정 (기본값: 14.0) |
| lang | String | 선택 | 응답 결과 언어 • ko: 한국어 (기본값) • en: 영어 • ja: 일본어 • zh: 중국어 간체 |
위에서 작성한 url과 헤더, 파라미터를 인자로 하여 네이버 서버에 데이터를 보내고 응답을 기다립니다.
response = requests.get(url, headers=headers, params=params)
summary는 네이버 API가 보내준 response(응답)입니다.
summary의 distance는 미터 단위로 km로 바꿔주고 duration은 ms초 단위이므로 min으로 바꿔줍니다.
summary = data['route']['trafast'][0]['summary']
result = {
"hospital_name": goal_name,
"distance_km": round(summary['distance'] / 1000, 2),
"duration_min": int(summary['duration'] / 1000 / 60),
"goal_lat": goal_lat,
"goal_lng": goal_lng
}
navigation.py
가장 먼저 navigation.py 코드입니다.
해당 코드는 사용자의 요청을 받아서 DB와 네이버 지도를 연결해주고 최종 결과를 정리해서 돌려주는 코드입니다.
# routers/navigation.py
from fastapi import APIRouter, HTTPException
from schemas import NavigationRequest
from services.db_service import get_hospitals_by_province
from utils.region_mapping import get_provinces_by_region
from services.map_service import calculate_route_to_hospital
import time
# router 변수 생성 (app 대신 사용)
router = APIRouter(
prefix="/api/v1/navigation", # 주소 앞에 자동으로 붙음
tags=["Navigation"] # 문서(Swagger) 자동 정리용
)
@router.post("")
async def recommend_hospitals(request: NavigationRequest):
# 시작 시간 측정
start_time = time.time()
print(f"[요청 수신] 지역 {request.region} / 내 위치 : {request.start_lat}, {request.start_lng}")
# 지역 코드 변환
target_provinces = get_provinces_by_region(request.region)
# DB에서 병원 리스트 쿼리
hopitals = get_hospitals_by_province(
target_provinces,
request.start_lat,
request.start_lng
)
final_results = []
print(f"🏥 {len(hopitals)}개 병원 후보 조회 완료. 경로 계산 시작...")
# 모든 병원에 대해 네이버 API 호출
for idx, hosp in enumerate(hopitals):
if not hosp.get("x_pos") or not hosp.get("y_pos"):
print(f"⚠️ 병원 좌표 정보 누락: {hosp.get('hospital_nm')}")
continue
# Oracle DB에서 가져온 값은 문자열일 수 있으므로 float 변환
try:
h_lat = float(hosp["y_pos"])
h_lng = float(hosp["x_pos"])
except ValueError:
print(f"⚠️ 병원 좌표 변환 오류: {hosp.get('hospital_nm')} - X_POS: {hosp['x_pos']}, Y_POS: {hosp['y_pos']}")
continue
# 네이버 API 호출
route_info = calculate_route_to_hospital(
start_lat=request.start_lat,
start_lng=request.start_lng,
goal_lat=h_lat,
goal_lng=h_lng,
goal_name=hosp["hospital_nm"]
)
if route_info:
# 프론트엔드로 보내기 위해 DB 추가 정보 병합
route_info['hpid'] = hosp['hpid']
route_info['telno'] = hosp['telno']
final_results.append(route_info)
# 가장 빠른 병원이 위로가게 정렬
final_results.sort(key=lambda x: x['duration_min'])
# 소요 시간 계산
total_duration = time.time() - start_time
print(f"✅ 경로 계산 완료 : {total_duration:.2f}초")
return {
"status": "success",
"region": request.region,
"data": final_results
}
main.py에서 사용할 라우터를 설정합니다.
router = APIRouter(
prefix="/api/v1/navigation",
tags=["Navigation"]
)
POST 방식의 요청이 들어오면 비동기 함수인 recommend_hospitals를 실행합니다.
여기서 들어오는 인자인 request는 schemas.py에서 정해둔 규칙을 따르는지 자동으로 검사합니다.
@router.post("")
async def recommend_hospitals(request: NavigationRequest):
요청한 권역을 target_provinces에 저장합니다.
target_provinces = get_provinces_by_region(request.region)
바뀐 권역 명과 나의 위도 경도를 get_hospitals_by_province의 인자로 하여 호출하고 hospitals에 저장합니다.
여기서 상위 10개의 병원이 hospitals에 저장됩니다.
hopitals = get_hospitals_by_province(
target_provinces,
request.start_lat,
request.start_lng
)
저장된 10개의 병원으로 차례대로 하나씩 calculate_route_to_hospital의 인자로 주어 route_info에 저장합니다.
for idx, hosp in enumerate(hopitals):
# 네이버 API 호출
route_info = calculate_route_to_hospital(
start_lat=request.start_lat,
start_lng=request.start_lng,
goal_lat=h_lat,
goal_lng=h_lng,
goal_name=hosp["hospital_nm"]
)
거리와 시간 정보가 담긴 route_info에 병원 ID와 전화번호를 합칩니다.
if route_info:
route_info['hpid'] = hosp['hpid']
route_info['telno'] = hosp['telno']
final_results.append(route_info)
final_results에 담긴 병원까지의 소요 시간을 기준으로 정렬을 합니다.
final_results.sort(key=lambda x: x['duration_min'])region_mapping.py
다음으로 region_mapping.py 코드입니다.
이 코드는 "프론트엔드의 권역명(영어)"를 "데이터베이스에 등록된 정보"로 변환해 주는 '통역사' 역할을 합니다.
#utils/region_mapping.py
def get_provinces_by_region(region_code: str):
mapping = {
"Seoul": ["서울특별시"],
"Gyeonggi": ["경기도", "인천광역시"],
"Honam": ["광주광역시", "전라남도", "전라북도"],
"Yeongnam": ["부산광역시", "대구광역시", "울산광역시", "경상남도", "경상북도"],
"Chungcheong": ["대전광역시", "세종특별자치시", "충청남도", "충청북도"],
"Gangwon": ["강원도"],
"All": ["서울특별시", "인천광역시", "경기도", "강원도", "대전광역시", "세종특별자치시", "충청남도", "충청북도",
"광주광역시", "전라남도", "전라북도", "부산광역시", "대구광역시", "울산광역시", "경상남도", "경상북도", "제주특별자치도"]
}
return mapping.get(region_code, [])schemas.py
다음으로 schemas.py 코드입니다.
이 코드는 프론트엔드(React)에서 서버(FastAPI)로 데이터를 보낼 때, "반드시 이 양식을 지켜서 보내라!" 라고 강제하는 주문서 양식입니다.
#schemas.py
from pydantic import BaseModel
class NavigationRequest(BaseModel):
region: str # Seoul, Honam 등
start_lat: float # GPS 위치 위도
start_lng: float # GPS 위치 경도검색 시간 단축
처음 계획했던 대로 권역별로 전체의 병원을 탐색하면 “Honam” 호남을 기준 50개를 탐색하고 66.11초가 소요됩니다. 시간을 단축하기 위한 2가지 방법을 찾아보았습니다. 첫 번째는 앞서 사용한 Oracle의 Spatial 기능입니다. Spatial 기능을 사용하여 가까운 병원 10개를 추리면 탐색 속도가 10.51초로 단축됩니다.
1. Oracle 데이터베이스의 역할 : "지리적 위치 기반 1차 선별"
첫 번째 단계는 방대한 병원 데이터 중에서 사용자에게 물리적으로 유의미한 후보군을 추려내는 작업입니다.
- 수학적 거리 계산: 단순히 저장된 데이터를 조회하는 것이 아니라, 사용자의 현재 좌표(위도/경도)와 DB에 저장된 각 병원의 좌표를 사용하여 직선 거리(반경)를 데이터베이스 내부에서 직접 계산합니다.
- 데이터 범위 축소: 전체 병원 리스트를 모두 서버로 가져오는 것은 불필요한 네트워크 부하를 일으킵니다. 따라서 DB 단계에서 거리순으로 데이터를 정렬한 뒤, 가장 가까운 최상위 10개 병원만 추출하여 서버로 전달합니다.
- 효과: 이를 통해 서버가 처리해야 할 데이터의 모수 자체를 줄여, 불필요한 연산 리소스를 낭비하지 않도록 합니다.
Oracle의 Spatial 말고 다른 DB에서 제공하는 거리 기반 쿼리 함수도 있습니다.
| 기술 | 파이썬 라이브러리 | FastAPI 사용 방식 | 특징 |
| Redis (Geo) | redis-py | r.geosearch(...) 함수 호출 | 가장 추천. 별도 쿼리 없이 함수 하나로 가장 빠른 거리 계산 가능. |
| Oracle (Spatial) | oracledb | cursor.execute("SELECT ... FROM ... WHERE SDO_NN(...)") | 이미 연결되어 있으니 SQL만 수정하면 바로 사용 가능. (추가 설치 불필요) |
| PostgreSQL (PostGIS) | asyncpg 또는 psycopg2 | session.execute(select(...).order_by(Geometry.distance(...))) | 공간 데이터의 세계 표준. 기능이 제일 막강함. |
또한 시간을 더 단축하고 싶다면 비동기 병렬처리고 또 하나의 방법이 될 수 있습니다.
2. Python 서버의 역할 : "비동기 병렬 처리를 통한 정밀 분석"
두 번째 단계는 선별된 10개의 병원에 대해 실제 주행 경로(네비게이션)를 계산하는 작업입니다.
- 동시 요청 처리 (Concurrency): 기존의 순차적 방식(1번 병원 계산 후 2번 병원 계산)은 병원 수가 늘어날수록 대기 시간이 정비례하여 늘어나는 단점이 있습니다. 이를 개선하기 위해 비동기(Asynchronous) 통신을 도입했습니다.
- 프로세스 가속: 서버는 선별된 10개의 병원에 대한 네이버 지도 API 요청을 동시에 전송합니다. 즉, 10개의 요청이 병렬적으로 수행되므로, 전체 소요 시간은 10번을 더한 시간이 아니라 가장 오래 걸리는 1번의 응답 시간과 거의 동일해집니다.
- 효과: 사용자 입장에서는 복잡한 경로 계산이 포함되어 있음에도 불구하고, 마치 1개의 데이터를 조회하는 것과 같은 빠른 응답 속도를 경험할 수 있습니다.
'FastAPI' 카테고리의 다른 글
| [FastAPI] Nginx 사용하기 (0) | 2025.12.18 |
|---|---|
| [FastAPI] Azure Service Bus와 Function App을 활용한 데이터 전송 (2) (0) | 2025.12.18 |
| [FastAPI] Azure Service Bus와 Function App을 활용한 데이터 전송 (1) (0) | 2025.12.18 |
| [FastAPI] websocket을 사용한 통신 (0) | 2025.12.17 |
| [FastAPI] Azure OpenAI를 활용한 오디오 분석 (0) | 2025.12.16 |