84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
|
"""
|
|||
|
|
Фоновая задача для проверки потери связи с объектами.
|
|||
|
|
Создаёт события CONNECTION_LOST если нет данных более N минут.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
from datetime import datetime, timedelta
|
|||
|
|
from sqlalchemy import select, desc
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.database import async_session_maker
|
|||
|
|
from app.models import Vehicle, Position, Event
|
|||
|
|
from app.config import settings
|
|||
|
|
from app.services.websocket_manager import manager
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def check_connections():
|
|||
|
|
"""Проверить все объекты на потерю связи"""
|
|||
|
|
async with async_session_maker() as db:
|
|||
|
|
# Получить все объекты
|
|||
|
|
result = await db.execute(select(Vehicle))
|
|||
|
|
vehicles = result.scalars().all()
|
|||
|
|
|
|||
|
|
now = datetime.utcnow()
|
|||
|
|
threshold = now - timedelta(minutes=settings.connection_lost_minutes)
|
|||
|
|
|
|||
|
|
for vehicle in vehicles:
|
|||
|
|
# Получить последнюю позицию
|
|||
|
|
pos_result = await db.execute(
|
|||
|
|
select(Position)
|
|||
|
|
.where(Position.vehicle_id == vehicle.id)
|
|||
|
|
.order_by(desc(Position.timestamp))
|
|||
|
|
.limit(1)
|
|||
|
|
)
|
|||
|
|
last_pos = pos_result.scalar_one_or_none()
|
|||
|
|
|
|||
|
|
if not last_pos:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# Проверить, прошло ли достаточно времени
|
|||
|
|
if last_pos.timestamp < threshold:
|
|||
|
|
# Проверить, не было ли уже события CONNECTION_LOST за последние N минут
|
|||
|
|
event_result = await db.execute(
|
|||
|
|
select(Event)
|
|||
|
|
.where(Event.vehicle_id == vehicle.id)
|
|||
|
|
.where(Event.type == "CONNECTION_LOST")
|
|||
|
|
.where(Event.timestamp > threshold)
|
|||
|
|
.limit(1)
|
|||
|
|
)
|
|||
|
|
existing_event = event_result.scalar_one_or_none()
|
|||
|
|
|
|||
|
|
if not existing_event:
|
|||
|
|
# Создать событие
|
|||
|
|
minutes_ago = (now - last_pos.timestamp).total_seconds() / 60
|
|||
|
|
event = Event(
|
|||
|
|
vehicle_id=vehicle.id,
|
|||
|
|
timestamp=now,
|
|||
|
|
type="CONNECTION_LOST",
|
|||
|
|
payload={
|
|||
|
|
"last_seen": last_pos.timestamp.isoformat(),
|
|||
|
|
"minutes_ago": round(minutes_ago, 1),
|
|||
|
|
"lat": last_pos.lat,
|
|||
|
|
"lon": last_pos.lon
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
db.add(event)
|
|||
|
|
await db.commit()
|
|||
|
|
await db.refresh(event)
|
|||
|
|
|
|||
|
|
# Отправить через WebSocket
|
|||
|
|
await manager.broadcast_event(event)
|
|||
|
|
print(f"⚠️ CONNECTION_LOST: {vehicle.name} (нет данных {minutes_ago:.0f} мин)")
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def connection_checker_loop():
|
|||
|
|
"""Бесконечный цикл проверки соединений"""
|
|||
|
|
print("🔍 Connection checker started")
|
|||
|
|
while True:
|
|||
|
|
try:
|
|||
|
|
await check_connections()
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"Connection checker error: {e}")
|
|||
|
|
await asyncio.sleep(60) # Проверять каждую минуту
|