Рассматриваемый код на Python анализирует данные о движении транспорта, собранные с дорожного сенсора, и выполняет четыре задачи: анализ данных, расчет параметров трафика, определение зон обнаружения и кластеризацию. Давайте разберем, что он делает, как работает и как это реализовано в Python.
Что делает код?
Код обрабатывает JSON-файл (148.A1-A2.DT1.json), содержащий данные о транспортных средствах (скорость, координаты, полоса движения и т.д.) и выполняет следующие задачи:
- Анализ данных и поиск аномалий: Смотрит, какие данные есть, и ищет машины с необычной скоростью (например, слишком быстрой или медленной).
- Анализ трафика: Считает среднюю скорость машин и расстояния между ними на разных полосах.
- Зоны обнаружения: Определяет, в каких зонах сенсор видит машины, двигающиеся быстрее 80 км/ч, и как эти зоны зависят от скорости.
- Кластеризация: Группирует машины по их координатам и скорости, чтобы угадать, на какой полосе они едут, без использования данных о полосах.
Как это работает?
Код разбит на четыре функции (task_1_analysis, task_2_solution, task_3_solution, task_4_solution), каждая решает свою задачу. Вот как они реализованы в Python:
Загрузка и подготовка данных
Код использует библиотеку json для чтения файла 148.A1-A2.DT1.json. Библиотека pandas преобразует данные в таблицу (DataFrame), где каждая строка — это информация о машине (скорость, координаты, время, полоса и т.д.). Время (time) переводится в формат datetime для удобной работы.
with open(filename) as f:
data = json.load(f)
df = pd.json_normalize(data['objects'])
df = df.explode('rows_data')
df = pd.DataFrame(df['rows_data'].tolist())
df['time'] = pd.to_datetime(df['time'])
Task 1: Анализ данных и поиск аномалий
Функция task_1_analysis выводит информацию о данных (типы, количество строк) с помощью df.info() и статистику (среднее, минимум, максимум) с помощью df.describe(). Для поиска аномалий в скорости (obj_speed) используется метод межквартильного размаха (IQR): машины, чья скорость сильно отличается от средней, считаются аномалиями. Результаты выводятся в консоль.
Q1 = df['obj_speed'].quantile(0.25)
Q3 = df['obj_speed'].quantile(0.75)
IQR = Q3 - Q1
outliers = df[(df['obj_speed'] < Q1 - 1.5 * IQR) | (df['obj_speed'] > Q3 + 1.5 * IQR)]
Task 2: Анализ трафика
Функция task_2_solution считает среднюю скорость всех машин и по каждой полосе с помощью df.groupby('lane')['obj_speed'].mean(). Для расчета расстояний между машинами в одной полосе данные сортируются по полосе, ID машины и времени. Расстояние между соседними машинами вычисляется как разница в координатах point_x с помощью df.groupby('lane')['point_x'].diff(). Средние и минимальные расстояния по полосам выводятся.
avg_speed = df['obj_speed'].mean()
avg_speed_by_lane = df.groupby('lane')['obj_speed'].mean()
df_sorted['distance'] = df_sorted.groupby('lane')['point_x'].diff().abs()
Task 3: Зоны обнаружения
Функция task_3_solution фильтрует машины со скоростью выше 80 км/ч и определяет зоны, где сенсор их видит, на основе минимальных и максимальных координат point_x. Зоны рассчитываются для каждой полосы и для разных порогов скорости (80, 90, 100, 110 км/ч). Результаты визуализируются в графике (с помощью matplotlib), показывающем, как длина зоны обнаружения зависит от скорости. График сохраняется в файл detection_zone_vs_speed.png.
high_speed_df = df[df['obj_speed'] > 80].copy()
detection_zones = high_speed_df.groupby('uuid').agg({'point_x': ['min', 'max'], 'lane': 'first'})
plt.plot(lane_data['threshold'], lane_data['length'], marker='o', label=f'Lane {lane}')
Task 4: Кластеризация
Функция task_4_solution использует алгоритм KMeans из библиотеки sklearn для группировки машин по координатам (point_x, point_y) и скорости (obj_speed), без учета полос. Предполагается, что есть 5 полос, поэтому задается 5 кластеров (n_clusters=5). Кластеры сопоставляются с реальными полосами, и считается точность угадывания полос. Результаты и объяснение выбора метода выводятся.
features = ['point_y', 'point_x', 'obj_speed']
X = df[features].dropna()
kmeans = KMeans(n_clusters=5, random_state=42)
X['cluster'] = kmeans.fit_predict(X[features])
Для чего нужен код?
Этот код нужен для анализа данных с дорожного сенсора, чтобы:
- Понять, как движется транспорт (скорости, расстояния, зоны обнаружения).
- Найти аномалии (например, слишком медленные или быстрые машины).
- Автоматически определять полосы движения без явных данных о них.
- Помочь в управлении трафиком, например, для оптимизации светофоров или выявления опасных ситуаций.
Как это реализовано в Python?
Код использует популярные Python-библиотеки для обработки и анализа данных:
json— для чтения JSON-файла.pandas— для работы с данными в виде таблицы.numpy— для числовых вычислений.matplotlib— для построения графиков.sklearn— для кластеризации (KMeans).datetime— для работы с датой и временем.
Ключевые приемы:
- Данные обрабатываются как таблица (
DataFrame), что упрощает группировку, фильтрацию и расчеты. - Группировка (
groupby) используется для анализа по полосам или ID машин. - Визуализация с помощью
matplotlibпомогает увидеть зависимости (например, зоны обнаружения). - Алгоритм
KMeansавтоматически находит группы в данных.
Итог
Код — это инструмент для анализа транспортных данных, который загружает JSON, обрабатывает его в таблице, выполняет расчеты (скорость, расстояния, зоны) и группирует машины по полосам. Он использует популярные Python-библиотеки, чтобы сделать анализ удобным и наглядным, а результаты выводит в консоль и графики. Это полезно для изучения трафика и улучшения дорожной инфраструктуры.