[benchmarker] Add active flight operation

monitoring/benchmarker/configurations/loads.py

@@ -19,6 +19,9 @@ class WorkflowType(StrEnum):
     FlightPlannerFlight = "flight_planner.flight"
     """An operation consisting of managing a flight from end to end, including all associated UTM actions like establishing an operational intent and deleting it."""
 
+    FlightPlannerActiveFlight = "flight_planner.active_flight"
+    """An operation involving only active flight time from the time the operator starts using the airspace until the time the operator finishes using the airspace."""
+
 
 class OperationType(str):
     """Type of operation providing load to a system being benchmarked."""

monitoring/benchmarker/engine/users/flight_planner/astm_net_rid.py

@@ -154,6 +154,10 @@ async def create_isa(self, flight: Flight, isa_id: str) -> list[FlightAction]:
                 causes_flight_failure=not isa_success,
             )
         )
+        if isa_success:
+            flight.extend_achieved_start_time(datetime.now(UTC))
+        else:
+            flight.clear_to_fly = False
 
         if isa_success and self.isa_per_flight.after_flight_end:
             new_actions.append(

monitoring/benchmarker/engine/users/flight_planner/flight_planner.py

@@ -167,6 +167,19 @@ async def run_custom_workflow(self, stop_event: asyncio.Event) -> None:
                             if a.causes_flight_failure
                         )
                     )
+
+                # Record the time period where the operator was actually using the airspace, or would have used the airspace
+                active_flight_op = ExecutedOperation(
+                    type=OperationType(WorkflowType.FlightPlannerActiveFlight),
+                    origin=self.user_id,
+                    initiated_at=StringBasedDateTime(flight.get_achieved_start_time()),
+                    completed_at=StringBasedDateTime(flight.actual_end_time),
+                    successful=flight.clear_to_fly,
+                    query=None,
+                )
+                self.record_operation(active_flight_op)
+
+                # Record the whole set of flight activities (including managing the flight with UTM activities)
                 flight_op = ExecutedOperation(
                     type=OperationType(WorkflowType.FlightPlannerFlight),
                     origin=self.user_id,

monitoring/benchmarker/engine/users/flight_planner/framework.py

@@ -45,7 +45,18 @@ class CompletedFlightAction:
 class Flight:
     id: FlightID
     volumes: Volume4DCollection
+
     actual_end_time: datetime
+    """When the operator actually stopped using the airspace (often before the end of the last volume)."""
+
+    clear_to_fly: bool = True
+    """Whether the operator was cleared for takeoff by UTM activities (all UTM activities have veto power)."""
+
+    achieved_start_time: datetime | None = None
+    """When the operator actually started using the airspace.
+
+    If None, implies the operator started using the airspace as soon as the first declared volume started."""
+
     completed_actions: list[CompletedFlightAction] = field(default_factory=lambda: [])
 
     @property
@@ -74,6 +85,16 @@ def successful(self) -> bool:
             not action.causes_flight_failure for action in self.completed_actions
         )
 
+    def extend_achieved_start_time(self, t: datetime) -> None:
+        """Declare that a takeoff prerequisite was completed at time t."""
+        t_start = self.get_achieved_start_time()
+        if t > t_start:
+            self.achieved_start_time = t
+
+    def get_achieved_start_time(self) -> datetime:
+        """Retrieve the time the operator actually started using the airspace."""
+        return self.achieved_start_time or self.start_time
+
     async def complete(self) -> list[FlightAction]:
         self.completed_actions.append(
             CompletedFlightAction(

monitoring/benchmarker/engine/users/flight_planner/scd.py

@@ -275,8 +275,10 @@ async def ensure_subscription_exists(
         )
         self.subscription_checked = True
         if success:
+            flight.extend_achieved_start_time(datetime.now(UTC))
             return list(self.get_create_actions(flight))
         else:
+            flight.clear_to_fly = False
             return []
 
     def get_create_actions(self, flight: Flight) -> Iterable[FlightAction]:
@@ -451,7 +453,10 @@ async def upsert_op_intent_ref(
                         requested_ovn,
                     )
 
-        if not success:
+        if success:
+            flight.extend_achieved_start_time(datetime.now(UTC))
+        else:
+            flight.clear_to_fly = False
             return []
 
         if op_intent_ref is None:
@@ -550,6 +555,7 @@ async def upsert_op_intent_ref(
                     )
                 )
                 if flight_aborted:
+                    flight.clear_to_fly = False
                     logger.debug(
                         f"Transition to {state.value} for flight {flight.id} completed {dt_s:.1f}s too late; removing op intent early"
                     )