Python/GPS 파일 읽고 속도와 경로를 지도에 그리기

GPS 신호를 추적해 경로를 저장하고 속도를 표시하는 것은 운동 앱들에겐 흔한 기능이다.

하지만 GPS 기록은 종종 오류를 포함하기 때문에 황당한 속도가 나오기도 한다.

그래서 GPS 기록을 검토하면서 보정하고 그 결과물을 지도에 그리는 프로그램을 만들었다.

아래 그래프는 GPS 파일의 속도를 분석한 것이다. 튀는 데이터가 두 개 보인다. 이것을 제거하고 지도에 그린다.

결과물은 아래와 같다.

속도의 범례를 포함하고 속도 변화에 따라 경로의 색을 표시한다.

728x90

필요한 라이브러리를 임포트한다.

my_package는 나만의 유틸리티 모음이다. 자신만의 유틸리티 라이브러리를 만드는 법은 여기에 설명되어 있다.

import folium
import pandas as pd
import gpxpy
import matplotlib.pyplot as plt
import os
from geopy.distance import geodesic
import branca.colormap as cm
from my_package import my_utils as u

main()

• gpx 파싱 또는 지도 그리기를 선택한다.
• 지도 그리기는 이미 만들어진 기존 csv를 지도 데이터로 사용한다.
• gpx processing은 새로운 파일을 처리한다.

def main():
    task = input(
        'Choose task\n(Enter) for new gpx processing, or (D)raw map with existing csv: '
    )

    if task.lower() == 'd':
        draw_path()
    else:
        df, path, f_name = gpx_to_df()
        draw_path(df, path, f_name)

GPX 파일을 Pandas dataframe으로 저장하는 gpx_to_df()

• 경로와 gpx 파일을 얻고 파일이름을 변수에 저장한다.
• gpxpy로 gpx 파일을 파싱한다.
• 좌표와 시간을 pandas dataframe으로 저장한다.
• dataframe을 csv로 저장한다.
• dataframe, 경로, 파일이름을 반환한다.

def gpx_to_df():
    # get gpx file
    path = u.get_path()
    files = u.list_files_with_extension(path, '.gpx')
    f_gps = u.select_item_from_list(files)
    f_name, _ = os.path.splitext(f_gps)

    # read gpx
    gpx_file = os.path.join(path, f_gps)
    with open(gpx_file, 'r') as f:
        gpx = gpxpy.parse(f)

    coordinates, times = get_route(gpx)

    # save dataframe
    df = pd.DataFrame({
        'latitude': [cord[0] for cord in coordinates],
        'longitude': [cord[1] for cord in coordinates],
        'time': [time for time in times]
    })

    df = calculate_time_diffence_distance_speed(df)

    # save csv file
    fp = os.path.join(path, f'{f_name}.csv')
    df.to_csv(fp, index=False)

    return df, path, f_name

몇 개의 서브 프로그램들이 있다.

경로를 반환하는 get_path()

경로를 입력받거나, 입력이 없으면 기본 경로를 반환한다.

기본 경로는 미리 지정된 경로이다.

def get_path():
    path = input('Enter path(No input for default path): ')
    if path == '':
        path = '/my/default/path'

    return path

특정 확장자의 파일을 리스트로 저장하는 list_files_with_extension()

디렉토리와 확장자를 인수로 받아 디렉토리 내의 특정 확장자를 가진 파일을 리스트로 반환한다.

def list_files_with_extension(directory, extension):
    files_with_extension = []
    for file in os.listdir(directory):
        full_path = os.path.join(directory, file)

        if os.path.isfile(full_path) and file.endswith(extension):
            files_with_extension.append(full_path)

    return files_with_extension

리스트에서 특정 파일을 선택하는 select_item_from_list()

리스트를 화면에 프린트하고 인덱스 번호로 파일을 선택하도록 한다.

리스트 길이가 1일 경우(파일이 하나인 경우) 선택없이 파일 이름을 반환한다.

def select_item_from_list(source_list, input_string=None):
    if len(source_list) == 1:
        item = source_list[0]
    else:
        if input_string == None:
            input_string = 'Enter index number: '

        for i in range(len(source_list)):
            print(f'{i}. {source_list[i]}')
        idx = int(input(input_string))

        item = source_list[idx]

    return item

gpx 데이터에서 좌표와 시간을 추출하는 get_gps_data()

gpx 데이터에서 좌표와 시간을 추출한다.

def get_gps_data(gpx):
    # get route
    coordinates = []
    times = []
    for track in gpx.tracks:
        for segment in track.segments:
            for point in segment.points:
                coordinates.append((point.latitude, point.longitude))
                times.append(point.time)

    return coordinates, times

시간차이와 거리, 속도를 계산하는 calculate_time_diffence_distance_speed()

좌표 간 시간차이와 거리를 얻고 속도를 계산한다.

속도는 km/h로 환산하여 저장한다.

def calculate_time_diffence_distance_speed(df):
    # calculate time difference & distance
    df['time_diff'] = df['time'].diff().dt.total_seconds().fillna(0)
    df['distance'] = df.apply(lambda row: geodesic(
        (row['latitude'], row['longitude']),
        (df.iloc[row.name - 1]['latitude'], df.iloc[row.name - 1]['longitude'])
    ).meters if row.name != 0 else 0, axis=1)

    # get speed, m/s -> km/h
    df['speed'] = round(df['distance'] / df['time_diff'] * 3.6, 2)

    return df

지도를 그리는 draw_path()

• dataframe, 디렉토리, 파일이름을 인수로 전달받지 못할 경우에는 기존 파일을 검색해서 dataframe으로 저장한 뒤 작업을 한다.
• 최대속도를 검토한다.
• 속도에 따른 색상을 저장한다.
• csv 파일로 저장한다.
• 최대속도를 화면에 출력하고 지도를 그린다.

def draw_path(df=None, path=None, f_name=None):
    if df is None:
        path = u.get_path()
        files = u.list_files_with_extension(path, '.csv')
        f_csv = u.select_item_from_list(files)
        f_name, _ = os.path.splitext(f_csv)

        fp = os.path.join(path, f_csv)
        df = pd.read_csv(fp)

    # check speed
    df['speed'] = df['speed'].fillna(0)

    max_speed_check = input('Max speed check: y/n: ')
    if max_speed_check.lower() == 'y':
        process_max_values(df, 'speed')

    colors = u.get_colors_by_numbers(df['speed'].tolist())
    df['color'] = colors

    fp = os.path.join(path, f'{f_name}.csv') # type: ignore
    df.to_csv(fp, index=False)

    print(f'Max speed: {df["speed"].max()}')
    get_map(df, path, f_name)

최대속도를 검토하는 process_max_values()

속도 차트를 출력해서 검토한다.

위의 차트처럼 gps는 종종 오류를 일으키는데, 두 개의 기록에 오류가 있는 것으로 보인다.

오류가 있는 값은 0으로 바꾼다.

def process_max_values(df, column):
    while True:
        plot_speed(df, column)

        # get max value in column
        max_value = df[column].max()

        if max_value == 0:
            print('No more max value.')
            break

        user_input = input(
            f'Max speed {max_value}, (C)hange to 0, Any key to stop max value check: '
        )

        if user_input.lower() == 'c':
            # change max value to 0
            df.loc[df[column] == max_value, column] = 0
            print(f'Changed {max_value} to 0.')
        else:
            break

차트를 그리는 plot_speed()

dataframe과 컬럼이름을 받아 컬럼의 그래프를 그린다.

def plot_speed(df, column):
    _, ax = plt.subplots()
    ax.plot(df[column])
    plt.show()

색상 리스트를 반환하는 get_colors_by_numbers()

숫자 리스트에서 최소값은 노란색, 최대값은 빨간색, 나머지는 크기에 따라 순차적으로 g값을 변경 후 색상 리스트를 반환한다.

def get_colors_by_numbers(number_list: list):
    number_max = max(number_list)
    number_min = min(number_list)

    colors = []
    for num in number_list:
        rgb_g = int(255 * (number_max - num) / (number_max - number_min))

        rgb = [255, rgb_g, 0]
        color = rgb_to_hex(rgb)

        colors.append(color)

    return colors

RGB 색상을 HEX 값으로 변경하는 rgb_to_hex()

(R, G, B)로 된 리스트를 받아 hex값으로 변경 후 반환한다.

def rgb_to_hex(rgb):
    return '#{:02x}{:02x}{:02x}'.format(rgb[0], rgb[1], rgb[2])

지도를 그리는 get_map()

• folium의 PolyLine을 이용해서 지도를 그린다. 좌표 간의 색깔을 입힌다.
• branca의 colormap을 이용해서 범례를 추가한다. 범례를 표시하기 위해 색상을 고유값으로 정리한 리스트를 만든다.
• 지도를 html 파일로 저장한다.

def get_map(df, path, f_name):
    # folium 지도 생성
    start_coords = (df.loc[0, 'latitude'], df.loc[0, 'longitude'])

    m = folium.Map(location=start_coords, zoom_start=14) # type: ignore

    # 경로 추가
    for i in range(1, len(df)):
        location = [
            [df.loc[i-1, 'latitude'], df.loc[i-1, 'longitude']],
            [df.loc[i, 'latitude'], df.loc[i, 'longitude']]
        ]

        folium.PolyLine(
            locations=location,
            color=df.loc[i, 'color'],
            weight=5
        ).add_to(m)

    colors = df['color'].tolist()

    colors = u.list_remove_duplicates_sort(colors, True)

    colormap = cm.LinearColormap(
        colors=colors,
        vmin=df['speed'].min(),
        vmax=df['speed'].max(),
        caption='km/h'
    )

    colormap.add_to(m)

    fp = os.path.join(path, f'{f_name}.html')
    m.save(fp)

리스트의 중복항목을 제거하는 list_remove_duplicates_sort()

리스트의 중복항목을 제거하고 정렬시킨다.

def list_remove_duplicates_sort(input_list, reverse=False):
    unique_items = list(set(input_list))
    unique_items.sort(reverse=reverse)

    return unique_items