feat(motor,flight): tank geometry + fluid density inputs, KML export

.gitignore

@@ -160,4 +160,10 @@ cython_debug/
 #.idea/
 
 # VSCode config
-.vscode/
+.vscode/
+
+# context specific ignores
+.context
+
+# lint
+.pylint.d/

src/controllers/flight.py

@@ -145,6 +145,28 @@ async def get_rocketpy_flight_rpy(
  flight_service = FlightService.from_flight_model(flight.flight)
  return flight_service.get_flight_rpy()
 
+ @controller_exception_handler
+ async def get_flight_kml(
+ self,
+ flight_id: str,
+ ) -> bytes:
+ """
+ Get the flight trajectory as a KML file.
+
+ Args:
+ flight_id: str
+
+ Returns:
+ bytes (KML XML)
+
+ Raises:
+ HTTP 404 Not Found: If the flight is not found
+ in the database.
+ """
+ flight = await self.get_flight_by_id(flight_id)
+ flight_service = FlightService.from_flight_model(flight.flight)
+ return flight_service.get_flight_kml()
+
  @controller_exception_handler
  async def get_flight_simulation(
  self,

src/models/motor.py

@@ -20,7 +20,11 @@ class MotorModel(ApiBaseModel):
 
  # Required parameters
  thrust_source: List[List[float]]
- burn_time: float
+ # burn_time is optional for Liquid/Hybrid/Solid motors — rocketpy
+ # auto-detects the burn window from the thrust_source array span.
+ # GenericMotor still requires it; the motor service re-raises an
+ # explicit error when the GENERIC path receives None.
+ burn_time: Optional[float] = None
  nozzle_radius: float
  dry_mass: float
  dry_inertia: Tuple[float, float, float] = (0, 0, 0)

src/models/sub/tanks.py

@@ -1,6 +1,7 @@
 from enum import Enum
-from typing import Optional, Tuple, List
-from pydantic import BaseModel
+from typing import Annotated, List, Literal, Optional, Tuple, Union
+
+from pydantic import BaseModel, Field, model_validator
 
 
 class TankKinds(str, Enum):
@@ -10,14 +11,64 @@ class TankKinds(str, Enum):
  ULLAGE: str = "ULLAGE"
 
 
+# Scalar: constant density (kg/m^3).
+# [(temp_K, density_kg_m3)]: temperature-dependent (LOX / N2O).
+# No pressure dependence.
+DensityInput = Union[float, List[Tuple[float, float]]]
+
+
 class TankFluids(BaseModel):
  name: str
- density: float
+ density: DensityInput
+
+
+class CustomTankGeometry(BaseModel):
+ geometry_kind: Literal["custom"] = "custom"
+ geometry: List[Tuple[Tuple[float, float], float]]
+
+
+class CylindricalTankGeometry(BaseModel):
+ geometry_kind: Literal["cylindrical"] = "cylindrical"
+ radius: float
+ height: float
+ spherical_caps: bool = False
+
+
+class SphericalTankGeometry(BaseModel):
+ geometry_kind: Literal["spherical"] = "spherical"
+ radius: float
+
+
+TankGeometryInput = Annotated[
+ Union[
+ CustomTankGeometry,
+ CylindricalTankGeometry,
+ SphericalTankGeometry,
+ ],
+ Field(discriminator="geometry_kind"),
+]
+
+
+# Map tank_kind → tuple of MotorTank field names that rocketpy's
+# corresponding Tank subclass requires.
+_REQUIRED_FIELDS_BY_TANK_KIND = {
+ TankKinds.MASS_FLOW: (
+ "initial_liquid_mass",
+ "initial_gas_mass",
+ "liquid_mass_flow_rate_in",
+ "liquid_mass_flow_rate_out",
+ "gas_mass_flow_rate_in",
+ "gas_mass_flow_rate_out",
+ ),
+ TankKinds.LEVEL: ("liquid_height",),
+ TankKinds.ULLAGE: ("ullage",),
+ TankKinds.MASS: ("liquid_mass", "gas_mass"),
+}
 
 
 class MotorTank(BaseModel):
  # Required parameters
- geometry: List[Tuple[Tuple[float, float], float]]
+ geometry: TankGeometryInput
  gas: TankFluids
  liquid: TankFluids
  flux_time: Tuple[float, float]
@@ -48,3 +99,19 @@ class MotorTank(BaseModel):
 
  # Computed parameters
  tank_kind: TankKinds = TankKinds.MASS_FLOW
+
+ @model_validator(mode='after')
+ def validate_tank_kind_fields(self):
+ # reject incoherent payloads at the API boundary
+ # instead of letting rocketpy crash during Tank construction.
+ missing = [
+ field
+ for field in _REQUIRED_FIELDS_BY_TANK_KIND[self.tank_kind]
+ if getattr(self, field) is None
+ ]
+ if missing:
+ raise ValueError(
+ f"tank_kind={self.tank_kind.value} requires: "
+ f"{', '.join(missing)}"
+ )
+ return self

src/routes/flight.py

@@ -301,6 +301,42 @@ async def update_flight_rocket(
  )
 
 
+@router.get(
+ "/{flight_id}/kml",
+ responses={
+ 200: {
+ "description": "KML trajectory file download",
+ "content": {"application/vnd.google-earth.kml+xml": {}},
+ }
+ },
+ status_code=200,
+ response_class=Response,
+)
+async def get_flight_kml(
+ flight_id: str,
+ controller: FlightControllerDep,
+):
+ """
+ Export a flight trajectory as a KML file for Google Earth.
+
+ ## Args
+ ``` flight_id: str ```
+ """
+ with tracer.start_as_current_span("get_flight_kml"):
+ kml = await controller.get_flight_kml(flight_id)
+ headers = {
+ "Content-Disposition": (
+ f'attachment; filename="flight_{flight_id}.kml"'
+ ),
+ }
+ return Response(
+ content=kml,
+ headers=headers,
+ media_type="application/vnd.google-earth.kml+xml",
+ status_code=200,
+ )
+
+
 @router.get("/{flight_id}/simulate")
 async def get_flight_simulation(
  flight_id: str,

src/routes/motor.py

@@ -37,6 +37,31 @@ async def create_motor(
 
  ## Args
  ``` models.Motor JSON ```
+
+ For liquid/hybrid motors the `tanks` field supports three geometry
+ kinds via the `geometry_kind` discriminator and a scalar-or-sampled
+ fluid `density`:
+
+ ```
+ {
+ "motor_kind": "LIQUID",
+ ...
+ "tanks": [{
+ "geometry": {
+ "geometry_kind": "cylindrical", // or "spherical", "custom"
+ "radius": 0.1, "height": 0.5
+ },
+ "liquid": {
+ "name": "LOX",
+ "density": [[90.0, 1141.0], [120.0, 1091.0]] // or scalar
+ },
+ "gas": {"name": "N2", "density": 1.2},
+ "tank_kind": "LEVEL",
+ "liquid_height": 0.25,
+ ...
+ }]
+ }
+ ```
  """
  with tracer.start_as_current_span("create_motor"):
  return await controller.post_motor(motor)

src/services/flight.py

@@ -1,9 +1,12 @@
 import json
+import os
+import tempfile
 from typing import Self, Tuple
 
 import numpy as np
 
 from rocketpy.simulation.flight import Flight as RocketPyFlight
+from rocketpy.simulation.flight_data_exporter import FlightDataExporter
 from rocketpy._encoders import RocketPyEncoder, RocketPyDecoder
 from rocketpy.mathutils.function import Function
 from rocketpy.motors.solid_motor import SolidMotor
@@ -224,6 +227,22 @@ def _to_float(value) -> float:
  case _:
  return float(value)
 
+ @staticmethod
+ def _extract_fluid_density(fluid):
+ """Project a rocketpy Fluid's density back onto the API schema.
+
+ The API accepts either a scalar or a list of (T_K, density)
+ samples. Rocketpy may store density as either a raw scalar or a
+ ``Function`` wrapping a 2D ``(T, P) -> density`` callable. A
+ full sample round-trip is not supported in this iteration;
+ Function-valued densities are collapsed to a scalar evaluated
+ at rocketpy's default reference (273.15 K, 101325 Pa).
+ """
+ density = fluid.density
+ if isinstance(density, Function):
+ return float(density(273.15, 101325))
+ return density
+
  @staticmethod
  def _extract_tanks(motor) -> list[MotorTank]:
  tanks: list[MotorTank] = []
@@ -240,20 +259,29 @@ def _extract_tanks(motor) -> list[MotorTank]:
  case _:
  tank_kind = TankKinds.MASS_FLOW
 
- geometry = [
+ # Geometry round-trip is lossy: even if the client originally
+ # sent a cylindrical/spherical geometry, we discretise it back
+ # to the generic piecewise form on read. Every rocketpy tank
+ # geometry exposes its internal piecewise dict via
+ # `tank.geometry.geometry`, so this path covers all three
+ # geometry subclasses uniformly.
+ geometry_segments = [
  (bounds, float(func(0)))
  for bounds, func in tank.geometry.geometry.items()
  ]
 
  data: dict = {
- "geometry": geometry,
+ "geometry": {
+ "geometry_kind": "custom",
+ "geometry": geometry_segments,
+ },
  "gas": TankFluids(
  name=tank.gas.name,
- density=tank.gas.density,
+ density=FlightService._extract_fluid_density(tank.gas),
  ),
  "liquid": TankFluids(
  name=tank.liquid.name,
- density=tank.liquid.density,
+ density=FlightService._extract_fluid_density(tank.liquid),
  ),
  "flux_time": tank.flux_time,
  "position": position,
@@ -474,6 +502,25 @@ def get_flight_simulation(self) -> FlightSimulation:
  flight_simulation = FlightSimulation(**encoded_attributes)
  return flight_simulation
 
+ def get_flight_kml(self) -> bytes:
+ """
+ Get the flight trajectory as a KML file for Google Earth.
+
+ Returns:
+ bytes (UTF-8 encoded KML)
+ """
+ with tempfile.NamedTemporaryFile(
+ suffix=".kml", delete=False
+ ) as tmp:
+ tmp_path = tmp.name
+ try:
+ FlightDataExporter(self.flight).export_kml(file_name=tmp_path)
+ with open(tmp_path, "rb") as fh:
+ return fh.read()
+ finally:
+ if os.path.exists(tmp_path):
+ os.unlink(tmp_path)
+
  def get_flight_rpy(self) -> bytes:
  """
  Get the portable JSON ``.rpy`` representation of the flight.