Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 42 additions & 1 deletion bluebird-dt/bluebird_dt/core/action.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ def __init__(
direction
- `change_heading_by`: change heading by given degrees
- `maintain_current_heading`: follow the current heading (useful if was following route)
- `intercept_radial`: intercept the radial at the specified degrees from the specified named Fix
- `change_flight_level_to`: go to the specified flight level
- `change_flight_level_by`: descend/ascend by specified flight levels
- `descend_when_ready,level_by_fix`: descend when ready to the specified flight level by the named Fix
Expand All @@ -77,6 +78,7 @@ def __init__(
- `change_heading_to_by_direction`: tuple[int, Literal['left', 'right', 'shortest']]
- `change_heading_by`: int
- `maintain_current_heading`: value will be automatically set to 0
- `intercept_radial`: tuple[Literal['immediate', 'when_ready', 'target_fix'], str, int]
- `change_flight_level_to`: int
- `change_flight_level_by`: int
- `change_cas_to`: int or None (None indicates "fly at own speed")
Expand Down Expand Up @@ -110,6 +112,7 @@ def __init__(
>>> action = Action("AIR123", "route_direct_to", ["PORT1"], sector="sector_xplus")
>>> action = Action("XYZ567", "change_cas_to", 450.0, "human", sector="sector_xplus")
>>> action = Action("AIR23", "descend_when_ready,level_by_fix", (200, "FIX1"), sector="sector_xplus")
>>> action = Action("AIR23", "intercept_radial", ("immediate", "FIX1", 200), sector="sector_xplus")
"""

if len(callsign) == 0:
Expand Down Expand Up @@ -162,6 +165,10 @@ def value(self, value: int | float | str | list[str] | tuple[int, str] | None) -
"route_turn_segment",
"heading_segment",
"heading_turn_segment",
"track_segment",
"fixed_rate_turn",
"fixed_radius_turn",
"intercept_radial",
"message",
"change_heading_to_by_direction",
]:
Expand All @@ -179,6 +186,32 @@ def value(self, value: int | float | str | list[str] | tuple[int, str] | None) -
if not isinstance(value[1], str):
raise ValueError(f"Action value[1] must be a string for {self.kind}. Got {type(value[1])}.")
self._value = value
if self.kind in ["fixed_rate_turn", "fixed_radius_turn"]:
# first, check if it is a string (from the clearance df) and convert it back to a tuple.
# the string should be of the form "(int/float, int/float)"
if isinstance(value, str):
value = ast.literal_eval(value)
if not isinstance(value, tuple):
raise ValueError(f"Action value must be a tuple for {self.kind}. Got {type(value)}.")
if not isinstance(value[0], int | float | np.integer | np.floating):
raise ValueError(f"Action value[0] must be an float or int for {self.kind}. Got {type(value[0])}.")
if not isinstance(value[1], int | float | np.integer | np.floating):
raise ValueError(f"Action value[1] must be a float or int for {self.kind}. Got {type(value[1])}.")
self._value = value
if self.kind == "intercept_radial":
# first, check if it is a string (from the clearance df) and convert it back to a tuple.
# the string should be of the form "(str, str, int/float)"
if isinstance(value, str):
value = ast.literal_eval(value)
if not isinstance(value, tuple):
raise ValueError(f"Action value must be a tuple for {self.kind}. Got {type(value)}.")
if not isinstance(value[0], str):
raise ValueError(f"Action value[0] must be a string for {self.kind}. Got {type(value[0])}.")
if not isinstance(value[1], str):
raise ValueError(f"Action value[1] must be a string for {self.kind}. Got {type(value[1])}.")
if not isinstance(value[2], int | float | np.integer | np.floating):
raise ValueError(f"Action value[2] must be a float or int for {self.kind}. Got {type(value[2])}.")
self._value = value
elif self.kind in ["route_direct_to", "route_segment", "route_turn_segment"]:
# allow single fix or list of fixes, and for route_direct_to value
# to be a string representing a list
Expand All @@ -194,6 +227,9 @@ def value(self, value: int | float | str | list[str] | tuple[int, str] | None) -
elif self.kind == "heading_segment" or self._kind == "heading_turn_segment":
# system generated heading so allow for floating point headings
self._value = float(value)
elif self.kind == "track_segment":
# system generated ground track so allow for floating point headings
self._value = float(value)
elif self.kind == "maintain_current_heading":
# value is irrelevant. Always set to zero for consistency
self._value = 0
Expand Down Expand Up @@ -333,7 +369,12 @@ def from_json(s: str) -> Action:
kind = data["kind"]
value = data["value"]

if "level_by_fix" in kind or kind == "change_heading_to_by_direction":
if "level_by_fix" in kind or kind in [
"change_heading_to_by_direction",
"fixed_rate_turn",
"fixed_radius_turn",
"intercept_radial",
]:
value = tuple(value)

voice_representation: str | None = data.get("voice_representation", None)
Expand Down
4 changes: 4 additions & 0 deletions bluebird-dt/bluebird_dt/utility/supported_actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
"change_heading_to_by_direction",
"change_heading_by",
"maintain_current_heading",
"intercept_radial",
],
"speed": [
"change_cas_to",
Expand All @@ -26,6 +27,9 @@
"route_turn_segment",
"heading_segment",
"heading_turn_segment",
"track_segment",
"fixed_radius_turn",
"fixed_rate_turn",
],
"message": ["message"],
}
21 changes: 19 additions & 2 deletions bluebird-dt/tests/unit/core/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,10 @@ def make_random_heading() -> float:
"""Return a random heading between -180 and +180 to 2 decimal places"""
return round(random.uniform(-180.00, 180.00), 2)

def make_random_ground_track() -> float:
"""Return a random ground track between -180 and +180 to 2 decimal places"""
return round(random.uniform(-180.00, 180.00), 2)

def make_random_callsign() -> str:
"""Return a random callsign AIRXYZ where XYZ are digits"""
return "AIR" + "".join(random.choice(string.digits) for _ in range(3))
Expand Down Expand Up @@ -337,8 +341,12 @@ def make_random_adjacent_volumes() -> tuple[Volume, Volume]:
return volume_1, volume_2

def make_random_rate_of_turn() -> float:
"""Return a random float - range is pure guesswork"""
return round(random.uniform(10.00, 20.00), 2)
"""Return a random rate of turn between 1 and 5 degrees per second"""
return round(random.uniform(1.00, 5.00), 2)

def make_random_radius_of_turn() -> float:
"""Return a random radius of turn between 5 and 15 nautical miles"""
return round(random.uniform(5.00, 15.00), 2)

def make_random_coordination(callsign=None, the_datetime=None) -> Coordination:
"""Return a coordination object with random parameters"""
Expand Down Expand Up @@ -552,6 +560,15 @@ def make_action(kind: str) -> Action:
value = make_random_heading()
if kind == "heading_turn_segment":
value = make_random_heading()
if kind == "track_segment":
value = make_random_ground_track()
if kind == "fixed_radius_turn":
value = (make_random_radius_of_turn(), make_random_ground_track())
if kind == "fixed_rate_turn":
value = (make_random_rate_of_turn(), make_random_heading())
if kind == "intercept_radial":
intercept_type = random.choice(["immediate", "when_ready", "target_fix"])
value = (intercept_type, make_random_fix_name(), make_random_ground_track())
if kind == "message":
value = make_random_message()
agent = random.choice(["Smith", "Bond", "Powers", "J"])
Expand Down
Loading