target.py

import numpy as np
import Modules.module_protocol as mp
from enum import Enum
from typing import Tuple, Callable
from utils import generate_rastered_circle
from errors import PositionNotOnCircleError
from random import randrange


class DirectionType(Enum):
    """
    Enumeration representing different movement directions.

    Attributes:
        NONE (int): No movement.
        UP (int): Upward movement.
        DOWN (int): Downward movement.
        LEFT (int): Leftward movement.
        RIGHT (int): Rightward movement.
        DIAGONAL_LEFT_DOWN (int): Diagonal movement to the left and down.
        DIAGONAL_LEFT_UP (int): Diagonal movement to the left and up.
        DIAGONAL_RIGHT_DOWN (int): Diagonal movement to the right and down.
        DIAGONAL_RIGHT_UP (int): Diagonal movement to the right and up.
        APPEAR (int): Represents appearing or no change in position.
    """

    NONE = 0
    UP = 1
    DOWN = 2
    LEFT = 3
    RIGHT = 4
    DIAGONAL_LEFT_DOWN = 5
    DIAGONAL_LEFT_UP = 6
    DIAGONAL_RIGHT_DOWN = 7
    DIAGONAL_RIGHT_UP = 8
    APPEAR = 9
    CIRCULAR = 10


class Direction:
    """
    Class representing a specific movement direction.

    Attributes:
        direction_type (DirectionType): The type of movement direction.

    Methods:
        reverse(): Reverse the current direction.
        movement_vector() -> np.ndarray: Get the movement vector associated with the direction.
    """

    def __init__(self, direction_type: DirectionType):
        """
        Initialize a Direction object with a given direction type.

        Parameters:
            direction_type (DirectionType): The initial direction type.
        """
        self.direction_type = direction_type

    def reverse(self):
        """
        Reverse the current direction.
        """
        match self.direction_type:
            case DirectionType.UP:
                self.direction_type = DirectionType.DOWN
            case DirectionType.DOWN:
                self.direction_type = DirectionType.UP
            case DirectionType.LEFT:
                self.direction_type = DirectionType.RIGHT
            case DirectionType.RIGHT:
                self.direction_type = DirectionType.LEFT
            case DirectionType.DIAGONAL_LEFT_DOWN:
                self.direction_type = DirectionType.DIAGONAL_RIGHT_UP
            case DirectionType.DIAGONAL_LEFT_UP:
                self.direction_type = DirectionType.DIAGONAL_RIGHT_DOWN
            case DirectionType.DIAGONAL_RIGHT_DOWN:
                self.direction_type = DirectionType.DIAGONAL_LEFT_UP
            case DirectionType.DIAGONAL_RIGHT_UP:
                self.direction_type = DirectionType.DIAGONAL_LEFT_DOWN
            case _:
                self.direction_type = DirectionType.NONE

    def movement_vector(self):
        """
        Get the movement vector associated with the direction.

        Returns:
            np.ndarray: A NumPy array representing the movement vector.
                        For example, np.array([1, 0]) for rightward movement.
        """
        match self.direction_type:
            case DirectionType.UP:
                return np.array([0, -1])
            case DirectionType.DOWN:
                return np.array([0, 1])
            case DirectionType.RIGHT:
                return np.array([1, 0])
            case DirectionType.LEFT:
                return np.array([-1, 0])
            case DirectionType.DIAGONAL_LEFT_UP:
                return np.array([-1, -1])
            case DirectionType.DIAGONAL_LEFT_DOWN:
                return np.array([-1, 1])
            case DirectionType.DIAGONAL_RIGHT_UP:
                return np.array([1, -1])
            case DirectionType.DIAGONAL_RIGHT_DOWN:
                return np.array([1, 1])
            case DirectionType.NONE | DirectionType.APPEAR:
                return np.array([0, 0])
            case _:
                return np.array([0, 0])

    def __eq__(self, other):
        if isinstance(self, other.__class__):
            return self.direction_type.name == other.direction_type.name
        return False


class TargetConfiguration(mp.ModuleProtocol):
    """
    Class representing the configuration of a target.

    Attributes:
        constant_start (bool): Flag indicating whether the target has a constant starting position.
        random_start (bool): Flag indicating whether the target has a random starting position.
        random_range (list[Tuple[int, int]]): The range of x and y position the agent can spawn. Is only used when random_start == True.
        leaving (bool): Flag indicating whether the target is in a leaving state.
        random_moving (bool): Flag indicating whether the target is moving randomly.
        not_moving (bool): Flag indicating whether the target is not moving.
        non_target (bool): Flag indicating whether the target is a non-target.
        position (tuple): Initial position of the target as a tuple (x, y).
        velocity (float): Velocity of the target.
        movement_mode (DirectionType): Mode of movement for the target.
        bandwidth: (float | None): The targets bandwidth in a marker environment

    Methods:
        __eq__(self, other): Check if two TargetConfiguration objects are equal.
        is_valid(self) -> bool: Check if the configuration is valid.
        export_to_yaml(self) -> dict: Export the configuration to a YAML-compatible dictionary.

    """

    def __init__(
        self,
        constant_start: bool = False,
        random_start=False,
        leaving: bool = False,
        random_moving: bool = False,
        random_range: list[tuple[int, int]] | None = None,
        not_moving: bool = False,
        non_target: bool = False,
        position=(2, 3),
        velocity: float = 0.5,
        movement_mode: DirectionType = DirectionType.UP,
        catastrophic: bool = False,
        bandwidth: float | None = None,
        center: np.ndarray | None = None,
        radius: int | None = None,
    ):
        """
        Initialize a TargetConfiguration object.

        Parameters:
            constant_start (bool): Flag indicating whether the target has a constant starting position.
            random_start (bool): Flag indicating whether the target has a random starting position.
            random_range (list[Tuple[int, int]]): The range of x and y position the agent can spawn. Is only used when random_start == True.
            leaving (bool): Flag indicating whether the target is in a leaving state.
            random_moving (bool): Flag indicating whether the target is moving randomly.
            not_moving (bool): Flag indicating whether the target is not moving.
            non_target (bool): Flag indicating whether the target is a non-target.
            position (tuple): Initial position of the target in the grid as a tuple (x, y).
            velocity (float): Velocity of the target.
            movement_mode (DirectionType): Mode of movement for the target.
            catastrophic (bool): Flag indicating if this target is a catastrophic one, i.e. missing it during an observation may result in a big penalty
            bandwidth: (float | None): The targets bandwidth in a marker environment
            center (np.ndarray): the center of a circle for a circular moving target
            radius (int): the radius of a circle for a circular moving target
        """

        self._constant_start = constant_start
        self._random_start = random_start
        self._random_range = random_range
        self._leaving = leaving
        self._random_moving = random_moving
        self._not_moving = not_moving
        self._non_target = non_target
        self._position = np.asarray(position)
        self._velocity = velocity
        self._movement_mode = movement_mode
        self._catastrophic = catastrophic
        self._bandwidth = bandwidth
        self._center = center
        self._radius = radius

    @property
    def constant_start(self):
        """bool: Flag indicating whether the target has a constant starting position."""
        return self._constant_start

    @property
    def random_start(self):
        """bool: Flag indicating whether the target has a random starting position."""
        return self._random_start

    @property
    def random_range(self):
        """(list[Tuple[int, int]]): The random range for x and y coordinates where the agent can spawn"""
        return self._random_range

    @property
    def leaving(self):
        """bool: Flag indicating whether the target is in a leaving state."""
        return self._leaving

    @property
    def random_moving(self):
        """bool: Flag indicating whether the target is moving randomly."""
        return self._random_moving

    @property
    def not_moving(self):
        """bool: Flag indicating whether the target is not moving."""
        return self._not_moving

    @property
    def non_target(self):
        """bool: Flag indicating whether the target is a non-target."""
        return self._non_target

    @property
    def position(self):
        """tuple: Initial position of the target as a tuple (x, y).
        This overwritten by random_start and random_range
        and won't be used when both are set."""
        return self._position

    @property
    def velocity(self):
        """float: Velocity of the target."""
        return self._velocity

    @property
    def movement_mode(self):
        """DirectionType: Mode of movement for the target."""
        return self._movement_mode

    @property
    def catastrophic(self) -> bool:
        """bool: Flag indicating whether the target is a catastrophic target or not"""
        return self._catastrophic

    @property
    def bandwidth(self) -> float | None:
        """(float | None): The target's bandwidth in a marker environment"""
        return self._bandwidth

    @property
    def center(self) -> np.ndarray | None:
        """(np.ndarray | None): The radius of the circle of circular_moving_target"""
        return self._center

    @property
    def radius(self) -> int | None:
        """(int | None): The centre of the circle of circular_moving_target"""
        return self._radius

    def __eq__(self, other):
        """
        Check if two TargetConfiguration objects are equal.

        Parameters:
            other (TargetConfiguration): Another TargetConfiguration object to compare with.

        Returns:
            bool: True if the objects are equal, False otherwise.
        """
        if isinstance(other, self.__class__):
            return (
                (
                    self.constant_start,
                    self.random_start,
                    self.leaving,
                    self.random_moving,
                    self.not_moving,
                    self.non_target,
                    self.velocity,
                    self._movement_mode,
                    self._random_range,
                    self._catastrophic,
                    self._bandwidth,
                    self.radius,
                )
                == (
                    other.constant_start,
                    other.random_start,
                    other.leaving,
                    other.random_moving,
                    other.not_moving,
                    other.non_target,
                    other.velocity,
                    other.movement_mode,
                    other.random_range,
                    other.catastrophic,
                    other.bandwidth,
                    other.radius,
                )
                and np.array_equal(self.position, other.position)
                and np.array_equal(self._center, other.center)
            )
        return NotImplemented

    def is_valid(self) -> bool:
        """
        Check if the configuration is valid.

        Returns:
            bool: True if the configuration is valid, False otherwise.
        """
        if self.constant_start and self.random_start:
            return False

        if self.not_moving and self.random_moving:
            return False

        return True

    def export_to_yaml(self) -> dict:
        """
        Export the configuration to a YAML-compatible dictionary.

        Returns:
            dict: A dictionary representing the configuration.
        """
        data = super().export_to_yaml()
        data["position"] = f"({self.position[0]},{self.position[1]})"
        data["movement_mode"] = self._movement_mode.name
        if self.center is not None:
            data["center"] = f"({self.center[0]},{self.center[1]})"

        if self._random_range is None:
            del data["random_range"]
        else:
            x_low, x_high = self._random_range[0]
            y_low, y_high = self._random_range[1]
            data["random_range"] = {
                "x_lower": x_low,
                "x_upper": x_high,
                "y_lower": y_low,
                "y_upper": y_high,
            }
        # do not export data where value is None
        for k in [k for k, v in data.items() if v is None]:
            del data[k]

        return data


class Target:
    """
    Class representing a target.

    Attributes:
        color (tuple): RGB color tuple representing the target's color.
        reward (int): Reward associated with the target.
        configuration (TargetConfiguration): Configuration object defining the target's behavior.
        position (np.ndarray): Current position of the target as a NumPy array.
        velocity (float): Velocity of the target.
        movement (Direction): Object representing the target's movement direction.
        _org_position (np.ndarray): Original position of the target.
        steps_per_timestep (int): Number of steps the target moves in a single timestep.
        steps_until_next_steps (int): Steps remaining until the target moves again.

    Methods:
        step(self): Move the target according to its configuration for one timestep.
        reverse_direction(self): Reverse the direction of the target's movement.
        update_position(self, new_position: np.ndarray): Update the target's position.
        is_hit(self, pos_to_compare: np.ndarray) -> bool: Check if a given position matches the target's position.

        @staticmethod
        dummy_target() -> Target: Create a dummy target with default properties.

    """

    def __init__(
        self,
        color,
        reward: int,
        position: np.ndarray,
        velocity: float,
        movement: Direction,
        configuration: TargetConfiguration,
        random_start: bool = False,
        random_start_range: list[tuple[int, int]] | None = None,
        random_moving: bool = False,
        catastrophic = False,
        bandwidth: float | None = None,
    ):
        """
        Initialize a Target object.

        Parameters:
            color (tuple): RGB color tuple representing the target's color.
            reward (int): Reward associated with the target.
            position (np.ndarray): Initial position of the target as a NumPy array.
            velocity (float): Velocity of the target.
            movement (Direction): Object representing the target's movement direction.
            configuration (TargetConfiguration): Configuration object defining the target's behavior.
            random_start (bool): Flag indicating if the target spawns randomly.
            random_start_range (list(Tuple[int, int])): The range of x and y coordinates where the agent can spawn.
            random_moving: Flag indicating if the target moves randomly (forward / backward) along the specified movement type
            catastrophic (bool): Flag indicating if this target is a catastrophic one, i.e. missing it during an observation may result in a big penalty
            bandwidth: (float | None): The targets bandwidth in a marker environment
        """

        self.color = color
        self.reward = reward
        self.configuration = configuration
        self.position = position
        self.velocity = velocity
        self.movement = movement
        self.random_start = random_start
        self.catastrophic = catastrophic
        self.bandwidth = bandwidth
        if bandwidth is not None:
            self._sine_functions = self._create_stacked_sine_function_with_bandwidth(
                bandwidth=bandwidth
            )
            self._previous_sine_value = 0
        self._random_start_range = random_start_range
        self.random_moving = random_moving
        if random_start and random_start_range is not None:
            self.set_random_position()
        self._org_position = position

        if velocity <= 0:
            self.steps_per_timestep = 0
            self.steps_until_next_steps = 0
            return

        velocity = 1 if np.isinf(velocity) else velocity

        self.steps_per_timestep = int(np.round(velocity)) if velocity >= 1 else 1
        self.steps_until_next_steps = 0 if velocity >= 1 else np.round(1 / velocity) - 1

    def step(self, elapsed_time: float | None = None) -> None:
        """
        Move the target according to its configuration for one timestep.
        :param elapsed_time (optional): The time in seconds elapsed in the environment, this is used to calculate if the target changes its direction
        """

        if self.steps_until_next_steps == 0:
            for step in range(self.steps_per_timestep):
                if self.movement.direction_type == DirectionType.APPEAR:
                    if self.random_start:
                        if self.is_visible:
                            self.position = np.array([-1, -1])
                        else:
                            self.set_random_position()
                    else:
                        self.position = (
                            np.array([-1, -1])
                            if np.array_equal(self.position, self._org_position)
                            else self._org_position
                        )
                else:
                    # choose direction (forward / backward) uniformly at random when random_moving is enabled
                    # reverse direction according to stacked sine function in mode with bandwidth
                    if self.random_moving or (self.bandwidth and elapsed_time):
                        if self.bandwidth and elapsed_time:
                            current_sine_value = self._sine_functions(elapsed_time)
                            if self._previous_sine_value * current_sine_value < 0:
                                self.movement.reverse()
                            self._previous_sine_value = current_sine_value
                        else:
                            if np.random.rand() > 0.5:
                                self.movement.reverse()
                    self.position += self.movement.movement_vector()
                self.steps_until_next_steps = (
                    0 if self.velocity >= 1 else np.round(1 / self.velocity) - 1
                )
                # Slower targets in Appear mode should not be visible only half the time as they are visible
                if (
                    self.movement.direction_type == DirectionType.APPEAR
                    and np.array_equal(self.position, [-1, -1])
                    and self.velocity <= 0.1
                ):
                    self.steps_until_next_steps = np.round(
                        self.steps_until_next_steps / 2
                    )
        else:
            self.steps_until_next_steps -= 1

    def reverse_direction(self):
        """
        Reverse the direction of the target's movement.
        """
        self.movement.reverse()

    def update_position(self, new_position: np.array):
        """
        Update the target's position.

        Parameters:
            new_position (np.ndarray): New position of the target as a NumPy array.
        """
        self.position = new_position

    def set_random_position(self):
        """
        Sets the target at a random position as specified with random_start_range
        """
        if (
            not self.random_start
            or self._random_start_range is None
            or len(self._random_start_range) != 2
        ):
            return
        self.position = np.array(
            [np.random.randint(low, high) for low, high in self._random_start_range]
        )

    def is_hit(self, pos_to_compare: np.array) -> bool:
        """
        Check if a given position matches the target's position.

        Parameters:
            pos_to_compare (np.ndarray): Position to compare with the target's position.

        Returns:
            bool: True if the positions match, False otherwise.
        """
        return np.array_equal(pos_to_compare, self.position)

    def reset(self):

        self._sine_functions = self._create_stacked_sine_function_with_bandwidth(
            bandwidth=self.bandwidth
        )

    def _create_stacked_sine_function_with_bandwidth(
        self, bandwidth: float
    ) -> Callable:
        """
        Creates a function that computes the sum of a random number of sine waves with constant bandwidth but random phase shift between 0 and 2pi.
        The result is normalized to +180° and -180° degree, independent of the number of randomly chosen sine waves
        :param bandwidth: the bandwidth of the sine functions
        :return: f(x) = sum^(rand[21,42])(sin(bandwidth * x + phase_shift)) adjusted to degrees given the number of sine waves
        """
        stacked = []
        num_of_sine_waves = 42  # np.random.randint(21, 43)
        frequency_0 = 0.001
        frequency_diff = (bandwidth - frequency_0) / (
            num_of_sine_waves - 1
        )  # because the first with 0.0001 is given
        for i in range(num_of_sine_waves):
            phase_shift = np.random.uniform(0.0, 2 * np.pi)
            frequency_i = 2 * np.pi * (frequency_0 + i * frequency_diff)
            stacked.append(
                lambda x, freq=frequency_i, p_shift=phase_shift: 360
                * np.sin(freq * x + p_shift)
            )  # 360 * np.sin(freq * x + p_shift)
        scaling_factor = 100 / (360 * np.sqrt(num_of_sine_waves) * np.sqrt(2) / 2)
        return lambda x: scaling_factor * np.sum([f(x) for f in stacked])

    @property
    def is_visible(self) -> bool:
        return bool((self.position >= 0).all())

    @staticmethod
    def dummy_target() -> "Target":
        """
        Create a dummy target with default properties.

        Returns:
            Target: A dummy Target object.
        """
        return Target(
            color=(255, 0, 0),
            reward=3,
            velocity=1.0,
            position=np.array([2, 3]),
            movement=Direction(DirectionType.UP),
            configuration=TargetConfiguration(not_moving=True),
        )


class CircularMovingTarget(Target):
    """
    A class for circular moving targets around a center with a given radius.
    As default the specified position is set as marker_position, it has to be on the circle
    """

    def __init__(
        self,
        color,
        reward: int,
        position: np.ndarray,
        velocity: float,
        configuration: TargetConfiguration,
        bandwidth: float,
        center: np.ndarray,
        radius: int | float,
        catastrophic=False,
    ):
        super().__init__(
            color,
            reward,
            position,
            velocity,
            Direction(DirectionType.CIRCULAR),
            configuration,
            False,
            None,
            False,
            catastrophic,
            bandwidth,
        )
        self._radius = radius
        self._center = center
        self._circle_coordinates = generate_rastered_circle(
            radius=radius, center=center
        )
        self._previous_sine = self._sine_functions(0)
        self._crossed_marker = False
        # self.set_marker_position(marker_position=position)

    @property
    def radius(self):
        return self._radius

    @property
    def center(self):
        return self._center

    @property
    def circle_coordinates(self):
        return self._circle_coordinates

    @property
    def marker_position(self):
        return self._circle_coordinates[0]

    def next_position(self, clockwise: bool = True):
        idx = self._circle_coordinates.index(tuple(self.position))
        increment = 1 if clockwise else -1
        next_idx = idx + increment
        if next_idx >= len(self._circle_coordinates):
            next_idx = 0  # continue at zero
        self.position = self._circle_coordinates[next_idx]

    def get_next_cell_index_by_angle(self, angle: float):
        num_points = len(self._circle_coordinates)
        angle_increment = 360 / num_points
        index = int(np.rint(angle / angle_increment)) % num_points
        return index

    def get_next_position_by_angle(self, angle: float) -> tuple:
        index = self.get_next_cell_index_by_angle(angle=angle)
        return self._circle_coordinates[index]

    def set_marker_position(self, marker_position: np.ndarray):
        """
        Rotates the circle coordinates in a way that the marker position is the first element.
        Raises an exception if the given marker_position is not part of the circle
        :param marker_position: The coordinates that should be the first in the circle_coordinates_list.
        :return: void, PositionNotOnCircleError
        """
        if tuple(marker_position) not in self._circle_coordinates:
            raise PositionNotOnCircleError(
                marker_position=marker_position,
                circle_coordinates=self._circle_coordinates,
            )
        idx = self._circle_coordinates.index(tuple(marker_position))
        self._circle_coordinates = (
            self._circle_coordinates[idx:] + self._circle_coordinates[:idx]
        )

    def set_random_position_at_circle(self):
        """
        Sets the target's current position to a random circle coordinate
        :return:
        """
        self.position = np.array(
            self._circle_coordinates[randrange(len(self._circle_coordinates))]
        )

    @property
    def current_sine_value(self):
        return (
            self._previous_sine
        )  # this is the current sine value after step, it is previous from the view_point of the next step

    @property
    def crossed_marker(self):
        return self._crossed_marker

    def step(self, elapsed_time: float):
        current_sine = self._sine_functions(elapsed_time)
        self._crossed_marker = (
            self._previous_sine * current_sine <= 0
        )  # if the function has a sign change from the previous to the current step, then the marker position was crossed
        self.position = np.array(self.get_next_position_by_angle(current_sine))
        self._previous_sine = current_sine

    @property
    def one_hot_encoding(self):
        try:
            index = self._circle_coordinates.index((self.position[0], self.position[1]))
        except ValueError:
            index = -1
        one_hot_encoding = np.zeros(len(self._circle_coordinates), dtype=np.int8)
        if index >= 0:
            one_hot_encoding[index] = 1
        return one_hot_encoding