Goal Observation

.github/workflows/pre-commit.yml

@@ -0,0 +1,21 @@
+# https://pre-commit.com
+# This GitHub Action assumes that the repo contains a valid .pre-commit-config.yaml file
+name: pre-commit
+on:
+  pull_request:
+  push:
+    branches: [master]
+
+permissions:
+  contents: read # to fetch code (actions/checkout)
+
+jobs:
+  pre-commit:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
+      - run: python -m pip install pre-commit
+      - run: python -m pre_commit --version
+      - run: python -m pre_commit install
+      - run: python -m pre_commit run --all-files

.pre-commit-config.yaml

@@ -0,0 +1,24 @@
+# See https://pre-commit.com for more information
+# See https://pre-commit.com/hooks.html for more hooks
+repos:
+  - repo: https://github.com/PyCQA/flake8
+    rev: 6.0.0
+    hooks:
+      - id: flake8
+        args:
+          - '--per-file-ignores=**/__init__.py:F401,F403,E402'
+          - --ignore=E203,W503,E741,E731
+          - --max-complexity=30
+          - --max-line-length=456
+          - --show-source
+          - --statistics
+  - repo: https://github.com/PyCQA/isort
+    rev: 5.12.0
+    hooks:
+      - id: isort
+        args: ["--profile", "black"]
+        exclude: "__init__.py"
+  - repo: https://github.com/python/black
+    rev: 23.3.0
+    hooks:
+      - id: black

flyer_env/envs/common/abstract.py

@@ -180,7 +180,7 @@ def reset(self,
         self.time = 0.0
         self.steps = 0
         self.done = False
-        self._reset(seed)
+        self._reset()
 
         # Second, to link the obs and actions to the vehicles once the scene is created
         self.define_spaces()

flyer_env/envs/common/action.py

@@ -168,6 +168,58 @@ def act(self, action: np.ndarray) -> None:
         self.last_action = action 
 
 
+class HeadingAction(ActionType):
+
+    HEADING_RANGE = (-np.pi, np.pi)
+
+    """
+    A continuous action space with a fixed altitude and speed.
+    Controls are only [aileron].
+    """
+
+    def __init__(self,
+                 env: 'AbstractEnv',
+                 heading_range: Optional[Tuple[float, float]] = None,
+                 powered: bool = True,
+                 clip: bool = True,
+                 **kwargs) -> None:
+        """
+        Create a continuous laterally constrained action space
+        """
+        super().__init__(env)
+
+        self.heading_range = heading_range if heading_range else self.HEADING_RANGE
+        self.powered = powered
+        self.clip = clip
+        self.size = 1
+        self.last_action = np.zeros(self.size)
+
+    def space(self) -> spaces.Box:
+        return spaces.Box(self.heading_range[0], self.heading_range[1], shape=(self.size,), dtype=np.float32)
+
+    @property
+    def vehicle_class(self) -> Callable:
+        return functools.partial(ControlledAircraft)
+
+    def act(self, action: np.ndarray) -> None:
+        """
+        Apply the action to the controlled vehicle
+
+        :param action: action array with [sine, cosine] mapped between ranges
+        """
+
+        if self.clip:
+            action = np.clip(action, self.heading_range[0], self.heading_range[1])
+
+        if self.powered:
+            self.controlled_vehicle.act({
+               'heading': action,
+               'alt': -1000.0,
+               'speed': 80.0
+            })
+        self.last_action = action
+
+
 class ControlledAction(ActionType):
     """
     An action that controls the aircraft using a PID controller to track towards the target.
@@ -328,6 +380,8 @@ def action_factory(env: 'AbstractEnv', config: dict) -> ActionType:
         return ContinuousAction(env, **config)
     elif config["type"] == "LongitudinalAction":
         return LongitudinalAction(env, **config)
+    elif config["type"] == "HeadingAction":
+        return HeadingAction(env, **config)
     elif config["type"] == "ControlledAction":
         return ControlledAction(env, **config)
     elif config["type"] == "PursuitAction":

flyer_env/envs/common/observation.py

@@ -101,6 +101,40 @@ def observe(self) -> np.ndarray:
         obs[0, 2] = self.goal[2] - obs[0, 2]
         return obs.astype(self.space().dtype)
 
+class LateralTrajectoryObservation(ObservationType):
+
+    """
+    Observe dynamics of vehicle relative to goal position, restricted to horizontal plane
+    ONLY FOR USE WITH TRAJECTORY ENV
+    """
+
+    FEATURES: List[str] = ['x', 'y', 'u', 'v', 'yaw']
+
+    def __init__(self,
+                 env: "AbstractEnv",
+                 features: List[str] = None,
+                 vehicles_count: int = 1,
+                 features_range: Dict[str, List[float]] = None,
+                 **kwargs: dict) -> None:
+
+        super().__init__(env)
+        self.features = features or self.FEATURES
+        self.vehicles_count = vehicles_count
+        self.features_range = features_range
+        if hasattr(env, "goal"):
+            self.goal = env.goal
+
+    def space(self) -> spaces.Space:
+        return spaces.Box(shape=(self.vehicles_count, len(self.features)), low=-np.inf, high=np.inf, dtype=np.float32)
+
+    def observe(self) -> np.ndarray:
+
+        df = pd.DataFrame.from_records([self.observer_vehicle.dict])[self.features]
+        df = df[self.features]
+        obs = df.values.copy()
+        obs[0, 0] = self.goal[0] - obs[0, 0]
+        obs[0, 1] = self.goal[1] - obs[0, 1]
+        return obs.astype(self.space().dtype)
 
 class ControlObservation(ObservationType):
 
@@ -163,15 +197,89 @@ def observe(self) -> np.ndarray:
         obs = df.values.copy()
         return obs.astype(self.space().dtype)
 
+class DynamicGoalObservation(DynamicObservation):
+
+    def __init__(self, 
+                 env: "AbstractEnv",
+                 **kwargs: dict) -> None:
+        super().__init__(env, **kwargs)
+        if hasattr(env, "goal"):
+            self.goal = env.goal
+
+    def space(self) -> spaces.Space:
+        try:
+            obs = self.observe()
+            return spaces.Dict(dict(
+                desired_goal=spaces.Box(-np.inf, np.inf, shape=obs["desired_goal"].shape, dtype=np.float64),
+                achieved_goal=spaces.Box(-np.inf, np.inf, shape=obs["achieved_goal"].shape, dtype=np.float64),
+                observation=spaces.Box(-np.inf, np.inf, shape=obs["observation"].shape, dtype=np.float64)
+            ))
+        except AttributeError:
+            return spaces.Space()
+
+    def observe(self) -> Dict[str, np.ndarray]:
+        df = pd.DataFrame.from_records([self.observer_vehicle.dict])[self.features]
+        df = df[self.features]
+        obs = df.values.copy()
+        # obs = obs.astype(self.space().dtype)
+        obs = OrderedDict([
+            ("observation", obs[0]),
+            ("achieved_goal", obs[0][0:3]),
+            ("desired_goal", self.goal)
+        ])
+        return obs
+
+
+class LateralGoalObservation(DynamicObservation):
+
+    FEATURES: List[str] = ['x', 'y',  'u', 'v', 'yaw']
+
+    def __init__(self,
+                 env: "AbstractEnv",
+                 features: List[str] = None,
+                 **kwargs: dict) -> None:
+        super().__init__(env, **kwargs)
+        self.features = features or self.FEATURES
+        if hasattr(env, "goal"):
+            self.goal = env.goal
+
+    def space(self) -> spaces.Space:
+        try:
+            obs = self.observe()
+            return spaces.Dict(dict(
+                desired_goal=spaces.Box(-np.inf, np.inf, shape=obs["desired_goal"].shape, dtype=np.float64),
+                achieved_goal=spaces.Box(-np.inf, np.inf, shape=obs["achieved_goal"].shape, dtype=np.float64),
+                observation=spaces.Box(-np.inf, np.inf, shape=obs["observation"].shape, dtype=np.float64)
+            ))
+        except AttributeError:
+            return spaces.Space()
+
+    def observe(self) -> Dict[str, np.ndarray]:
+        df = pd.DataFrame.from_records([self.observer_vehicle.dict])[self.features]
+        df = df[self.features]
+        obs = df.values.copy()
+        obs = OrderedDict([
+            ("observation", obs[0]),
+            ("achieved_goal", obs[0][0:2]),
+            ("desired_goal", self.goal[0:2])
+        ])
+        return obs
+
 
 def observation_factory(env: "AbstractEnv", config: dict) -> ObservationType:
     if config["type"] == "Dynamics" or config["type"] == "dynamics":
         return DynamicObservation(env, **config)
     elif config["type"] == "Trajectory" or config["type"] == "trajectory":
         return TrajectoryObservation(env, **config)
+    elif config["type"] == "LateralTrajectory" or config["type"] == "lateral_trajectory":
+        return LateralTrajectoryObservation(env, **config)
     elif config["type"] == "Control" or config["type"] == "control":
         return ControlObservation(env, **config)
     elif config["type"] == "Longitudinal" or config["type"] == "longitudinal":
         return LongitudinalObservation(env, **config)
+    elif config["type"] == "Goal" or config["type"] == "goal" or config["type"] == "DynamicGoal":
+        return DynamicGoalObservation(env, **config)
+    elif config["type"] == "LateralGoal" or config["type"] == "lateral_goal":
+        return LateralGoalObservation(env, **config)
     else:
         raise ValueError("Unknown observation type")

flyer_env/envs/control_env.py

@@ -40,19 +40,21 @@ def default_config(cls) -> dict:
         })
         return config
 
-    def _reset(self, seed) -> None:
-        if not seed: seed = 1
-        self._create_world(seed)
+    def _reset(self) -> None:
+
+        self.np_random = np.random.RandomState()
+        self._create_world()
         self._create_vehicles()
 
-    def _create_world(self, seed) -> None:
+    def _create_world(self) -> None:
         """Create the world map"""
         self.world = World()
         path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "assets")
         self.world.assets_dir = path
         path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "terrain_data")
         self.world.terrain_data_dir = path
-        self.world.create_map(seed, area=self.config["area"])
+        world_seed = self.np_random.randint(100)  # set 100 possible seeds by default 
+        self.world.create_map(world_seed, area=self.config["area"])
 
     def _create_vehicles(self) -> None:
         """Create an aircraft to fly around the world"""