diff --git a/docs/source/release_notes.rst b/docs/source/release_notes.rst index af8aa696..4121b16c 100644 --- a/docs/source/release_notes.rst +++ b/docs/source/release_notes.rst @@ -78,6 +78,7 @@ Version 1.3.0 * Allow for the Vizard output path to be specified as a .bin file instead of just a directory. * Use Vizard 2.3.1 locations for visualization; results in significantly smaller output files. +* Add an example script demonstrating target location uncertainty modeling with uncertainty growth over time: `Target Location Uncertainty `_ example. * Allow for a simpler Earth model to be used in Vizard by setting ``use_simple_earth=True`` in the Vizard settings dictionary. This is helpful for when visualizing may Earth-fixed targets. diff --git a/examples/_default.rst b/examples/_default.rst index ca7d27a5..a0be6028 100644 --- a/examples/_default.rst +++ b/examples/_default.rst @@ -23,6 +23,7 @@ Earth Observation cloud_environment_with_reimaging hybrid_image_quality aeos + target_location_uncertainty RSO Missions ~~~~~~~~~~~~ diff --git a/examples/target_location_uncertainty.ipynb b/examples/target_location_uncertainty.ipynb new file mode 100644 index 00000000..85d4cce3 --- /dev/null +++ b/examples/target_location_uncertainty.ipynb @@ -0,0 +1,729 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Target Location Uncertainty\n", + "This tutorial demonstrates how to model uncertainty in target locations when imaging ground targets using a simple single-satellite BSK-RL environment. It also shows how target location uncertainty can grow over time when a target is not imaged.\n", + "\n", + "## Loading Modules" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from collections.abc import Iterable\n", + "from typing import Any, ClassVar, Union\n", + "\n", + "import numpy as np\n", + "from Basilisk.architecture import bskLogging\n", + "from Basilisk.utilities import macros, orbitalMotion\n", + "from scipy.stats import norm\n", + "\n", + "from bsk_rl import act, obs, sats\n", + "from bsk_rl.data.unique_image_data import UniqueImageData, UniqueImageReward\n", + "from bsk_rl.gym import SatelliteTasking\n", + "from bsk_rl.scene.targets import UniformTargets\n", + "from bsk_rl.sim import dyn, fsw\n", + "from bsk_rl.sim.fsw import action\n", + "from bsk_rl.utils.orbital import random_orbit\n", + "\n", + "bskLogging.setDefaultLogLevel(bskLogging.BSK_WARNING)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Making a Scenario with Location Uncertain Targets\n", + "\n", + "To account for location uncertainty in the simulation process, the following parameters are assigned to each target using UniformTargets as a base:\n", + "\n", + "* `search_area` represents a square region on the surface within which the estimated location is the center of the area.\n", + "\n", + "* `acquiring_speed` represents the apparent ground-track speed, sampled uniformly to reflect orbital geometry and latitude effects.\n", + "\n", + "* `scan_width` represents the fixed width of the imaging swath of the satellite.\n", + "\n", + "* `search_duration` represents the total imaging duration, proportional to the number of sensor swaths required to cover the area.\n", + "\n", + "* `image_time` represents the true imaging time within the duration window, drawn from a Gaussian distribution centered at the mid-point of $T_i$ ($\\mu_i$) with standard deviation $0.3T_i$ ($\\sigma_i$), truncated to nonnegative values. Approximately \\(4.8\\%\\) of samples occur after the window ends.\n", + "\n", + "* `speed` represents the target's ground speed, sampled uniformly to reflect typical ship speeds and fixed for each target." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "class AreaTargets(UniformTargets):\n", + " def regenerate_targets(self) -> None:\n", + " super().regenerate_targets()\n", + " for target in self.targets:\n", + " side_length = np.random.uniform(10.0, 100.0) * 1e3 # m\n", + " target.search_area = side_length**2 # m^2\n", + " target.acquiring_speed = np.random.uniform(3.0, 6.0) * 1e3 # m/s\n", + " scan_width = 20.0 * 1e3 # m\n", + " search_duration = (\n", + " side_length * np.ceil(side_length / scan_width) / target.acquiring_speed\n", + " )\n", + " target.search_duration = search_duration # s\n", + "\n", + " mean_duration = search_duration / 2.0\n", + " std_duration = search_duration * 0.3\n", + " image_time = np.random.normal(loc=mean_duration, scale=std_duration)\n", + " image_time = max(0.0, image_time)\n", + " target.image_time = image_time # s\n", + "\n", + " target.speed = np.random.uniform(22e3 / 3600, 46e3 / 3600) # m/s" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Making a Reward Function That Accounts for Target Uncertainty\n", + "\n", + "When targets have location uncertainty, imaging the highest-priority target is not always the optimal choice. If a target is not imaged for some time, its uncertainty grows, which can reduce future imaging effectiveness. Therefore, the reward function should balance target priority and uncertainty. One approach is to multiply the target priority by its scaled uncertainty. Alternative reward formulations are explored in a forthcoming publication.\n", + "The `UncertainMultiplyReward` is built on the [UniqueImageReward](../api_reference/data/index.rst) base class and defines the reward function as\n", + "\n", + "$$\n", + "R(s, a, s') = \n", + "\\begin{cases}\n", + "p_n \\cdot \\tilde{u}_n, & \\text{if } a = a_{\\text{image}, n} \\text{ and } t_{\\text{search}} = t_n^{\\ast} \\\\\n", + "0, & \\text{otherwise}\n", + "\\end{cases}\n", + "$$\n", + "\n", + "where $p_n$ is the target priority, $\\tilde{u}_n$ is the scaled target uncertainty, $t{\\text{search}}$ is the time spent searching for the target, $t_n^{\\ast}$ is the true imaging duration, and $a_{\\text{image}, n}$ denotes the imaging action for the selected target $n$.\n", + "The scaled uncertainty $\\tilde{u}_n$ is calculated by normalizing the target's search area by the maximum possible search area at the end of a three-orbit simulation (U_NORM²). The tunable parameter U_SCALE controls the relative contribution of normalized uncertainty to the reward.\n", + "Unlike `UniqueImageReward`, this class removes the imaging list filter, allowing the same target to be imaged multiple times." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from typing import TYPE_CHECKING\n", + "\n", + "if TYPE_CHECKING: # pragma: no cover\n", + " from bsk_rl.sats import Satellite\n", + " from bsk_rl.scene.targets import Target\n", + "\n", + "ALTITUDE = 800 # altitude in km\n", + "T_ORBIT = ( # one orbital period in seconds\n", + " 2\n", + " * np.pi\n", + " * np.sqrt((orbitalMotion.REQ_EARTH + ALTITUDE) ** 3 / orbitalMotion.MU_EARTH)\n", + ")\n", + "U_NORM = (\n", + " 100.0 * 1e3 + 46e3 / 3600 * T_ORBIT * 3\n", + ") # max search area in meters, for normalization\n", + "U_SCALE = 0.1 # scale factor for uncertainty reward\n", + "\n", + "\n", + "def _uncertainty(target: \"Target\") -> float:\n", + " return target.search_area / (U_NORM**2)\n", + "\n", + "\n", + "def scaled_clipped_uncertainty(target: \"Target\", scale: float = 1.0) -> float:\n", + " if scale <= 0.0:\n", + " raise ValueError(\"reward_uncertainty_scale must be positive\")\n", + " return min(_uncertainty(target) / scale, 1.0)\n", + "\n", + "\n", + "class UncertainMultiplyReward(UniqueImageReward):\n", + " def create_data_store(self, satellite: \"Satellite\") -> None:\n", + " \"\"\"Create a data store for a satellite.\n", + "\n", + " Args:\n", + " satellite: Satellite to create a data store for.\n", + " \"\"\"\n", + " satellite.data_store = self.data_store_type(\n", + " satellite,\n", + " initial_data=self.initial_data(satellite),\n", + " **self.data_store_kwargs,\n", + " )\n", + " self.cum_reward[satellite.name] = 0.0\n", + "\n", + " def calculate_reward(\n", + " self, new_data_dict: dict[str, UniqueImageData]\n", + " ) -> dict[str, float]:\n", + " \"\"\"Reward images of targets based on their priority and current uncertainty.\n", + "\n", + " Targets may be imaged multiple times; each image earns reward based on\n", + " ``self.reward_fn(target.priority * scaled_uncertainty)``, with the reward\n", + " shared among all simultaneous images of the same target.\n", + "\n", + " Args:\n", + " new_data_dict: Record of new images for each satellite\n", + "\n", + " Returns:\n", + " reward: Cumulative reward across satellites for one step\n", + " \"\"\"\n", + " reward = {}\n", + " imaged_counts = {}\n", + " for new_data in new_data_dict.values():\n", + " for target in new_data.imaged:\n", + " if target not in imaged_counts:\n", + " imaged_counts[target] = 0\n", + " imaged_counts[target] += 1\n", + "\n", + " for sat_id, new_data in new_data_dict.items():\n", + " reward[sat_id] = 0.0\n", + " for target in new_data.imaged:\n", + " uncertainty = scaled_clipped_uncertainty(target, scale=U_SCALE)\n", + " reward[sat_id] += (\n", + " self.reward_fn(target.priority * uncertainty)\n", + " / imaged_counts[target]\n", + " )\n", + " return reward" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Configuring the Satellite to Have Access to Target Uncertainty Information\n", + "\n", + "The satellite has observations and actions associated with it that are relevant to the decision-making process. The observation space can be modified to include information about the targets and the uncertainty which allows better informed decision-making.\n", + "\n", + "* [Observations](../api_reference/obs/index.rst): \n", + " - SatProperties: Body angular velocity, instrument pointing direction, body position, and body velocity in [dynamics model](../api_reference/sim/dyn/index.rst).\n", + " - OpportunityProperties: Target priority, pointing error, pointing rate error, time until the target observation window opens and closes, scaled uncertainty, completion probability, and expected reward for each of the next 32 targets. The completion probability is calculated based on the available observation window duration and slew time, representing the probability of completing the search over the entire search area. The slew time is estimated using a regression model developed from simulation data.\n", + " - Density: Upcoming target request density. \n", + "* [Actions](../api_reference/act/index.rst):\n", + " - RetaskImage: An extension of act.Image that allows the satellite to image the same target consecutively while selecting from the next 32 targets.\n", + "* [Dynamics model](../api_reference/sim/dyn/index.rst): FullFeaturedDynModel is used.\n", + "* [Flight software model](../api_reference/sim/fsw/index.rst): SteeringImagerFSWModel used as the base model and modified for duration-based imaging." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "class Density(obs.Observation):\n", + " def __init__(\n", + " self,\n", + " interval_duration=60 * 3,\n", + " intervals=10,\n", + " norm=3,\n", + " ):\n", + " self.satellite: sats.ImagingSatellite\n", + " super().__init__()\n", + " self.interval_duration = interval_duration\n", + " self.intervals = intervals\n", + " self.norm = norm\n", + "\n", + " def get_obs(self):\n", + " if self.intervals == 0:\n", + " return []\n", + "\n", + " self.satellite.calculate_additional_windows(\n", + " self.simulator.sim_time\n", + " + (self.intervals + 1) * self.interval_duration\n", + " - self.satellite.window_calculation_time\n", + " )\n", + " soonest = self.satellite.upcoming_opportunities_dict(types=\"target\")\n", + " rewards = np.array([opportunity.priority for opportunity in soonest])\n", + " times = np.array([opportunities[0][1] for opportunities in soonest.values()])\n", + " time_bins = np.floor((times - self.simulator.sim_time) / self.interval_duration)\n", + " densities = [sum(rewards[time_bins == i]) for i in range(self.intervals)]\n", + " return np.array(densities) / self.norm\n", + "\n", + "\n", + "def target_angle(sat_r_BN_P, sat_c_hat_P, opp_r_LP_P):\n", + " vector_target_spacecraft_P = opp_r_LP_P - sat_r_BN_P\n", + " vector_target_spacecraft_P_hat = vector_target_spacecraft_P / np.linalg.norm(\n", + " vector_target_spacecraft_P\n", + " )\n", + " return np.arccos(np.dot(vector_target_spacecraft_P_hat, sat_c_hat_P))\n", + "\n", + "\n", + "def target_angle_rate(sat_r_BN_P, sat_v_BN_P, sat_omega_BP_P, opp_r_LP_P):\n", + " r_BN_P = sat_r_BN_P\n", + " v_BN_P = sat_v_BN_P\n", + " r_LP_P = opp_r_LP_P\n", + " omega_BP_P = sat_omega_BP_P\n", + " omega_CP_ref = (\n", + " omega_BP_P\n", + " - np.cross(v_BN_P, r_LP_P - r_BN_P) / np.linalg.norm(r_LP_P - r_BN_P) ** 2\n", + " )\n", + " return np.linalg.norm(omega_CP_ref)\n", + "\n", + "\n", + "def regression_model(\n", + " tgt_angle,\n", + " tgt_angle_rate,\n", + " x_0=0.4320,\n", + " x_1=101.1396,\n", + " x_2=641.8838,\n", + " x_3=-45.9458,\n", + " x_4=-7681.1881,\n", + " x_5=-610.2113,\n", + "):\n", + " slew_time = (\n", + " x_0\n", + " + x_1 * tgt_angle\n", + " + x_2 * tgt_angle_rate\n", + " + x_3 * tgt_angle**2\n", + " + x_4 * tgt_angle_rate**2\n", + " + x_5 * tgt_angle * tgt_angle_rate\n", + " )\n", + " return np.clip(slew_time, 1e-3, None)\n", + "\n", + "\n", + "def completion_probability(t_now, t_open, t_close, slew_time, search_duration):\n", + " t_start = max(t_now + slew_time, t_open)\n", + " t_avail = t_close - t_start\n", + "\n", + " if t_avail <= 0:\n", + " return 0.0\n", + "\n", + " if search_duration <= 0:\n", + " return 1.0\n", + "\n", + " mu = search_duration / 2.0\n", + " sigma = max(search_duration * 0.3, 1e-6)\n", + "\n", + " return float(np.clip(norm.cdf(t_avail, loc=mu, scale=sigma), 0.0, 1.0))\n", + "\n", + "\n", + "def opportunity_completion_probability(sat, opp):\n", + " target = opp[\"object\"]\n", + " r_LP_P = target.r_LP_P\n", + " tgt_angle = target_angle(sat.dynamics.r_BN_P, sat.fsw.c_hat_P, r_LP_P)\n", + " tgt_angle_rate = target_angle_rate(\n", + " sat.dynamics.r_BN_P,\n", + " sat.dynamics.v_BN_P,\n", + " sat.dynamics.omega_BP_P,\n", + " r_LP_P,\n", + " )\n", + " slew_time = regression_model(tgt_angle, tgt_angle_rate)\n", + "\n", + " t_open, t_close = opp[\"window\"]\n", + " return completion_probability(\n", + " sat.simulator.sim_time, t_open, t_close, slew_time, target.search_duration\n", + " )\n", + "\n", + "\n", + "def reward_value(target: \"Target\", scale: float = 1.0) -> float:\n", + " \"\"\"Calculate the reward value for a target.\n", + "\n", + " Args:\n", + " target: Target to calculate reward for.\n", + " scale: Scale factor for uncertainty reward.\n", + "\n", + " Returns:\n", + " The reward value for the target.\n", + " \"\"\"\n", + " return target.priority * scaled_clipped_uncertainty(target, scale=scale)\n", + "\n", + "\n", + "class RetaskImage(act.Image):\n", + " \"\"\"Image action that always reissues the task, even for same target consecutively.\"\"\"\n", + "\n", + " def image(\n", + " self,\n", + " target: Union[int, \"Target\", str],\n", + " prev_action_key: str | None = None,\n", + " max_duration: float | None = None,\n", + " ) -> str:\n", + " target = self.satellite.parse_target_selection(target)\n", + " self.satellite.task_target_for_imaging(target, max_duration=max_duration)\n", + " return target.id\n", + "\n", + "\n", + "class CustomFswModel(fsw.SteeringImagerFSWModel):\n", + " @action\n", + " def action_image(\n", + " self,\n", + " r_LP_P: Iterable[float],\n", + " data_name: str,\n", + " acquisitionTime: float,\n", + " allowedTime: float,\n", + " ) -> None:\n", + " \"\"\"Attempt to image a target at a location.\n", + "\n", + " This action sets the target attitude to one tracking a ground location. If the\n", + " target is within the imaging constraints for the specified time, an image will be taken and stored in\n", + " the data buffer. The instrument power sink will be active as long as the task is\n", + " enabled.\n", + "\n", + " Args:\n", + " r_LP_P: [m] Planet-fixed planet relative target location.\n", + " data_name: Data buffer to store image data to.\n", + " acquisitionTime: [s] Time required to acquire the image.\n", + " allowedTime: [s] Maximum time allowed to attempt to acquire the image.\n", + " \"\"\"\n", + " self.insControl.controllerStatus = 1\n", + " self.dynamics.instrumentPowerSink.powerStatus = 1\n", + " self.dynamics.imagingTarget.r_LP_P_Init = r_LP_P\n", + " self.dynamics.instrument.nodeDataName = data_name\n", + " self.insControl.imaged = 0\n", + " self.insControl.useDurationImaging = True\n", + " self.insControl.acquisitionTime = macros.sec2nano(acquisitionTime)\n", + " self.insControl.allowedTime = macros.sec2nano(allowedTime)\n", + " self.simulator.enableTask(self.LocPointTask.name + self.satellite.name)\n", + "\n", + "\n", + "class CustomSatComposed(sats.ImagingSatellite):\n", + " observation_spec: ClassVar[list[obs.Observation]] = [\n", + " obs.SatProperties(\n", + " dict(prop=\"omega_BN_B\", norm=0.03),\n", + " dict(prop=\"c_hat_H\"),\n", + " dict(prop=\"r_BN_P\", norm=orbitalMotion.REQ_EARTH * 1e3),\n", + " dict(prop=\"v_BN_P\", norm=7616.5),\n", + " ),\n", + " Density(intervals=20, norm=5),\n", + " obs.OpportunityProperties(\n", + " dict(prop=\"priority\"),\n", + " dict(prop=\"r_LB_H\", norm=orbitalMotion.REQ_EARTH * 1e3),\n", + " dict(prop=\"target_angle\", norm=np.pi / 2),\n", + " dict(prop=\"target_angle_rate\", norm=0.03),\n", + " dict(prop=\"opportunity_open\", norm=300.0),\n", + " dict(prop=\"opportunity_close\", norm=300.0),\n", + " dict(\n", + " prop=\"search_area\",\n", + " fn=lambda sat, opp: opp[\"object\"].search_area,\n", + " norm=(U_SCALE * (U_NORM) ** 2),\n", + " ),\n", + " dict(prop=\"completion_probability\", fn=opportunity_completion_probability),\n", + " dict(\n", + " prop=\"reward_value\",\n", + " fn=lambda sat, opp: reward_value(opp[\"object\"], scale=U_SCALE),\n", + " ),\n", + " type=\"target\",\n", + " n_ahead_observe=32,\n", + " ),\n", + " ]\n", + "\n", + " action_spec: ClassVar[list[act.Action]] = [\n", + " RetaskImage(n_ahead_image=32),\n", + " ]\n", + "\n", + " dyn_type = dyn.FullFeaturedDynModel\n", + " fsw_type = CustomFswModel\n", + "\n", + " def task_target_for_imaging(\n", + " self, target: \"Target\", max_duration: float | None = None\n", + " ):\n", + " \"\"\"Task the satellite to image a target.\n", + "\n", + " Args:\n", + " target: Selected target\n", + " max_duration: [s] Maximum duration to wait for imaging. If None, wait until\n", + " the end of the target's access window.\n", + " \"\"\"\n", + " msg = f\"{target} tasked for imaging, image_time: {target.image_time:.1f}, duration: {target.search_duration:.1f}\"\n", + " self.logger.info(msg)\n", + " self.fsw.action_image(\n", + " target.r_LP_P, self.buffer_name, target.image_time, target.search_duration\n", + " ) # times in seconds\n", + " self.enable_target_window(target, max_duration=max_duration)\n", + " self.draw_imaging_line(target)\n", + " self.latest_target = target" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "When instantiating a satellite, these parameters can be overriden with a constant or \n", + "rerandomized every time the environment is reset using the ``sat_args`` dictionary." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "dataStorageCapacity = (\n", + " 20 * 8e6 * 100\n", + ") # Large storage to avoid filling up in three orbits\n", + "batteryStorageCapacity = (\n", + " 1000.0 * 3600 * 2\n", + ") # Large storage to avoid battery running out in three orbits\n", + "\n", + "sat_args = CustomSatComposed.default_sat_args(\n", + " oe=lambda: random_orbit(\n", + " alt=ALTITUDE, # 800 km altitude\n", + " i=45, # 45 degrees inclination\n", + " ),\n", + " imageAttErrorRequirement=0.01,\n", + " imageRateErrorRequirement=0.01,\n", + " batteryStorageCapacity=batteryStorageCapacity,\n", + " storedCharge_Init=lambda: np.random.uniform(0.4, 1.0) * batteryStorageCapacity,\n", + " u_max=0.4,\n", + " K1=0.25,\n", + " K3=3.0,\n", + " servo_P=150 / 5,\n", + " nHat_B=np.array([0, 0, -1]),\n", + " omega_max=np.radians(5.0), # Maximum rate command in degrees per second\n", + " imageTargetMinimumElevation=np.arctan(800 / 500), # 58 degrees minimum elevation\n", + " rwBasePower=20,\n", + " maxWheelSpeed=1500,\n", + " storageInit=lambda: np.random.randint(\n", + " 0 * dataStorageCapacity,\n", + " 0.01 * dataStorageCapacity,\n", + " ), # Initialize storage use close to zero\n", + " wheelSpeeds=lambda: np.random.uniform(\n", + " -1, 1, 3\n", + " ), # Initialize reaction wheel speeds close to zero\n", + " dataStorageCapacity=dataStorageCapacity,\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Making Target Uncertainty Growing Environment\n", + "\n", + "To model the growth of target location uncertainty over time, `SatelliteUncertaintyTargets` extends [SatelliteTasking](../api_reference/index.rst) and overrides its `step()` method. The search-region side length grows according to the elapsed simulation time and the target's speed, which is assigned in `AreaTargets`:\n", + "\n", + "$$\n", + "s_{\\mathrm{new}} = \\sqrt{A_{\\mathrm{old}}} + v\\Delta t,\n", + "\\qquad\n", + "A_{\\mathrm{new}} = s_{\\mathrm{new}}^2.\n", + "$$\n", + "\n", + "Here, $A_{\\mathrm{old}}$ and $A_{\\mathrm{new}}$ are the old and updated search areas in m², $s_{\\mathrm{new}}$ is the updated side length in meters, $v$ is the target's speed in m/s, and $\\Delta t$ is the elapsed simulation time in seconds, returned as `info[\"d_ts\"]`.\n", + "\n", + "The update order and modeling assumptions are:\n", + "\n", + "- The parent `step()` computes the reward **before** uncertainty is updated, using the target uncertainty available at the start of the step.\n", + "- After every step, search areas and search durations are updated, and acquisition times (`image_time`) are sampled again for all known targets.\n", + "- If imaging succeeds, the selected target's search area, search duration, and acquisition time are then reset to zero at the step's end. The returned observation is recomputed after these updates.\n", + "- This implementation models growing uncertainty around fixed target coordinates. The target speed controls uncertainty growth; it does not move the target coordinates.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "class SatelliteUncertaintyTargets(SatelliteTasking):\n", + " def step(self, action) -> tuple[Any, float, bool, bool, dict[str, Any]]:\n", + " \"\"\"Take a single-satellite action, run parent step, and update target uncertainties.\"\"\"\n", + " sat = self.satellites[0]\n", + " target_selected = sat.parse_target_selection(action)\n", + " old_stored_data = np.array(\n", + " sat.dynamics.storageUnit.storageUnitDataOutMsg.read().storedData,\n", + " dtype=float,\n", + " )\n", + " obs, reward, terminated, truncated, info = super().step(action)\n", + " d_ts = info[\"d_ts\"]\n", + " new_stored_data = np.array(\n", + " sat.dynamics.storageUnit.storageUnitDataOutMsg.read().storedData,\n", + " dtype=float,\n", + " )\n", + " imaged_this_step = np.any(new_stored_data - old_stored_data > 0.0)\n", + "\n", + " targets = sat.data_store.data.known\n", + " area_factor = 20e3 # scan width in meters (20 km)\n", + "\n", + " # Compute all new values in vectorized form\n", + " search_areas = np.array([t.search_area for t in targets])\n", + " side_lengths = np.sqrt(search_areas)\n", + " acquiring_speeds = np.array([t.acquiring_speed for t in targets])\n", + " speeds = np.array([t.speed for t in targets])\n", + "\n", + " new_side_lengths = side_lengths + speeds * d_ts\n", + " new_search_areas = new_side_lengths**2\n", + " new_search_durations = (\n", + " new_side_lengths\n", + " * np.ceil(new_side_lengths / area_factor)\n", + " / acquiring_speeds\n", + " )\n", + "\n", + " means = new_search_durations / 2.0\n", + " stds = new_search_durations * 0.3\n", + " new_image_times = np.maximum(0.0, np.random.normal(loc=means, scale=stds))\n", + "\n", + " # Update the target information\n", + " for t, area, dur, time in zip(\n", + " targets, new_search_areas, new_search_durations, new_image_times\n", + " ):\n", + " t.search_area = area\n", + " t.search_duration = dur\n", + " t.image_time = time\n", + "\n", + " # Reset the imaged target's uncertainty\n", + " if imaged_this_step:\n", + " target_selected.search_area = 0.0\n", + " target_selected.search_duration = 0.0\n", + " target_selected.image_time = 0.0\n", + "\n", + " sat.observation_builder.obs_dict_cache = None\n", + " obs = sat.get_obs()\n", + "\n", + " return obs, reward, terminated, truncated, info" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Initializing and Interacting with the Environment\n", + "For this example, the custom environment created above is used along with passing the satellite that is configured, the environment takes a [scenario](../api_reference/scene/index.rst), which defines the environment the satellite is acting in, and a [rewarder](../api_reference/data/index.rst), which defines how data collected from the scenario is rewarded. The filter is added to prevent the satellite from reimaging recently observed targets with small search areas." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "threshold = (10e3) ** 2 # threshold for search area in m^2, (10 km)^2\n", + "\n", + "\n", + "def search_area_filter(opportunity, threshold=threshold):\n", + " if opportunity[\"type\"] == \"target\":\n", + " return opportunity[\"object\"].search_area > threshold\n", + " return True\n", + "\n", + "\n", + "satellite = CustomSatComposed(\"EO\", sat_args)\n", + "satellite.add_access_filter(search_area_filter, types=\"target\")\n", + "n_targets = (1000, 10000)\n", + "\n", + "env = SatelliteUncertaintyTargets(\n", + " satellite=satellite,\n", + " scenario=AreaTargets(n_targets=n_targets),\n", + " rewarder=UncertainMultiplyReward(),\n", + " sim_rate=0.5,\n", + " max_step_duration=T_ORBIT, # Duration long enough to prevent interruption of the imaging action\n", + " time_limit=T_ORBIT * 3, # three orbits\n", + " log_level=\"INFO\",\n", + " failure_penalty=0.0,\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "First, reset the environment. It is possible to specify the seed when resetting the environment." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "observation, info = env.reset(seed=1)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "It is possible to print out the actions and observations. The composed satellite [action_description](../api_reference/sats/index.rst) returns a human-readable action map. Each satellite has the same action space and similar observation space." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Actions:\", env.satellites[0].action_description, \"\\n\")\n", + "print(\"States:\", env.satellites[0].observation_description, \"\\n\")\n", + "\n", + "# Using the composed satellite features also provides a human-readable state:\n", + "for satellite in env.satellites:\n", + " for k, v in satellite.observation_builder.obs_dict().items():\n", + " print(f\"{k}: {v}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Then, run the simulation until timeout." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "while True:\n", + " action_step = np.random.randint(0, 32) # Random action\n", + "\n", + " observation, reward, terminated, truncated, info = env.step(action_step)\n", + "\n", + " if terminated or truncated:\n", + " print(\"Episode complete.\")\n", + " break" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "After running the simulation, we can check the total reward, the number of unique targets that were imaged at least once, and the number of duplicated (re-imaged) targets collected by the rewarder." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Total reward:\", env.rewarder.cum_reward)\n", + "print(\n", + " \"Number of unique imaged targets:\",\n", + " len(set(env.rewarder.data.imaged)),\n", + ")\n", + "print(\n", + " \"Number of duplicated targets:\",\n", + " env.rewarder.data.duplicates,\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Check [Training with RLlib PPO](../examples/rllib_training.ipynb) for an example on how to train the agent in this environment." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.12" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/pyproject.toml b/pyproject.toml index 0a196a88..98ad47f2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ readme = "README.md" requires-python = ">=3.10.0" license = { text = "MIT" } dependencies = [ - "bsk>=2.8.9", + "bsk>=2.12.0", "Deprecated", "gymnasium", "numpy",