mirror of
https://git.intern.spaceteamaachen.de/ALPAKA/SPATZ.git
synced 2025-06-10 01:55:59 +00:00
added time specification for dataset and sensor classes
This commit is contained in:
dario
parent
6555009a94
commit
825884c279
332
spatz/dataset.py
332
spatz/dataset.py
@ -19,6 +19,33 @@ class Phase(Enum):
|
||||
ADI = 5
|
||||
|
||||
|
||||
def T1(angle):
|
||||
# return Rotation.from_euler('X', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[1, 0, 0],
|
||||
[0, math.cos(angle), math.sin(angle)],
|
||||
[0, -math.sin(angle), math.cos(angle)],
|
||||
])
|
||||
|
||||
|
||||
def T2(angle):
|
||||
# return Rotation.from_euler('Y', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[math.cos(angle), 0, -math.sin(angle)],
|
||||
[0, 1, 0],
|
||||
[math.sin(angle), 0, math.cos(angle)]
|
||||
])
|
||||
|
||||
|
||||
def T3(angle):
|
||||
# return Rotation.from_euler('Z', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[math.cos(angle), math.sin(angle), 0],
|
||||
[-math.sin(angle), math.cos(angle), 0],
|
||||
[0, 0, 1]
|
||||
])
|
||||
|
||||
|
||||
class Dataset(Advanceable):
|
||||
def __init__(self, path: str, interpolation: str = 'linear'):
|
||||
"""A wrapper class for a Pandas dataframe containing simulation data.
|
||||
@ -42,9 +69,23 @@ class Dataset(Advanceable):
|
||||
def _on_reset(self):
|
||||
pass
|
||||
|
||||
def _get_closest_idx(self, t: float) -> int:
|
||||
"""Gets an index _idx_ for the dataframe _df_ such that the values at the given time _t_ are somewhere between
|
||||
_idx_ and _idx+1_.
|
||||
|
||||
Args:
|
||||
t (float): The requested time.
|
||||
|
||||
Returns:
|
||||
int: The computed index.
|
||||
"""
|
||||
idx = (self.__df['time'] - t).abs().idxmin()
|
||||
idx = idx if self.__df['time'].loc[idx] < t else idx - 1
|
||||
|
||||
return idx
|
||||
|
||||
def _on_step(self, _: float):
|
||||
idx = (self.__df['time'] - self.get_time()).abs().idxmin()
|
||||
self.__idx = idx if self.__df['time'].loc[idx] < self.get_time() else idx - 1
|
||||
self.__idx = self._get_closest_idx(self.get_time())
|
||||
|
||||
def get_phase(self) -> Phase:
|
||||
"""
|
||||
@ -98,59 +139,102 @@ class Dataset(Advanceable):
|
||||
"""
|
||||
return max(self.__df['time'])
|
||||
|
||||
@staticmethod
|
||||
def T1(angle):
|
||||
# return Rotation.from_euler('X', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[1, 0, 0],
|
||||
[0, math.cos(angle), math.sin(angle)],
|
||||
[0, -math.sin(angle), math.cos(angle)],
|
||||
])
|
||||
def fetch_init_value(self, name: str) -> float:
|
||||
"""Get the initial value for a given attribute from the dataframe.
|
||||
|
||||
@staticmethod
|
||||
def T2(angle):
|
||||
# return Rotation.from_euler('Y', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[math.cos(angle), 0, -math.sin(angle)],
|
||||
[0, 1, 0],
|
||||
[math.sin(angle), 0, math.cos(angle)]
|
||||
])
|
||||
Args:
|
||||
name (str): The name of the value to fetch.
|
||||
|
||||
@staticmethod
|
||||
def T3(angle):
|
||||
# return Rotation.from_euler('Z', angle, degrees=False).as_matrix()
|
||||
return np.array([
|
||||
[math.cos(angle), math.sin(angle), 0],
|
||||
[-math.sin(angle), math.cos(angle), 0],
|
||||
[0, 0, 1]
|
||||
])
|
||||
|
||||
def local_to_body(self) -> ArrayLike:
|
||||
Returns:
|
||||
float: Returns the requested value.
|
||||
"""
|
||||
return self.__df[name].iloc[0]
|
||||
|
||||
def fetch_init_values(self, names: List[str]) -> ArrayLike:
|
||||
"""Get the initial value for given attributes from the dataframe.
|
||||
|
||||
Args:
|
||||
names (List[str]): Names of the values to get.
|
||||
|
||||
Returns:
|
||||
np.array: Returns a numpy array containing the requested values in the same order as in the input list.
|
||||
"""
|
||||
return np.asarray([self.fetch_init_value(name) for name in names])
|
||||
|
||||
def fetch_value(self, name: str, t: float | None = None) -> float:
|
||||
"""Get a specific value from the dataframe.
|
||||
|
||||
Args:
|
||||
name (str): The name of the value to fetch.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
float: Returns the requested value.
|
||||
"""
|
||||
idx = idx if t is None else self._get_closest_idx(t)
|
||||
|
||||
if self.__interpolation == 'linear':
|
||||
t_min = self.__df.at[idx, 'time']
|
||||
t_max = self.__df.at[idx + 1, 'time']
|
||||
|
||||
# Sometimes no time passes in-between two samples.
|
||||
if t_max == t_min:
|
||||
return self.__df.at[name, idx]
|
||||
|
||||
# Compute the weight for interpolation.
|
||||
alpha = (self.get_time() - t_min) / (t_max - t_min)
|
||||
|
||||
# Interpolate linearly between the two data points.
|
||||
return (1 - alpha) * self.__df.at[idx, name] + alpha * self.__df.at[idx + 1, name]
|
||||
|
||||
def fetch_values(self, names: List[str], t: float | None = None) -> ArrayLike:
|
||||
"""Get specific values from the dataframe.
|
||||
|
||||
Args:
|
||||
names (List[str]): Names of the values to get.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
np.array: Returns a numpy array containing the requested values in the same order as in the input list.
|
||||
"""
|
||||
return np.asarray([self.fetch_value(name, t) for name in names])
|
||||
|
||||
def local_to_body(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): The time to get the transformation matrix for.
|
||||
|
||||
Returns:
|
||||
ArrayLike: The current transformation matrix from local to body-fixed coords.
|
||||
"""
|
||||
# Get the rotation in the local coordinate system.
|
||||
rots = self.fetch_values(['pitch_l', 'yaw_l', 'roll_l'])
|
||||
rots = self.fetch_values(['pitch_l', 'yaw_l', 'roll_l'], t)
|
||||
pitch_l, yaw_l, roll_l = rots[0], rots[1], rots[2]
|
||||
|
||||
return self.T1(roll_l) @ self.T2(pitch_l - math.pi/2) @ self.T1(-yaw_l)
|
||||
|
||||
def global_to_local(self) -> ArrayLike:
|
||||
def global_to_local(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): The time to get the transformation matrix for.
|
||||
|
||||
Returns:
|
||||
ArrayLike: The current transformation matrix from global to local coords.
|
||||
"""
|
||||
decl = self.fetch_value('declination')
|
||||
long = self.fetch_value('longitude')
|
||||
decl = self.fetch_value('declination', t)
|
||||
long = self.fetch_value('longitude', t)
|
||||
t0 = self.__df['time'].iloc[0]
|
||||
|
||||
omega_E = (2*math.pi) / (24*60*60)
|
||||
|
||||
return self.T2(-decl) @ self.T3(long + omega_E * t0)
|
||||
|
||||
def global_to_launch_rail(self) -> ArrayLike:
|
||||
def global_to_launch_rail(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): The time to get the transformation matrix for. Doesn't do anything here because
|
||||
the transformation remains the same across all time steps.
|
||||
|
||||
Returns:
|
||||
ArrayLike: The current transformation matrix from global to launch rail coords.
|
||||
"""
|
||||
@ -159,178 +243,180 @@ class Dataset(Advanceable):
|
||||
|
||||
return self.T2(-math.pi/2 - init_lat) @ self.T3(init_long)
|
||||
|
||||
def local_to_launch_rail(self) -> ArrayLike:
|
||||
def local_to_launch_rail(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): The time to get the transformation matrix for.
|
||||
|
||||
Returns:
|
||||
ArrayLike: The current transformation matrix from local to launch rail coords.
|
||||
"""
|
||||
return self.global_to_launch_rail() @ np.linalg.inv(self.global_to_local())
|
||||
return self.global_to_launch_rail(t) @ np.linalg.inv(self.global_to_local(t))
|
||||
|
||||
def launch_rail_to_body(self) -> ArrayLike:
|
||||
def launch_rail_to_body(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): The time to get the transformation matrix for.
|
||||
|
||||
Returns:
|
||||
ArrayLike: The current transformation matrix from launch rail to local coords.
|
||||
"""
|
||||
return self.local_to_body() @ np.linalg.inv(self.local_to_launch_rail())
|
||||
return self.local_to_body(t) @ np.linalg.inv(self.local_to_launch_rail(t))
|
||||
|
||||
def is_transsonic(self) -> bool:
|
||||
def get_mach_number(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
bool: Returns True if the rocket is flying with transsonic speed at the current time of the simulation.
|
||||
float: Returns the mach number at the specified time.
|
||||
"""
|
||||
mach = self.get_mach_number()
|
||||
return self.fetch_value('mach', t)
|
||||
|
||||
def is_transsonic(self, t: float | None = None) -> bool:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
bool: Returns True if the rocket is flying with transsonic speed at the specified time.
|
||||
"""
|
||||
mach = self.get_mach_number(t)
|
||||
|
||||
return mach > 0.8 and mach < 1.2
|
||||
|
||||
def is_supersonic(self) -> bool:
|
||||
def is_supersonic(self, t: float | None = None) -> bool:
|
||||
"""
|
||||
Returns:
|
||||
bool: True if the rocket is flying with supersonic speed at the current time of the simulation.
|
||||
"""
|
||||
return self.get_mach_number() > 1
|
||||
|
||||
def fetch_value(self, name: str) -> float:
|
||||
"""Get a specific value from the dataframe.
|
||||
|
||||
Args:
|
||||
name (str): The name of the value to fetch.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
float: Returns the requested value.
|
||||
bool: True if the rocket is flying with supersonic speed at the specified time.
|
||||
"""
|
||||
if self.__interpolation == 'linear':
|
||||
t_min = self.__df.at[self.__idx, 'time']
|
||||
t_max = self.__df.at[self.__idx + 1, 'time']
|
||||
|
||||
# Sometimes no time passes in-between two samples.
|
||||
if t_max == t_min:
|
||||
return self.__df.at[name, self.__idx]
|
||||
|
||||
# Compute the weight for interpolation.
|
||||
alpha = (self.get_time() - t_min) / (t_max - t_min)
|
||||
|
||||
# Interpolate linearly between the two data points.
|
||||
return (1 - alpha) * self.__df.at[self.__idx, name] + alpha * self.__df.at[self.__idx + 1, name]
|
||||
|
||||
def fetch_values(self, names: List[str]) -> np.array:
|
||||
"""Get specific values from the dataframe.
|
||||
return self.get_mach_number(t) > 1
|
||||
|
||||
def get_velocity(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
names (List[str]): Names of the values to get.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
np.array: Returns a numpy array containing the requested values in the same order as in the input list.
|
||||
np.array: Returns the velocity at the current time.
|
||||
"""
|
||||
return np.asarray([self.fetch_value(name) for name in names])
|
||||
|
||||
def get_velocity(self) -> float:
|
||||
"""
|
||||
Returns:
|
||||
np.array: Returns the velocity at the current time of the simulation.
|
||||
"""
|
||||
return self.fetch_value('velocity')
|
||||
return self.fetch_value('velocity', t)
|
||||
|
||||
def get_acceleration(self, frame='FL') -> ArrayLike:
|
||||
"""_summary_
|
||||
|
||||
def get_acceleration(self, frame='FL', t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
frame (str, optional): _description_. Defaults to 'FL'.
|
||||
frame (str, optional): The coordinate frame to compute the acceleration for. Defaults to 'FL'.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
ArrayLike: _description_
|
||||
ArrayLike: Returns the spacecraft's acceleration at the given time.
|
||||
"""
|
||||
acc = self.fetch_values(['ax', 'ay', 'az'])
|
||||
acc = self.fetch_values(['ax', 'ay', 'az'], t)
|
||||
|
||||
if frame == 'B':
|
||||
return self.launch_rail_to_body() @ acc
|
||||
return self.launch_rail_to_body(t) @ acc
|
||||
|
||||
return acc
|
||||
|
||||
def get_angular_velocities(self) -> ArrayLike:
|
||||
def get_angular_velocities(self, t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
ArrayLike: Gets the derivatives in angular velocity across all axes of the rocket.
|
||||
"""
|
||||
return self.fetch_values(['omega_X', 'omega_Y', 'omega_Z'])
|
||||
return self.fetch_values(['OMEGA_X', 'OMEGA_Y', 'OMEGA_Z'], t)
|
||||
|
||||
def get_velocity(self, frame='FL') -> ArrayLike:
|
||||
def get_velocity(self, frame='FL', t: float | None = None) -> ArrayLike:
|
||||
"""
|
||||
Args:
|
||||
frame (str, optional): _description_. Defaults to 'FL'.
|
||||
frame (str, optional): The coordinate frame to compute the velocity for. Defaults to 'FL'.
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
ArrayLike: _description_
|
||||
ArrayLike: Returns the spacecraft's velocity at a given time.
|
||||
"""
|
||||
|
||||
vel = self.fetch_values(['vx', 'vy', 'vz'])
|
||||
vel = self.fetch_values(['vx', 'vy', 'vz'], t)
|
||||
|
||||
if frame == 'B':
|
||||
return self.launch_rail_to_body() @ vel
|
||||
return self.launch_rail_to_body(t) @ vel
|
||||
|
||||
return vel
|
||||
|
||||
def get_mach_number(self) -> float:
|
||||
def get_speed_of_sound(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Returns:
|
||||
float: Returns the mach number at the current time of the simulation.
|
||||
"""
|
||||
return self.fetch_value('mach')
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
def get_speed_of_sound(self) -> float:
|
||||
"""
|
||||
Returns:
|
||||
float: Returns the speed of sound at the current time of the simulation.
|
||||
float: Returns the speed of sound at the specified time.
|
||||
"""
|
||||
return self.fetch_value('speedofsound')
|
||||
return self.fetch_value('speedofsound', t)
|
||||
|
||||
def get_rotation_rates(self) -> np.array:
|
||||
def get_rotation(self, t: float | None = None) -> np.array:
|
||||
"""
|
||||
Returns:
|
||||
np.array: Returns the rotation rates at the current time of the simulation.
|
||||
"""
|
||||
return self.fetch_values(['OMEGA_X', 'OMEGA_Y', 'OMEGA_Z'])
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
def get_rotation(self) -> np.array:
|
||||
"""
|
||||
Returns:
|
||||
np.array: _description_
|
||||
np.array: returns the rotation at the specified time.
|
||||
"""
|
||||
return self.fetch_values(['pitch_l', 'yaw_l', 'roll_l'])
|
||||
return self.fetch_values(['pitch_l', 'yaw_l', 'roll_l'], t)
|
||||
|
||||
def get_temperature(self) -> float:
|
||||
def get_temperature(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
np.array: Returns the temperature at the current time of the simulation.
|
||||
np.array: Returns the temperature at the spepcified time.
|
||||
"""
|
||||
return self.fetch_value('temperature')
|
||||
return self.fetch_value('temperature', t)
|
||||
|
||||
def get_pressure(self) -> float:
|
||||
def get_pressure(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
np.array: Returns the pressure at the current time of the simulation.
|
||||
"""
|
||||
return self.fetch_value('pressure')
|
||||
return self.fetch_value('pressure', t)
|
||||
|
||||
def get_thrust(self) -> float:
|
||||
def get_thrust(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
float: Returns the thrust value for the current time of the simulation.
|
||||
float: Returns the thrust value for the specified time.
|
||||
"""
|
||||
return self.fetch_value('thrust')
|
||||
return self.fetch_value('thrust', t)
|
||||
|
||||
def get_drag(self) -> float:
|
||||
def get_drag(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
float: Returns the drag value for the current time of the simulation.
|
||||
float: Returns the drag value for the specified time.
|
||||
"""
|
||||
return self.fetch_value('drag')
|
||||
return self.fetch_value('drag', t)
|
||||
|
||||
def get_mass(self) -> float:
|
||||
def get_mass(self, t: float | None = None) -> float:
|
||||
"""
|
||||
Args:
|
||||
t (float): Allows specification of a different time instead of the current time. None for current time.
|
||||
|
||||
Returns:
|
||||
float: Returns the mass value for the current time of the simulation.
|
||||
float: Returns the mass value for the specified time.
|
||||
"""
|
||||
return self.fetch_value('mass')
|
||||
return self.fetch_value('mass', t)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@ -37,11 +37,14 @@ class Sensor:
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def _get_data(self) -> ArrayLike | float:
|
||||
def _get_data(self, t: float = None) -> ArrayLike | float:
|
||||
raise NotImplementedError()
|
||||
|
||||
def get_init_data() -> ArrayLike:
|
||||
pass
|
||||
|
||||
def __call__(self) -> ArrayLike | float:
|
||||
out = self._get_data()
|
||||
def __call__(self, t: float = None) -> ArrayLike | float:
|
||||
out = self._get_data(t = t)
|
||||
out = self._sensor_specific_effects(out)
|
||||
|
||||
for transform in self._transforms:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user