86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
import asyncio
|
|
import os
|
|
from typing import Dict, Optional
|
|
|
|
import socketio
|
|
|
|
from .geo_cache import AsyncReverseGeocoder
|
|
from .icloud_monitor import FindMyMonitor
|
|
from .models import iCloudLocationData
|
|
|
|
|
|
ACTIVE_SIMULATION_STATES = {"RUNNING", "PAUSED"}
|
|
|
|
|
|
def env_flag(name: str, default: str = "True") -> bool:
|
|
return os.getenv(name, default).lower() == "true"
|
|
|
|
|
|
class LocationSimulationState:
|
|
def __init__(self):
|
|
self.connected_clients: set[str] = set()
|
|
self.current_location: Optional[Dict[str, str]] = None
|
|
self.device_name: Optional[str] = os.getenv("SELECTED_DEVICE_NAME")
|
|
self.fmf_location: Optional[iCloudLocationData] = None
|
|
self.fmf_queue: asyncio.Queue = asyncio.Queue()
|
|
self.icloud_consumer_task = None
|
|
self.icloud_monitor = FindMyMonitor(self.fmf_queue)
|
|
self.icloud_monitor_task = None
|
|
self.icloud_monitor_enabled: bool = False
|
|
self.tunnel_watcher_task = None
|
|
self.next_move: Optional[float] = None
|
|
self.simulation_queue: asyncio.Queue = asyncio.Queue()
|
|
self.simulation_queue_data: Dict = {}
|
|
self.simulation_queue_order: list[str] = []
|
|
self.simulation_queue_deleted_items: list[str] = []
|
|
self.simulation_queue_pending_ids: set[str] = set()
|
|
self.simulation_queue_state: str = "STOPPED"
|
|
self.simulation_noise: bool = True
|
|
self.simulation_task: Optional[asyncio.Task] = None
|
|
self.sio: socketio.AsyncServer = socketio.AsyncServer(
|
|
async_mode="asgi", cors_allowed_origins="*"
|
|
)
|
|
self.test_mode: bool = env_flag("TEST_MODE", "True")
|
|
self.tunnel = None
|
|
self.udid: Optional[str] = None
|
|
self.reverse_geocode = AsyncReverseGeocoder()
|
|
|
|
@property
|
|
def simulation_active(self) -> bool:
|
|
task_active = (
|
|
self.simulation_task is not None
|
|
and not self.simulation_task.done()
|
|
)
|
|
return self.simulation_queue_state in ACTIVE_SIMULATION_STATES and task_active
|
|
|
|
@simulation_active.setter
|
|
def simulation_active(self, active: bool) -> None:
|
|
if active:
|
|
if self.simulation_queue_state not in ACTIVE_SIMULATION_STATES:
|
|
self.simulation_queue_state = "RUNNING"
|
|
elif self.simulation_queue_state != "SHUTDOWN":
|
|
self.simulation_queue_state = "STOPPED"
|
|
|
|
def get_current_loc_id(self) -> Optional[str]:
|
|
if not isinstance(self.current_location, dict):
|
|
return None
|
|
loc_id = self.current_location.get("loc_id")
|
|
if isinstance(loc_id, str) and loc_id:
|
|
return loc_id
|
|
return None
|
|
|
|
def set_current_loc_id(self, loc_id: Optional[str]) -> None:
|
|
if isinstance(loc_id, str) and loc_id:
|
|
self.current_location = {"loc_id": loc_id}
|
|
else:
|
|
self.current_location = None
|
|
|
|
def get_current_item(self) -> Optional[dict]:
|
|
current_loc_id = self.get_current_loc_id()
|
|
if current_loc_id is None:
|
|
return None
|
|
item = self.simulation_queue_data.get(current_loc_id)
|
|
if isinstance(item, dict):
|
|
return item
|
|
return None
|