traffic/algorithms/navigation.py
def holding_pattern(
        self,
        min_altitude=7000,
        turning_threshold=0.5,
        low_limit=pd.Timedelta("30 seconds"),  # noqa: B008
        high_limit=pd.Timedelta("10 minutes"),  # noqa: B008
        turning_limit=pd.Timedelta("5 minutes"),  # noqa: B008
    ) -> Iterator["Flight"]:
        """Iterates on parallel segments candidates for identifying
        a holding pattern.

        .. warning::

            This API is not stable yet. The interface may change in a near
            future.

        """
        # avoid parts that are really way too low
        alt_above = self.query(f"altitude > {min_altitude}")  # type: ignore
        if alt_above is None:
            return

        straight_line = (
            alt_above.unwrap()
            .assign(
                turning_rate=lambda x: x.track_unwrapped.diff()
                / x.timestamp.diff().dt.total_seconds()
            )
            .filter(turning_rate=17)
            .query(f"turning_rate.abs() < {turning_threshold}")
        )
        if straight_line is None:
            return

        chunk_candidates = list(
            (chunk.start, chunk.duration, chunk.mean("track_unwrapped"), chunk)
            for chunk in straight_line.split("10s")
            if low_limit <= chunk.duration < high_limit
        )

        next_ = None
        for (
            (start1, duration1, track1, chunk1),
            (start2, _, track2, chunk2),
        ) in zip(chunk_candidates, chunk_candidates[1:]):
            if (
                start2 - start1 - duration1 < turning_limit
                and abs(abs(track1 - track2) - 180) < 15
            ):
                yield chunk1
                next_ = chunk2
            else:
                if next_ is not None:
                    yield next_
                next_ = None
Iterates on parallel segments candidates for identifying
        a holding pattern.

        .. warning::

            This API is not stable yet. The interface may change in a near
            future.

        
Iterates on parallel segments candidates for identifying
a holding pattern.



This API is not stable yet. The interface may change in a near
future.
Iterates on parallel segments candidates for identifying a holding pattern . This API is not stable yet . The interface may change in a near future .
def holding_pattern(
        self,
        min_altitude=7000,
        turning_threshold=0.5,
        low_limit=pd.Timedelta("30 seconds"),  
        high_limit=pd.Timedelta("10 minutes"),  
        turning_limit=pd.Timedelta("5 minutes"),  
    ) -> Iterator["Flight"]:
        alt_above = self.query(f"altitude > {min_altitude}")  
        if alt_above is None:
            return
        straight_line = (
            alt_above.unwrap()
            .assign(
                turning_rate=lambda x: x.track_unwrapped.diff()
                / x.timestamp.diff().dt.total_seconds()
            )
            .filter(turning_rate=17)
            .query(f"turning_rate.abs() < {turning_threshold}")
        )
        if straight_line is None:
            return
        chunk_candidates = list(
            (chunk.start, chunk.duration, chunk.mean("track_unwrapped"), chunk)
            for chunk in straight_line.split("10s")
            if low_limit <= chunk.duration < high_limit
        )
        next_ = None
        for (
            (start1, duration1, track1, chunk1),
            (start2, _, track2, chunk2),
        ) in zip(chunk_candidates, chunk_candidates[1:]):
            if (
                start2 - start1 - duration1 < turning_limit
                and abs(abs(track1 - track2) - 180) < 15
            ):
                yield chunk1
                next_ = chunk2
            else:
                if next_ is not None:
                    yield next_
                next_ = None
def holding_pattern ( self , min_altitude = 7000 , turning_threshold = 0.5 , low_limit = pd . Timedelta ( "30 seconds" ) , high_limit = pd . Timedelta ( "10 minutes" ) , turning_limit = pd . Timedelta ( "5 minutes" ) , ) -> Iterator [ "Flight" ] : alt_above = self . query ( f"altitude > {min_altitude}" ) if alt_above is None : return straight_line = ( alt_above . unwrap ( ) . assign ( turning_rate = lambda x : x . track_unwrapped . diff ( ) / x . timestamp . diff ( ) . dt . total_seconds ( ) ) . filter ( turning_rate = 17 ) . query ( f"turning_rate.abs() < {turning_threshold}" ) ) if straight_line is None : return chunk_candidates = list ( ( chunk . start , chunk . duration , chunk . mean ( "track_unwrapped" ) , chunk ) for chunk in straight_line . split ( "10s" ) if low_limit <= chunk . duration < high_limit ) next_ = None for ( ( start1 , duration1 , track1 , chunk1 ) , ( start2 , _ , track2 , chunk2 ) , ) in zip ( chunk_candidates , chunk_candidates [ 1 : ] ) : if ( start2 - start1 - duration1 < turning_limit and abs ( abs ( track1 - track2 ) - 180 ) < 15 ) : yield chunk1 next_ = chunk2 else : if next_ is not None : yield next_ next_ = None
Iterates on parallel segments candidates for identifying
a holding pattern.
Iterates on parallel segments candidates for identifying a holding pattern .
# noqa: B008 # noqa: B008 # noqa: B008 """Iterates on parallel segments candidates for identifying
        a holding pattern.

        .. warning::

            This API is not stable yet. The interface may change in a near
            future.

        """ # avoid parts that are really way too low # type: ignore
self
min_altitude
turning_threshold
low_limit
high_limit
turning_limit