generate_swagger.py

"""
Generate swagger to view models/schemas.
Command:
1/ python3 generate_swagger.py
2/ Go to http://127.0.0.1:5000/soardocs
WARNING: API Endpoints are NOT functional. Purely for easy-reading.
"""
from flask import Flask, request
from flasgger import Swagger, LazyString
from flask_restx import Api, fields, Resource


app = Flask(__name__)
api = Api(app)

swagger_template = dict(
    info={
        "title": LazyString(lambda: "SoAR Backbone Message Formats"),
        "version": LazyString(lambda: "0.1"),
        "description": LazyString(
            lambda: "Backbone Message Format component for the Squad of Autonomous Robots (SoAR) message definitions."
        ),
    },
    host=LazyString(lambda: request.host),
)
message_header_schema = api.model(
    "MessageHeader",
    {
        "timestamp": fields.DateTime(
            required=True,
            description="Timestamp of message",
            example="2022-11-16T00:00:00Z",
        ),
        "source": fields.String(
            required=True,
            description="Where is this message from",
            example="autonomy_engine",
        ),
        "destination": fields.String(
            required=True,
            description="What is the destination of this message",
            example="ah-1",
        ),
        "encoded": fields.Boolean(
            required=True,
            description="Indicate that message raw (encoded) or decoded. "
            + "Options: encoded=True, decoded=False",
            example=False,
        ),
        "type": fields.String(
            required=True,
            description="Type of message",
            example="platform_status",
        ),
        "payload": fields.Raw(required=True)  # TODO: schema applicable changes according to "type"
    },
)

full_message_schema = api.model(
    "FullMessageSchema",
    {
        "uuid": fields.DateTime(   # add this
            required=True,
            description="Timestamp of message",
            example="2022-11-16T00:00:00Z",
        ),
        "timestamp": fields.DateTime(
            required=True,
            description="Timestamp of message",
            example="2022-11-16T00:00:00Z",
        ),
        "source": fields.String(
            required=True,
            description="Where is this message from",
            example="autonomy_engine",
        ),
        "destination": fields.String(
            required=True,
            description="What is the destination of this message",
            example="ah-1",
        ),
        "encoded": fields.Boolean(
            required=True,
            description="Indicate that message raw (encoded) or decoded. "
            + "Options: encoded=True, decoded=False",
            example=False,
        ),
        "type": fields.String(
            required=True,
            description="Type of message",
            example="platform_status",
        ),
        "payload": fields.Raw(
            required=True,
            description="Content of Message",
            # example="{}",
        ),
    }
)


constraints_schema = api.model(
    "ConstraintsSchema",
    {
        "min_altitude": fields.Float(
            required=True,
            description="Minimum altitude set for squad.",
            example=15.2,
        ),
        "min_velocity": fields.Float(
            required=True,
            description="Minimum velocity set for squad.",
            example=0.1,
        ),
        "max_velocity": fields.Float(
            required=True,
            description="Maximum altitude set for squad.",
            example=0.9,
        ),
    },
)

platform_schema = api.model(
    "PlatformSchema",
    {
        "platform_ID": fields.Integer(
            required=True,
            description="unique identifier for platform",
            example="ah-1",
        ),
        "serial": fields.Integer(
            required=True,
            description="platform serial number",
            example="ah-1",
        ),
        "model": fields.Integer(
            required=True,
            description="platform serial number",
            example="ah-1",
        ),
        "constraints": fields.Nested(constraints_schema),
        "active": fields.Boolean(
            required=False,
            description="When a platform is in deployment (executing a mission plan) this should be True",
            example=True,
        ),
    },
)

squad_metadata_schema = api.model(
    "SquadMetadataSchema",
    {
        "squad_ID": fields.Integer(
            required=True,
            description="Identifier of given squad",
            example="ah-1",
        ),
        "no_of_platforms": fields.Integer(
            required=True,
            description="number of platforms",
            example="ah-1",
        ),
        "platforms": fields.List(
            fields.Nested(platform_schema),
            required=True,
            description="Squad consists of these platforms",
        ),
        "squad_mission_type": fields.String(
            required=True,
            description="Mission of given squad: `tracking`, `survey`, `inspection`",
            example="survey",
        ),
        "squad_state": fields.Boolean(
            required=True,
            description="In execution, Waiting.. <define further>",
            example=False,
        ),
    },
)

gps_schema = api.model(
    "GPS",
    {
        "gps_source": fields.Float(  # TODO: TBD with partners
            required=False,
            description="Source of gps position. E.g. USBL (external), platform itself (internal)",
            example="internal",
        ),
        "latitude_type": fields.String(
            required=False,
            description="",
            example="",
        ),
        "longitude_type": fields.String(
            required=False,
            description="",
            example="",
        ),
        "latitude": fields.Float(
            required=False,
            description="Latitude in <DEFINE UNITS>",
            example="",
        ),
        "longitude": fields.Float(
            required=False,
            description="Longitude in <DEFINE UNITS>",
            example="",
        ),
        "depth": fields.Float(
            required=False,
            description="Depth in <DEFINE UNITS>",
            example="",
        ),
        "altitude": fields.Float(
            required=False,
            description="Altitude in <DEFINE UNITS>",
            example="",
        ),
        # "gps_fix_seconds_ago": fields.Float(
        #     required=False,
        #     description="",
        #     example="",
        # ),
    },
)

observation_schema = api.model(
    "Observation",
    {
        "message": fields.Nested(
            message_header_schema,
            required=True,
            description="Message header",
        ),
        "platform": fields.Nested(platform_schema),
        "time": fields.String(
            required=True,
            description="Timestamp of message",
            example="2022-11-16T00:00:00Z",
        ),
        # "observation_type" ==> payloads tied to different types maybe?
        # properties of each observation?
        "points_of_interest": fields.Float(
            required=False,
            description="Points from features of interest identified by platform if any found. DEFINE FORMAT.",
            example="",
        ),
        "region_surveyed": fields.Float(
            required=False,
            description="Region surveyed by given platform. DEFINE FORMAT. GEOJSON?",
            example="",
        ),
        "quality_of_points": fields.Float(
            required=False,
            description="Quality/strength of points from features of interest identified by platform. DEFINE FORMAT.",
            example=0.98,
        ),
        "additional_data": fields.Raw(
            required=False,
            description="Placeholder field for any additional data",
            example={"sensor_payload": False},
        ),
    }
)
sensor_schema = api.model(
    "Sensor",
    {
        "sensor_name": fields.String(
            required=False,
            description="Name of sensor (e.g. MBES for AH1 and SideScan for Ecosub",
            example="MBES",
        ),
        "sensor_status": fields.Boolean(
            required=False,
            description="Sensor switched on (True) or off (False)",
            example=True,
        ),
        "additional_data": fields.Raw(
            required=False,
            description="Add any additional sensor-related data here.",
            example={"sensor_loadtime_seconds": 30.0},
        ),
    }
)
platform_status_message_schema = api.model(
    "platformStatusMessage",
    {
        "message": fields.Nested(
            message_header_schema,
            required=True,
            description="Message header",
        ),
        "platform": fields.Nested(platform_schema),
        "platform_state": fields.String(
            # TODO: Define dictionary with potential STATES of each platform
            required=False,
            description="Current state executed by platform. E.g. "
            + "STOP, IDLE, ABORT.",
            example="IDLE",
        ),
        "autonomy_mission_ID": fields.Integer(
            required=False,
            description="Last mission plan ID (according to Autonomy Engine's mission plan number) executed by platform",
            example=1,
        ),
        "mission_track_ID": fields.Integer(
            required=False,
            description=(
                "Track number - stage in mission (e.g. "
                + "4 --> Waypoint 3 to Waypoint 4)"
            ),
            example=4,
        ),
        "mission_action_ID": fields.Integer(
            required=False,
            description="to add description",
            example=1,
        ),
        "range_to_go": fields.Float(
            required=False,
            description="Estimated distance to reach next waypoint",
            example=124.3,
        ),
        "speed_over_ground": fields.Float(
            required=False,
            description="",
            example=124.3,
        ),
        "water_current_velocity": fields.Float(
            required=False,
            description="",
            example=124.3,
        ),
        "thrust_applied": fields.Float(
            required=False,
            description="TODO: Needs further consideration",
            example=124.3,
        ),
        "health_status": fields.String(
            required=False,
            description="Health status extracted by respective platform "
            + "if any diagnosis available checks on sensors",
            example="Warning",
        ),
        "gps_data": fields.List(
            fields.Nested(gps_schema),  # TODO: TBD Do we want a list of gps readings to allow > 1 reading i.e. platform + usbl
            required=True,
            description="GPS of each platform",
        ),
        "localisation_error": fields.Float(
            required=False,
            description="Localisation error at last USBL update.",
            example="",
        ),
        "usbl_fix_seconds_ago": fields.Float(
            required=False,
            description="",
            example="",
        ),
        "battery_remaining_capacity": fields.Float(
            required=True,
            description="Battery remaining capacity % provided by respective platform/C2.",
            example=80.0,
        ),
        "sensor_config": fields.Nested(sensor_schema),  # TODO: TBD Do we want a list of sensors to allow > 1 sensor
    },
)

mission_plan_schema = api.model(
    "MissionPlan",
    {
        "message": fields.Nested(
            message_header_schema,
            required=True,
            description="Message header",
        ),
        "ID": fields.Integer(
            required=True,
            description="Unique identifier tagged to version of this"
            + " configuration plan",
            example=3,
        ),
        "time": fields.String(
            required=True,
            description="",
            example="",
        ),
        "platform_ID": fields.Integer(
            required=False,
            description="Details of each squad",
        ),
        "payload": fields.Raw(
            required=True,
        )
    },
)


swagger_config = {
    "headers": [],
    "specs": [
        {
            "endpoint": "swagger",
            "route": "/swagger.json",
            "rule_filter": lambda rule: True,
            "model_filter": lambda tag: True,
        }
    ],
    "static_url_path": "/flasgger_static",
    "swagger_ui": True,
    "specs_route": "/soardocs/",
    "swagger": "2.0",
    "basePath": "/soar",
    "info": {
        "title": "soar",
        "version": "0.1",
        "description": "SoAR message schemas",
    },
    "produces": ["application/json"],
    "consumes": ["application/json"],
}

autonomy_configuration_schema = api.model(
    "SquadConfigurationSchema",
    {
        "message": fields.Nested(
            message_header_schema,
            required=True,
            description="Message header",
        ),
        "ID": fields.Integer(
            required=True,
            description="Unique identifier tagged to version of this"
            + " configuration plan",
            example=3,
        ),
        "time": fields.String(
            required=True,
            description="",
            example="",
        ),
        "squads": fields.Nested(
            squad_metadata_schema,
            required=False,
            description="Details of each squad",
        ),
        "region_of_operation": fields.Raw(
            required=True,
            description="Using GEOJSON, exact region of interest EXCLUDING exclusion zone",
            example={
                "type": "FeatureCollection",
                "features": [
                    {
                    "type": "Feature",
                    "properties": {},
                    "geometry": {
                        "coordinates": [
                        [
                            [
                            -4.187143188645706,
                            50.37072283932642
                            ],
                            [
                            -4.202697005964865,
                            50.368816892405874
                            ],
                            [
                            -4.203156724702808,
                            50.365640144076906
                            ],
                            [
                            -4.19449868846155,
                            50.362267670845654
                            ],
                            [
                            -4.187373048015246,
                            50.35933489062816
                            ],
                            [
                            -4.1859938917992565,
                            50.36207215780212
                            ],
                            [
                            -4.188598964651192,
                            50.36471151593676
                            ],
                            [
                            -4.192506573927574,
                            50.367350727294706
                            ],
                            [
                            -4.188675584440716,
                            50.36837704762925
                            ],
                            [
                            -4.187143188645706,
                            50.37072283932642
                            ]
                        ]
                        ],
                        "type": "Polygon"
                        # "projection": coordinate system used
                    }
                    }
                ]
            }
        ),
    },
)


swagger = Swagger(app, template=swagger_template, config=swagger_config)

ns1 = api.namespace("message", description="Message Wrapper (Full Message Schema) Format")
@ns1.route("/wrapper")
class MessageWrapper(Resource):
    @ns1.response(200, "Success", full_message_schema)
    def get(self):
        pass


ns2 = api.namespace("platform_status", description="platform Status Message Format")
@ns2.route("")
class platformStatus(Resource):
    @ns2.response(200, "Success", platform_status_message_schema)
    def get(self):
        pass


ns3 = api.namespace(
    "autonomy_configuration", description="Autonomy Configuration Format. Do we want region of interest to be per squad, per platform or in the main schema?"
)

@ns3.route("")
class AutonomyConfiguration(Resource):
    @ns3.response(200, "Success", autonomy_configuration_schema)
    def get(self):
        pass


# @api.route('/mission-plan/<str:platform_type')
# @api.doc(params={"platform_type": "The type of platform of the mission plan to target."})
ns4 = api.namespace("mission_plan", description="Mission Plan Format Per platform")

@ns4.route("")
class MissionPlan(Resource):
    @ns4.response(200, "Success", mission_plan_schema)
    def get(self):
        pass

@ns4.route("/ecosub")
class MissionPlanEcosub(Resource):
    @ns4.response(200, "Success", mission_plan_schema)
    def get(self):
        pass
@ns4.route("/reav")
class MissionPlanReav(Resource):
    @ns4.response(200, "Success", mission_plan_schema)
    def get(self):
        pass
@ns4.route("/autosubhover")
class MissionPlanAutosubHover(Resource):
    @ns4.response(200, "Success", mission_plan_schema)
    def get(self):
        pass

ns5 = api.namespace("observation", description="Observation Format --> Per platform or generic?")
@ns5.route("")
class Observation(Resource):
    @ns5.response(200, "Success", observation_schema)
    def get(self):
        pass

if __name__ == "__main__":
    app.run()