blackify project

Author: GabrielBarberini

lib/api.py

@@ -1,6 +1,7 @@
 """
 This is the main API file for the RocketPy API.
 """
+
 import logging
 
 from fastapi import FastAPI, Request, status
@@ -79,7 +80,9 @@ async def __perform_healthcheck():
 
 # Errors
 @app.exception_handler(RequestValidationError)
-async def validation_exception_handler(request: Request, exc: RequestValidationError):
+async def validation_exception_handler(
+    request: Request, exc: RequestValidationError
+):
     exc_str = f"{exc}".replace("\n", " ").replace("   ", " ")
     logging.error(f"{request}: {exc_str}")
     content = {"status_code": 10422, "message": exc_str, "data": None}

lib/controllers/flight.py

@@ -51,7 +51,10 @@ class FlightController:
     """
 
     def __init__(
-        self, flight: Flight, rocket_option: RocketOptions, motor_kind: MotorKinds
+        self,
+        flight: Flight,
+        rocket_option: RocketOptions,
+        motor_kind: MotorKinds,
     ):
         rocketpy_rocket = RocketController(
             flight.rocket, rocket_option=rocket_option, motor_kind=motor_kind
@@ -110,7 +113,8 @@ async def get_flight(flight_id: int) -> "Union[Flight, HTTPException]":
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         return successfully_read_flight
@@ -136,13 +140,18 @@ async def get_rocketpy_flight(
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         successfully_read_rocketpy_flight = FlightController(
             flight=successfully_read_flight,
-            rocket_option=RocketOptions(successfully_read_flight.rocket._rocket_option),
-            motor_kind=MotorKinds(successfully_read_flight.rocket.motor._motor_kind),
+            rocket_option=RocketOptions(
+                successfully_read_flight.rocket._rocket_option
+            ),
+            motor_kind=MotorKinds(
+                successfully_read_flight.rocket.motor._motor_kind
+            ),
         ).rocketpy_flight
 
         return FlightPickle(
@@ -171,12 +180,15 @@ async def update_flight(
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         successfully_updated_flight = await FlightRepository(
             flight=self.flight, flight_id=flight_id
-        ).update_flight(rocket_option=self.rocket_option, motor_kind=self.motor_kind)
+        ).update_flight(
+            rocket_option=self.rocket_option, motor_kind=self.motor_kind
+        )
         if not successfully_updated_flight:
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
@@ -207,7 +219,8 @@ async def update_env(
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         flight = successfully_read_flight.dict()
@@ -246,7 +259,8 @@ async def update_rocket(
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         flight = successfully_read_flight.dict()
@@ -267,7 +281,9 @@ async def update_rocket(
         return FlightUpdated(new_flight_id=str(successfully_updated_flight))
 
     @staticmethod
-    async def delete_flight(flight_id: int) -> "Union[FlightDeleted, HTTPException]":
+    async def delete_flight(
+        flight_id: int,
+    ) -> "Union[FlightDeleted, HTTPException]":
         """
         Delete a flight from the database.
 
@@ -285,7 +301,8 @@ async def delete_flight(flight_id: int) -> "Union[FlightDeleted, HTTPException]"
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         successfully_deleted_flight = await FlightRepository(
@@ -300,7 +317,9 @@ async def delete_flight(flight_id: int) -> "Union[FlightDeleted, HTTPException]"
         return FlightDeleted(deleted_flight_id=str(flight_id))
 
     @staticmethod
-    async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
+    async def simulate(
+        flight_id: int,
+    ) -> "Union[FlightSummary, HTTPException]":
         """
         Simulate a rocket flight.
 
@@ -318,7 +337,8 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
         ).get_flight()
         if not successfully_read_flight:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Flight not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Flight not found.",
             )
 
         try:
@@ -351,7 +371,9 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
             )
 
             _numerical_integration_settings = NumericalIntegrationSettings(
-                max_time="Maximum Allowed Flight Time: {:f} s".format(flight.max_time),
+                max_time="Maximum Allowed Flight Time: {:f} s".format(
+                    flight.max_time
+                ),
                 max_time_step="Maximum Allowed Time Step: {:f} s".format(
                     flight.max_time_step
                 ),
@@ -371,11 +393,15 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
             )
 
             _launch_rail_conditions = LaunchRailConditions(
-                rail_length="Launch Rail Length: {:.2f} m".format(flight.rail_length),
+                rail_length="Launch Rail Length: {:.2f} m".format(
+                    flight.rail_length
+                ),
                 flight_inclination="Launch Rail Inclination: {:.2f}°".format(
                     flight.inclination
                 ),
-                flight_heading="Launch Rail Heading: {:.2f}°".format(flight.heading),
+                flight_heading="Launch Rail Heading: {:.2f}°".format(
+                    flight.heading
+                ),
             )
 
             _surface_wind_conditions = SurfaceWindConditions(
@@ -413,17 +439,25 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
                     flight.rocket.motor.burn_out_time
                 ),
                 burnout_altitude="Altitude at burn out: {:.3f} m (AGL)".format(
-                    flight.z(flight.rocket.motor.burn_out_time) - flight.env.elevation
+                    flight.z(flight.rocket.motor.burn_out_time)
+                    - flight.env.elevation
                 ),
                 burnout_rocket_velocity="Rocket velocity at burn out: {:.3f} m/s".format(
                     flight.speed(flight.rocket.motor.burn_out_time)
                 ),
                 burnout_freestream_velocity="Freestream velocity at burn out: {:.3f} m/s".format(
                     (
-                        flight.stream_velocity_x(flight.rocket.motor.burn_out_time) ** 2
-                        + flight.stream_velocity_y(flight.rocket.motor.burn_out_time)
+                        flight.stream_velocity_x(
+                            flight.rocket.motor.burn_out_time
+                        )
+                        ** 2
+                        + flight.stream_velocity_y(
+                            flight.rocket.motor.burn_out_time
+                        )
                         ** 2
-                        + flight.stream_velocity_z(flight.rocket.motor.burn_out_time)
+                        + flight.stream_velocity_z(
+                            flight.rocket.motor.burn_out_time
+                        )
                         ** 2
                     )
                     ** 0.5
@@ -457,14 +491,17 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
                     flight.max_reynolds_number, flight.max_reynolds_number_time
                 ),
                 maximum_dynamic_pressure="Maximum Dynamic Pressure: {:.3e} Pa at {:.2f} s".format(
-                    flight.max_dynamic_pressure, flight.max_dynamic_pressure_time
+                    flight.max_dynamic_pressure,
+                    flight.max_dynamic_pressure_time,
                 ),
                 maximum_acceleration_during_motor_burn="Maximum Acceleration During Motor Burn: {:.3f} m/s² at {:.2f} s".format(
                     flight.max_acceleration, flight.max_acceleration_time
                 ),
                 maximum_gs_during_motor_burn="Maximum Gs During Motor Burn: {:.3f} g at {:.2f} s".format(
                     flight.max_acceleration
-                    / flight.env.gravity(flight.z(flight.max_acceleration_time)),
+                    / flight.env.gravity(
+                        flight.z(flight.max_acceleration_time)
+                    ),
                     flight.max_acceleration_time,
                 ),
                 maximum_acceleration_after_motor_burn="Maximum Acceleration After Motor Burn: {:.3f} m/s² at {:.2f} s".format(
@@ -491,17 +528,25 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
 
             if len(flight.impact_state) != 0:
                 _impact_conditions = ImpactConditions(
-                    x_impact_position="X Impact: {:.3f} m".format(flight.x_impact),
-                    y_impact_position="Y Impact: {:.3f} m".format(flight.y_impact),
-                    time_of_impact="Time of Impact: {:.3f} s".format(flight.t_final),
+                    x_impact_position="X Impact: {:.3f} m".format(
+                        flight.x_impact
+                    ),
+                    y_impact_position="Y Impact: {:.3f} m".format(
+                        flight.y_impact
+                    ),
+                    time_of_impact="Time of Impact: {:.3f} s".format(
+                        flight.t_final
+                    ),
                     impact_velocity="Velocity at Impact: {:.3f} m/s".format(
                         flight.impact_velocity
                     ),
                 )
             elif flight.terminate_on_apogee is False:
                 _impact_conditions = ImpactConditions(
                     time="Time: {:.3f} s".format(flight.solution[-1][0]),
-                    altitude="Altitude: {:.3f} m".format(flight.solution[-1][3]),
+                    altitude="Altitude: {:.3f} m".format(
+                        flight.solution[-1][3]
+                    ),
                 )
 
             if len(flight.parachute_events) == 0:
@@ -519,10 +564,14 @@ async def simulate(flight_id: int) -> "Union[FlightSummary, HTTPException]":
                     name = parachute.name.title()
                     events[name] = []
                     events[name].append(
-                        name + " Ejection Triggered at: {:.3f} s".format(trigger_time)
+                        name
+                        + " Ejection Triggered at: {:.3f} s".format(
+                            trigger_time
+                        )
                     )
                     events[name].append(
-                        name + " Parachute Inflated at: {:.3f} s".format(open_time)
+                        name
+                        + " Parachute Inflated at: {:.3f} s".format(open_time)
                     )
                     events[name].append(
                         name

lib/controllers/motor.py

@@ -116,16 +116,21 @@ async def get_motor(motor_id: int) -> "Union[Motor, HTTPException]":
         Raises:
             HTTP 404 Not Found: If the motor is not found in the database.
         """
-        successfully_read_motor = await MotorRepository(motor_id=motor_id).get_motor()
+        successfully_read_motor = await MotorRepository(
+            motor_id=motor_id
+        ).get_motor()
         if not successfully_read_motor:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Motor not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Motor not found.",
             )
 
         return successfully_read_motor
 
     @staticmethod
-    async def get_rocketpy_motor(motor_id: int) -> "Union[MotorPickle, HTTPException]":
+    async def get_rocketpy_motor(
+        motor_id: int,
+    ) -> "Union[MotorPickle, HTTPException]":
         """
         Get a rocketpy motor object encoded as jsonpickle string from the database.
 
@@ -138,10 +143,13 @@ async def get_rocketpy_motor(motor_id: int) -> "Union[MotorPickle, HTTPException
         Raises:
             HTTP 404 Not Found: If the motor is not found in the database.
         """
-        successfully_read_motor = await MotorRepository(motor_id=motor_id).get_motor()
+        successfully_read_motor = await MotorRepository(
+            motor_id=motor_id
+        ).get_motor()
         if not successfully_read_motor:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Motor not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Motor not found.",
             )
 
         successfully_read_rocketpy_motor = MotorController(
@@ -155,7 +163,9 @@ async def get_rocketpy_motor(motor_id: int) -> "Union[MotorPickle, HTTPException
             )
         )
 
-    async def update_motor(self, motor_id: int) -> "Union[MotorUpdated, HTTPException]":
+    async def update_motor(
+        self, motor_id: int
+    ) -> "Union[MotorUpdated, HTTPException]":
         """
         Update a motor in the database.
 
@@ -168,10 +178,13 @@ async def update_motor(self, motor_id: int) -> "Union[MotorUpdated, HTTPExceptio
         Raises:
             HTTP 404 Not Found: If the motor is not found in the database.
         """
-        successfully_read_motor = await MotorRepository(motor_id=motor_id).get_motor()
+        successfully_read_motor = await MotorRepository(
+            motor_id=motor_id
+        ).get_motor()
         if not successfully_read_motor:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Motor not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Motor not found.",
             )
 
         successfully_updated_motor = await MotorRepository(
@@ -186,7 +199,9 @@ async def update_motor(self, motor_id: int) -> "Union[MotorUpdated, HTTPExceptio
         return MotorUpdated(new_motor_id=str(successfully_updated_motor))
 
     @staticmethod
-    async def delete_motor(motor_id: int) -> "Union[MotorDeleted, HTTPException]":
+    async def delete_motor(
+        motor_id: int,
+    ) -> "Union[MotorDeleted, HTTPException]":
         """
         Delete a motor from the database.
 
@@ -199,10 +214,13 @@ async def delete_motor(motor_id: int) -> "Union[MotorDeleted, HTTPException]":
         Raises:
             HTTP 404 Not Found: If the motor is not found in the database.
         """
-        successfully_read_motor = await MotorRepository(motor_id=motor_id).get_motor()
+        successfully_read_motor = await MotorRepository(
+            motor_id=motor_id
+        ).get_motor()
         if not successfully_read_motor:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Motor not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Motor not found.",
             )
 
         successfully_deleted_motor = await MotorRepository(
@@ -230,10 +248,13 @@ async def simulate(motor_id: int) -> "Union[MotorSummary, HTTPException]":
         Raises:
             HTTP 404 Not Found: If the motor does not exist in the database.
         """
-        successfully_read_motor = await MotorRepository(motor_id=motor_id).get_motor()
+        successfully_read_motor = await MotorRepository(
+            motor_id=motor_id
+        ).get_motor()
         if not successfully_read_motor:
             raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="Motor not found."
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Motor not found.",
             )
 
         try:
@@ -249,7 +270,9 @@ async def simulate(motor_id: int) -> "Union[MotorSummary, HTTPException]":
                 + "{:.3f}".format(motor.propellant_initial_mass)
                 + " kg",
                 average_propellant_exhaust_velocity="Average Propellant Exhaust Velocity: "
-                + "{:.3f}".format(motor.exhaust_velocity.average(*motor.burn_time))
+                + "{:.3f}".format(
+                    motor.exhaust_velocity.average(*motor.burn_time)
+                )
                 + " m/s",
                 average_thrust="Average Thrust: "
                 + "{:.3f}".format(motor.average_thrust)