mirror of
https://github.com/dcs-retribution/dcs-retribution.git
synced 2025-11-10 15:41:24 +00:00
656a98675e
Rather than polling at 60Hz (which may be faster than the tick rate, wasting cycles; and also makes synchronization annoying), collect events during the tick and emit them after (rate limited, pooling events until it is time for another event to send). This can be improved by paying attention to the aircraft update list, which would allow us to avoid updating aircraft that don't have a status change. To do that we need to be able to quickly lookup a FlightJs matching a Flight through, and Flight isn't hashable. We should also be removing dead events and de-duplicating. Currently each flight has an update for every tick, but only the latest one matters. Combat update events also don't matter if the same combat is new in the update. https://github.com/dcs-liberation/dcs_liberation/issues/1680
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import TYPE_CHECKING
|
|
|
|
from dcs import Point
|
|
|
|
from game.ato.flightstate import InFlight
|
|
from game.ato.starttype import StartType
|
|
from game.utils import Distance, LBS_TO_KG, Speed, meters
|
|
|
|
if TYPE_CHECKING:
|
|
from game.sim.gameupdateevents import GameUpdateEvents
|
|
|
|
|
|
def lerp(v0: float, v1: float, t: float) -> float:
|
|
return (1 - t) * v0 + t * v1
|
|
|
|
|
|
class Navigating(InFlight):
|
|
def on_game_tick(
|
|
self, events: GameUpdateEvents, time: datetime, duration: timedelta
|
|
) -> None:
|
|
super().on_game_tick(events, time, duration)
|
|
events.update_flight(self.flight)
|
|
|
|
def progress(self) -> float:
|
|
return (
|
|
self.elapsed_time.total_seconds()
|
|
/ self.total_time_to_next_waypoint.total_seconds()
|
|
)
|
|
|
|
def estimate_position(self) -> Point:
|
|
x0 = self.current_waypoint.position.x
|
|
y0 = self.current_waypoint.position.y
|
|
x1 = self.next_waypoint.position.x
|
|
y1 = self.next_waypoint.position.y
|
|
progress = self.progress()
|
|
return Point(lerp(x0, x1, progress), lerp(y0, y1, progress))
|
|
|
|
def estimate_altitude(self) -> tuple[Distance, str]:
|
|
return (
|
|
meters(
|
|
lerp(
|
|
self.current_waypoint.alt.meters,
|
|
self.next_waypoint.alt.meters,
|
|
self.progress(),
|
|
)
|
|
),
|
|
self.current_waypoint.alt_type,
|
|
)
|
|
|
|
def estimate_speed(self) -> Speed:
|
|
return self.flight.flight_plan.speed_between_waypoints(
|
|
self.current_waypoint, self.next_waypoint
|
|
)
|
|
|
|
def estimate_fuel(self) -> float:
|
|
initial_fuel = self.estimate_fuel_at_current_waypoint()
|
|
ppm = self.flight.flight_plan.fuel_rate_to_between_points(
|
|
self.current_waypoint, self.next_waypoint
|
|
)
|
|
if ppm is None:
|
|
return initial_fuel
|
|
position = self.estimate_position()
|
|
distance = meters(self.current_waypoint.position.distance_to_point(position))
|
|
consumption = distance.nautical_miles * ppm * LBS_TO_KG
|
|
return initial_fuel - consumption
|
|
|
|
@property
|
|
def is_waiting_for_start(self) -> bool:
|
|
return False
|
|
|
|
@property
|
|
def spawn_type(self) -> StartType:
|
|
return StartType.IN_FLIGHT
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return f"Flying to {self.next_waypoint.name}"
|