최적의 순찰경로 추천 시스템 및 위험도 모니터링 시스템(위험도 산출)

순찰 경로 추천 격자가 반환되었다면, 각 격자별로 얼마나 위험한지를 평가하여 위험도를 산출하는 알고리즘을 개발하겠습니다.

위험도 산출 알고리즘

# Handling
import numpy as np
import json

# Moduel 
from selectDatasetFluid import *


def extract_optimized_path_data(optimal_path):
    # 1 데이터 프레임 리스트 
    dataframes = []
    
    # 2. 서울 시 격자 순위 데이터 로드
    data = get_dataframe_from_database_fluid('an_rank')
    
    # 3. 리스트 컴프리헨션을 이용한 ID 추출
    ids_only = [item[2] for item in optimal_path]
    
    # 4. 반복문을 이용한 KEY 데이터 프레임 추출
    for key in ids_only:
        
        # 4.1 최적 경로 격자 데이터 추출
        selected_data = data[data['ID']==key]
        
        # 4.2 선택된 격자 데이터 리스트 추가
        dataframes.append(selected_data)
    
    # 5. 데이터 프레임화 시행   
    concat = pd.concat(dataframes)
    
    # 6.중복 제거
    check_columns=['LON','LAT']
    concat = concat.drop_duplicates(subset=check_columns)
    
    return concat,data
def convert_jason(data_tuples):
#!/usr/bin/env python
# coding: utf-8


def extract_optimized_path_data(optimal_path):
    # 1 데이터 프레임 리스트 
    dataframes = []
    
    # 2. 서울 시 격자 순위 데이터 로드
    data = get_dataframe_from_database_fluid('an_rank')
    
    # 3. 리스트 컴프리헨션을 이용한 ID 추출
    ids_only = [item[2] for item in optimal_path]
    
    # 4. 반복문을 이용한 KEY 데이터 프레임 추출
    for key in ids_only:
        
        # 4.1 최적 경로 격자 데이터 추출
        selected_data = data[data['ID']==key]
        
        # 4.2 선택된 격자 데이터 리스트 추가
        dataframes.append(selected_data)
    
    
    # 5. 데이터 프레임화 시행   
    concat = pd.concat(dataframes)
    
    # 6.중복 제거
    check_columns=['LON','LAT']
    concat = concat.drop_duplicates(subset=check_columns)
    
    return concat,data

def calculate_risk_scores(optimal_path):
    """
    주어진 경로 데이터를 바탕으로 안전 관련 변수들에 대한 위험 점수를 계산합니다.
    위험도 점수는 데이터의 상대적인 안전도를 나타내는 정규화된 점수로 계산됩니다.
    
    Parameters:
    optimal_path (str): 분석할 최적 경로 데이터의 경로 또는 식별자입니다.
    
    Returns:
    DataFrame: 계산된 위험도 점수가 포함된 데이터프레임을 반환합니다.
    """
    
    try:
        # 데이터 추출
        data, rank_data = extract_optimized_path_data(optimal_path)
        
        # 데이터 유효성 검증
        required_columns = ['RANK','EMERGENCY_BELL_AND_DISTANCE', 'SAFETY_CENTER_AND_DISTANCE', 
                            'GRID_SHELTER_DISTANCE_SCORE', 'GRID_FACILITIES_DISTANCE_SCORE', 'NUMBER_OF_CCTV']
        if not all(column in rank_data.columns for column in required_columns):
            raise ValueError("입력 데이터가 필요한 컬럼을 모두 포함하고 있지 않습니다.")
        
        # 변수 범위 계산
        def calculate_range(column):
            return rank_data[column].max() - rank_data[column].min()
        
        ranges = {column: calculate_range(column) for column in required_columns}
        
        
        
        sorted_rank = data[1:-1].sort_values(by='RANK',ascending=False)[-5:].sort_values(by='RANK')
        
        departure = data.iloc[[0]]
        arrival = data.iloc[[-1]]
        
        # 출발지, 데이터 프레임, 도착지를 결합
        high_ranking_data = pd.concat([departure, sorted_rank , arrival])
        
        # 위험도 점수 계산
        for column in required_columns:
            if ranges[column] == 0:  # 분모가 0인 경우 방지
                high_ranking_data[f'{column}_SCORE'] = 100
            else:
                high_ranking_data[f'{column}_SCORE'] = ((high_ranking_data[column] - rank_data[column].min()) / ranges[column] * 100
                ).round(2)
        
        # Rank Score 정제
        high_ranking_data['RANK_SCORE'] = (100 - high_ranking_data['RANK_SCORE'])
        
        # 필요한 컬럼만 선택
        score_columns = [f'{column}_SCORE' for column in required_columns]
        result_data = high_ranking_data[['ID','LON','LAT'] + score_columns]
        
        return result_data
    
    except Exception as e:
        print(f"에러 발생: {e}")
        return pd.DataFrame()  # 에러 시 빈 데이터프레임 반환


def throw_tuple_price(data):
    
    # 각 행을 튜플로 변환하여 리스트에 추가
    tuples_list = [tuple(row) for row in data.itertuples(index=False, name=None)]
    
    return tuples_list

def convert_jason(data_tuples):

    # 튜플의 각 요소에 대한 키 목록
    keys = ['id', 'longitude', 'latitude', 'rank_score', 'emergency_bell_and_distance_score', 'safety_center_and_distacne_score',
            'grid_shelter_distance_score', 'grid_facilities_distance_score', 'number_of_cctv_score']

    # 튜플 데이터를 JSON 객체로 변환
    json_data = [dict(zip(keys, tuple_data)) for tuple_data in data_tuples]
    
    return json_data

def calculate_group_mean(data):
    
    slice_data = data.iloc[1:-1]
    
    # 각 변수의 평균 
    required_columns=['RANK_SCORE','EMERGENCY_BELL_AND_DISTANCE_SCORE','SAFETY_CENTER_AND_DISTANCE_SCORE',
                      'GRID_SHELTER_DISTANCE_SCORE_SCORE','GRID_FACILITIES_DISTANCE_SCORE_SCORE','NUMBER_OF_CCTV_SCORE']
    
    mean_bowl = [slice_data[col].mean() for col in required_columns]
    
    # 튜플의 각 요소에 대한 키 목록
    keys =  ['rank_score_mean', 'emergency_bell_and_distance_score_mean', 'safety_center_and_distacne_score_mean',
            'grid_shelter_distance_score_mean', 'grid_facilities_distance_score_mean', 'number_of_cctv_score_mean']
    
    # 튜플 데이터를 JSON 객체로 변환
    json_mean_data = [dict(zip(keys, mean_bowl))]
    
    return json_mean_data

def combine_dictionaries(dica,dicb):
    # 현재 딕셔너리에 평균 딕셔너리 추가
    dica.append(dicb)
    
    return dica

 

최적의 순찰 경로의 데이터 반환 → 순찰 경로의 위험도 산출(위험도는 각 격자의 안전센터, CCTV, 안전 시설물, 인구밀도 등의 점수를 이용하여 산출)  → 산출된 데이터를 순위가 높은 데이터 5개를 선정하여  JSON 형태로 반환

 

시스템 적용

 

 

 

위의 서울 시 이미지에서 확인할 수 있듯이, 순찰 추천 경로의 보라색 지점이 순위가 높은 5개의 데이터가 반환되었습니다. 그리고 JSON 파일로 가 격자의 위험도 점수에 대한 값을 반환합니다.