AI-based personalization system for connected cars
A connected car is a vehicle with a constant internet connection that collects data about driving, routes, preferences, and system status. AI uses this data to personalize the interface, provide predictive maintenance, and provide real-time situational recommendations.
Driver profile and HMI adaptation
import numpy as np
import pandas as pd
from anthropic import Anthropic
import json
from collections import defaultdict
class DriverProfileBuilder:
"""Построение профиля водителя из телематических данных"""
def build_driving_profile(self, telemetry: pd.DataFrame,
driver_id: str,
days: int = 30) -> dict:
"""Профиль стиля вождения и предпочтений"""
driver_data = telemetry[
(telemetry['driver_id'] == driver_id) &
(telemetry['timestamp'] >= pd.Timestamp.now() - pd.Timedelta(days=days))
]
if driver_data.empty:
return {'driver_id': driver_id, 'is_new': True}
profile = {
'driver_id': driver_id,
# Стиль вождения
'avg_speed_kmh': driver_data['speed_kmh'].mean(),
'hard_braking_events_per_100km': (
driver_data['hard_braking'].sum() /
max(driver_data['distance_km'].sum(), 1) * 100
),
'aggressive_acceleration': (driver_data['acceleration_g'] > 0.3).mean(),
'highway_ratio': (driver_data['road_type'] == 'highway').mean(),
# Маршруты и время
'most_common_start_hour': int(driver_data['hour'].mode().iloc[0]),
'avg_trip_distance_km': driver_data.groupby('trip_id')['distance_km'].sum().mean(),
'top_destinations': driver_data['destination_category'].value_counts().head(3).to_dict(),
# Предпочтения в машине
'preferred_temp_celsius': driver_data['cabin_temp_set'].median() if 'cabin_temp_set' in driver_data.columns else 21,
'music_genre_preference': driver_data.get('music_genre', pd.Series(['pop'])).mode().iloc[0],
'uses_voice_control': (driver_data.get('voice_commands', 0) > 0).mean() > 0.3,
# Eco-score
'eco_score': self._compute_eco_score(driver_data),
}
profile['driving_style'] = self._classify_driving_style(profile)
return profile
def _compute_eco_score(self, data: pd.DataFrame) -> float:
"""Эко-скор вождения (0-100)"""
score = 100.0
# Штрафы за агрессивное вождение
if 'hard_braking' in data.columns:
score -= data['hard_braking'].mean() * 200
if 'acceleration_g' in data.columns:
score -= (data['acceleration_g'] > 0.3).mean() * 50
# Оценка эффективности скорости (оптимально 80-100 км/ч на трассе)
if 'speed_kmh' in data.columns:
high_speed = (data['speed_kmh'] > 130).mean()
score -= high_speed * 30
return round(float(np.clip(score, 0, 100)), 1)
def _classify_driving_style(self, profile: dict) -> str:
eco = profile.get('eco_score', 70)
hard_braking = profile.get('hard_braking_events_per_100km', 2)
if eco > 80 and hard_braking < 1:
return 'eco'
elif hard_braking > 5 or profile.get('aggressive_acceleration', 0) > 0.3:
return 'sporty'
return 'normal'
class InCarPersonalizationSystem:
"""Персонализация интерфейса и систем автомобиля"""
def __init__(self):
self.llm = Anthropic()
def auto_configure_car_settings(self, driver_profile: dict) -> dict:
"""Автоматическая настройка автомобиля при посадке водителя"""
settings = {
# Климат
'cabin_temperature': driver_profile.get('preferred_temp_celsius', 21),
'fan_speed': 'auto',
'seat_heating': self._determine_seat_heating(driver_profile),
# Интерфейс
'dashboard_layout': 'sport' if driver_profile.get('driving_style') == 'sporty' else 'comfort',
'display_brightness': 'auto',
'voice_activation': driver_profile.get('uses_voice_control', False),
# Звук
'radio_station': self._get_preferred_station(driver_profile),
'volume_level': driver_profile.get('preferred_volume', 30),
# Помощники
'adaptive_cruise': driver_profile.get('driving_style') == 'eco',
'lane_assist_sensitivity': 'medium' if driver_profile.get('driving_style') != 'sporty' else 'low',
}
return settings
def _determine_seat_heating(self, profile: dict) -> bool:
"""Включать ли подогрев сидений"""
hour = profile.get('most_common_start_hour', 12)
return hour < 8 # Утром → включаем
def _get_preferred_station(self, profile: dict) -> str:
genre_stations = {
'rock': 'rock_station_id',
'pop': 'pop_station_id',
'classical': 'classical_station_id',
'electronic': 'electronic_station_id',
}
genre = profile.get('music_genre_preference', 'pop')
return genre_stations.get(genre, 'top_40_station_id')
def generate_contextual_suggestions(self, context: dict,
driver_profile: dict) -> list[dict]:
"""
Контекстные рекомендации в реальном времени.
context: location, time, fuel_level, battery_level, next_appointment
"""
suggestions = []
# Низкий заряд/топливо
fuel_level = context.get('fuel_level_pct', 100)
if fuel_level < 20:
nearby_stations = context.get('nearby_stations', [])
if nearby_stations:
suggestions.append({
'type': 'fuel_alert',
'priority': 'high',
'message': f"Топливо {fuel_level}%. Ближайшая заправка через {nearby_stations[0].get('distance_km', 2):.1f} км",
'action': 'navigate_to_station'
})
# Следующая встреча/назначение
next_appointment = context.get('next_calendar_event')
if next_appointment:
event_time = pd.to_datetime(next_appointment.get('time'))
now = pd.Timestamp.now()
minutes_to_event = (event_time - now).total_seconds() / 60
travel_time_est = context.get('travel_time_to_event_min', 20)
if minutes_to_event < travel_time_est * 1.2:
suggestions.append({
'type': 'departure_alert',
'priority': 'high',
'message': f"Пора выехать на {next_appointment.get('title')}. Дорога займёт {travel_time_est:.0f} мин",
'action': 'start_navigation'
})
# Эко-рекомендация
if driver_profile.get('driving_style') != 'eco' and context.get('current_speed', 0) > 130:
savings_pct = round((context.get('current_speed', 0) - 110) / 110 * 15, 0)
suggestions.append({
'type': 'eco_tip',
'priority': 'low',
'message': f"Снизьте скорость до 110 км/ч — сэкономите ~{savings_pct}% топлива"
})
return suggestions
class PredictiveMaintenanceAdvisor:
"""Предиктивное техобслуживание"""
def predict_maintenance_needs(self, telemetry: pd.DataFrame,
maintenance_history: pd.DataFrame,
vehicle: dict) -> list[dict]:
"""Прогноз необходимого обслуживания"""
alerts = []
current_odometer = vehicle.get('odometer_km', 0)
# Замена масла (каждые 10000 км или 12 мес)
last_oil_change_km = maintenance_history[
maintenance_history['service_type'] == 'oil_change'
]['odometer_km'].max() if len(maintenance_history) > 0 else 0
km_since_oil = current_odometer - last_oil_change_km
if km_since_oil > 8000:
urgency = 'urgent' if km_since_oil > 10000 else 'upcoming'
alerts.append({
'service_type': 'oil_change',
'urgency': urgency,
'km_overdue': max(0, km_since_oil - 10000),
'message': f'Замена масла {"просрочена" if urgency == "urgent" else "через"} {max(0, 10000 - km_since_oil):.0f} км'
})
# Тормозные колодки (из телеметрии hard braking)
recent_hard_braking = telemetry[
telemetry['timestamp'] >= pd.Timestamp.now() - pd.Timedelta(days=90)
]['hard_braking'].sum() if 'hard_braking' in telemetry.columns else 0
if recent_hard_braking > 50:
alerts.append({
'service_type': 'brake_inspection',
'urgency': 'upcoming',
'message': 'Рекомендуем проверить тормозные колодки — зафиксировано интенсивное торможение'
})
return alerts
Connected car personalization increases driver satisfaction by 20-30% (NPS) and reduces accidents by 10-15% through adaptive alerts. Predictive maintenance reduces unscheduled breakdowns by 25-40% and repair costs by 15-20%.