Привет, Хабр! Сегодня хочу поделиться интересным проектом, который мы сделали для конкурса. Задача — превратить сырые GPX-треки (треки с GPS-устройств) в структурированные данные с визуализацией, метеорологической и географической аналитикой. Всё это — на Python, с использованием открытых API и библиотек для работы с геоданными.

Что делает проект?

Код представляет собой пайплайн обработки GPS-треков, который:

Скачивает GPX-файлы по ссылкам
Визуализирует треки на карте
Извлекает данные о каждой точке (координаты, время, высота)
Добавляет погодные данные (температуру) в момент записи трека
Определяет регион и тип местности
Рассчитывает частоту шагов (для пеших походов)
Аугментирует изображения треков для расширения датасета
Визуализирует распределения данных

Загрузка и парсинг GPX

import requests
import pandas as pd
import os

def download_gpx(links):
    """
    Скачивает GPX-файлы по списку ссылок
    """
    for num, url in enumerate(links):
        try:
            response = requests.get(url)
            filename = f"track{num}.gpx"
            
            with open(f"data/gpx/{filename}", mode="wb") as f:
                f.write(response.content)
        except Exception as e:
            print("Ошибка при скачивании")

# Инициализация DataFrame для хранения всех точек
df = pd.DataFrame(columns=["track_id", "track_time", "latitude", "longitude", "altitude"])

Визуализация треков на картах

import gpxpy
import geopandas as gpd
import contextily as ctx
from shapely.geometry import box, LineString
import matplotlib.pyplot as plt

def gpx_to_png(df):
    """
    Конвертирует GPX-файлы в изображения карт с треками
    """
    margin = 0.02
    img_path = "data/image"
    os.makedirs(img_path, exist_ok=True)

    for i in os.listdir("data/gpx"):
        png_name = f"{i[:-4]}.png"
        with open(f"data/gpx/{i}", encoding="UTF-8") as f:
            gpx = gpxpy.parse(f)
        lats, lons = [], []
        for track in gpx.tracks:
            for segment in track.segments:
                for point in segment.points:
                    lats.append(point.latitude)
                    lons.append(point.longitude)
                    df.loc[len(df)] = [i, point.time, point.latitude, point.longitude, point.elevation]

        # Создаем bounding box с отступами
        bbox = box(
                min(lons) - margin,
                min(lats) - margin,
                max(lons) + margin,
                max(lats) + margin
            )
        
        track_line = LineString(zip(lons, lats))

        # Конвертируем в Web Mercator для отображения
        gdf_bbox = gpd.GeoDataFrame(geometry=[bbox], crs="EPSG:4326")
        gdf_bbox_web = gdf_bbox.to_crs(epsg=3857)

        gdf_track = gpd.GeoDataFrame(geometry=[track_line], crs="EPSG:4326")
        gdf_track_web = gdf_track.to_crs(epsg=3857)

        # Создаем карту
        fig, ax = plt.subplots(figsize=(10, 8))
        gdf_bbox_web.plot(ax=ax, alpha=0)
        gdf_track_web.plot(ax=ax, color="red", linewidth=2)

        ctx.add_basemap(ax, crs=gdf_bbox_web.crs, source=ctx.providers.OpenStreetMap.Mapnik)

        ax.set_axis_off()
        plt.savefig(f"{img_path}/{png_name}", dpi=150, bbox_inches="tight", pad_inches=0)
        plt.close(fig)

    return df

Получение исторических погодных данных

def temp(lat, lon, date):
    """
    Получает температуру для заданных координат и даты
    """
    url = "https://archive-api.open-meteo.com/v1/archive"

    params = {
        'latitude': lat,        # Широта
        'longitude': lon,       # Долгота
        'start_date': date,
        'end_date': date,
        'hourly': 'temperature_2m',
        "timezone":"auto"
    }
    headers = {'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:89.0) Gecko/20100101 Firefox/89.0'}

    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()
    if response.status_code == 200:
        data = response.json()
        return data["hourly"]['temperature_2m'][12]  # Температура в полдень
    else:
        print(f"Данные не получены")

def analysis_weather(df):
    '''
    Функция для интерполяции температур между опорными точками.
    Использует 5 опорных точек для интерполяции температуры для всех строк.
    '''
    n = len(df)
    key_indexes = [0, n//4, n//2, 3*n//4, n-1]
    temperatures_at_key_points = {}
    
    for idx in key_indexes:
        lat = df.iloc[idx]["latitude"]
        lon = df.iloc[idx]["longitude"]
        date = df.iloc[idx]["track_time"]
        
        temp_value = temp(lat, lon, date)
        
        temperatures_at_key_points[idx] = temp_value

    
    if len(temperatures_at_key_points) < 2:
        print("Недостаточно данных для интерполяции, проверьте работоспособность API")
        df["temperature"] = None
        return df
    
    all_temperatures = []
    left_idx = 0
    right_idx = 1
    
    for i in range(n):
        if (right_idx < len(key_indexes) - 1 and i >= key_indexes[right_idx]):
            left_idx += 1
            right_idx += 1
        
        left_key = key_indexes[left_idx]
        right_key = key_indexes[right_idx]
        
        left_temp = temperatures_at_key_points[left_key]
        right_temp = temperatures_at_key_points[right_key]
        
        if i in temperatures_at_key_points:
            temperature = temperatures_at_key_points[i]
        elif left_temp is not None and right_temp is not None:
            temperature = left_temp + (right_temp - left_temp) * (i - left_key) / (right_key - left_key)
        else:
            temperature = left_temp if left_temp is not None else right_temp
        
        all_temperatures.append(temperature)
    
    df = df.copy()
    df["temperature"] = all_temperatures
    
    return df

def get_temp(df):
    """
    Обрабатывает температуру для всех треков
    """
    try:
        df_temp = pd.DataFrame()
        for i in range(0, 3):
            track_data = df[df["track_id"] == f"track{i}.gpx"]
            track_data_weather = analysis_weather(track_data)
            df_temp = pd.concat([df_temp, track_data_weather])
            print(f"track{i} добавлен")
        df = df_temp.copy()
        return df
    except Exception as e:
        print(f"Ошибка вызова функции: analysis_weather {e}")

Географический анализ: регион и тип местности

import time

def extract_map_region(lat: float, lon: float):
    """
    Определяет регион по координатам через Nominatim API
    """
    try:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/536.36 (KHTML, like Gecko) Chrome/58.0.3029.100 Safari/536.3'}
        response = requests.get(f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json", headers=headers)
        json = response.json()
        time.sleep(1.5)  # Уважаем лимиты API
        if "county" in json["address"]:
            return json["address"]["county"]
        if "state" in json["address"]:
            return json["address"]["state"]
        return json["address"]["country"]
    except Exception as e:
        print(f"Error {e}")

def analysis_region(df):
    '''
    Функция для определения регионов между опорными точками.
    '''
    try:
        lat = df.iloc[0]["latitude"]
        lon = df.iloc[0]["longitude"]
        df = df.copy()
        df["region"] = extract_map_region(lat, lon)
        return df
    except Exception as e:
        print(f"Ошибка вызова функции extract_map_region {e}")

def terrain_type(df):
    """
    Определяет тип местности и ключевые объекты через Overpass API
    """
    overpass_endpoints = [
    "https://overpass-api.de/api/interpreter",
    "https://overpass.kumi.systems/api/interpreter",
    "https://overpass.openstreetmap.ru/cgi/interpreter"
    ]
    try:
        points = list(zip(df["latitude"], df["longitude"]))
        rep_points = [points[0], points[len(points)//2], points[-1]] # 3 опорные точки
        
        all_landuse, all_natural, all_key_objects = [], [], set()
        
        for lat, lon in rep_points:
            overpass_query = f"""
            [out:json][timeout:45];
            (
                way(around:500,{lat},{lon})["landuse"];
                way(around:500,{lat},{lon})["natural"];
                way(around:500,{lat},{lon})["leisure"];
                way(around:500,{lat},{lon})["waterway"="river"];
                way(around:500,{lat},{lon})["waterway"="stream"];
                way(around:500,{lat},{lon})["natural"="water"];
                node(around:500,{lat},{lon})["place"="city"];
                node(around:500,{lat},{lon})["place"="town"];
                node(around:500,{lat},{lon})["place"="village"];
                node(around:500,{lat},{lon})["natural"="peak"];
                node(around:500,{lat},{lon})["natural"="mountain"];
            );
            out tags center;
            """
            
            for endpoint in overpass_endpoints:
                response = requests.get(endpoint, params={'data': overpass_query}, timeout=60)
                if response.status_code == 200:
                    data = response.json()
                    break
            
            for element in data.get('elements', []):
                tags = element.get('tags', {})
                
                if 'landuse' in tags:
                    all_landuse.append(tags['landuse'])
                elif 'natural' in tags:
                    all_natural.append(tags['natural'])
                    if tags['natural'] in ['peak', 'mountain'] and 'name' in tags:
                        all_key_objects.add(f"Mountain: {tags['name']}")
                
                if 'waterway' in tags and tags['waterway'] in ['river', 'stream'] and 'name' in tags:
                    all_key_objects.add(f"River: {tags['name']}")
                
                if 'place' in tags and tags['place'] in ['city', 'town', 'village'] and 'name' in tags:
                    all_key_objects.add(f"Settlement: {tags['name']} ({tags['place']})")
                
                if element.get('type') == 'way' and 'natural' in tags and tags['natural'] == 'water' and 'name' in tags:
                    all_key_objects.add(f"Lake: {tags['name']}")
            
            time.sleep(1.5)
        
        terrain_type = "unknown"
        if all_landuse:
            terrain_type = max(set(all_landuse), key=all_landuse.count)
        elif all_natural:
            terrain_type = max(set(all_natural), key=all_natural.count)
        
        key_objects_str = "; ".join(sorted(all_key_objects)) if all_key_objects else None
        
        df["terrain_type"] = terrain_type
        df["key_objects_str"] = key_objects_str
    except Exception as e:
        print("Ошибка", e)
    return df

Расчет физических параметров

from geopy.distance import geodesic

def step_frequency(df):
    """
    Рассчитывает частоту шагов на основе расстояния между точками
    """
    points = list(zip(df["latitude"], df["longitude"]))
    step = [0]
    
    for p1, p2 in zip(points, points[1:]):
        dist = geodesic(p1, p2).meters
        step.append(dist / 0.75)  # Предполагаем среднюю длину шага 0.75 метра
    df["steps"] = step
    return df

Аугментация изображений

from PIL import Image, ImageEnhance
from random import randint, uniform

def data_augmentation():
    """
    Создает аугментированные версии изображений треков
    """
    images_path = "data/image"
    os.makedirs(images_path, exist_ok=True)
    
    for filename in os.listdir(images_path):
        if filename.lower().endswith(".png") and not any(word in filename for word in ["rotated", "contrasted", "brightness"]):
            img_path = os.path.join(images_path, filename)
            img = Image.open(img_path)

            base_name = os.path.splitext(filename)[0]

            # Поворот
            rotated_img = img.rotate(randint(10, 60))
            rotated_img.save(os.path.join(images_path, f"{base_name}_rotated.png"))

            # Изменение контраста
            contrasted_img = ImageEnhance.Contrast(img).enhance(randint(2, 4))
            contrasted_img.save(os.path.join(images_path, f"{base_name}_contrasted.png"))

            # Изменение яркости
            brightness_img = ImageEnhance.Brightness(img).enhance(uniform(1.2, 1.6))
            brightness_img.save(os.path.join(images_path, f"{base_name}_brightness.png"))

Визуализация распределений

import seaborn as sns

def norm_or_not(df):
    """
    Визуализирует распределения всех числовых колонок
    """
    n = len(df.columns) // 5 + bool(len(df.columns) % 5)
    _, axes = plt.subplots(nrows=n, ncols=5, figsize=(15,20))
    
    for idx, i in enumerate(df.columns):
        sns.kdeplot(data=df, x=i, common_norm=False, ax=axes[idx // 5][idx%5])
        plt.title(i) 
    plt.show()

Полный пайплайн обработки

# Пример использования всего пайплайна
def main():
    # 1. Скачиваем треки
    links = [
        "https://example.com/track1.gpx",
        "https://example.com/track2.gpx", 
        "https://example.com/track3.gpx"
    ]
    download_gpx(links)
    
    # 2. Конвертируем в изображения и DataFrame
    df = gpx_to_png(df)
    
    # 3. Форматируем даты
    df['track_time'] = df['track_time'].dt.strftime('%Y-%m-%d')
    
    # 4. Получаем погодные данные
    df = get_temp(df)
    
    # 5. Определяем регион
    df = analysis_region(df)
    
    # 6. Рассчитываем частоту шагов
    df = step_frequency(df)
    
    # 7. Определяем тип местности
    df = terrain_type(df)
    
    # 8. Визуализируем распределения
    norm_or_not(df)
    
    # 9. Аугментируем изображения
    data_augmentation()
    
    print("Обработка завершена!")
    print(f"Обработано {len(df)} точек")
    print(f"Сохранено треков: {len(os.listdir('data/gpx'))}")
    
    return df

if __name__ == "__main__":
    result_df = main()

Заключение

Этот проект демонстрирует, как можно превратить сырые GPS-данные в богатый источник аналитики. Комбинация географического, метеорологического и статистического анализа позволяет извлечь максимум информации из, казалось бы, простых треков.

Код модульный и легко расширяемый — можно добавлять новые источники данных, метрики и виды анализа. И все это — на открытых данных и бесплатных API.

Требования: Python 3.8+, установка зависимостей:

Важно: Все используемые API бесплатны для некоммерческого использования, но имеют лимиты запросов.

Автоматический анализ GPX-треков: от трека до аналитики