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":# TODO: schema applicable changes according to "type"
    },
)

platform_schema = api.model(
    "Platform",
    {
        "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",
        ),
        "active": fields.Boolean(
            required=False,
            description="platform in mission",
            example=True,
        ),
    },
)

squad_metadata_schema = api.model(
    "SquadMetadata",
    {
        "squad_ID": fields.Integer(
            required=True,
            description="platform serial number",
            example="ah-1",
        ),
        "no_of_platforms": fields.Integer(
            required=True,
            description="number of platform serial number",
            example="ah-1",
        ),
        "platforms": fields.List(
            fields.Nested(platform_schema),
            required=True,
            description="Metadata pf each platform",
        ),
        "squad_mission_type": fields.String(
            required=True,
            description="Survey or Detail",
            example="survey",
        ),
        "squad_state": fields.Boolean(
            required=True,
            description="True if given Squad is executing mission type "
            + "according to the Autonomy Engine plan",
            example=False,
        ),
    },
)

constraints_schema = api.model(
    "Constraints",
    # TODO: Should this be per platform instead of squad?
    {
        "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,
        ),
    },
)

environment_config_schema = api.model(  # TODO: Discuss how regions are defined
    "EnvironmentConfig",
    {
        "region_of_interest": fields.String(),
        "exclusion_zone": fields.String(),
    },
)
# Main Autonomy Configuration Schema
autonomy_configuration_schema = api.model(
    "Configuration",
    {
        "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",
        ),
        "environment": fields.Nested(
            environment_config_schema,
            required=False,
            description="Region of interest and exclusion zone",
        ),
    },
)

gps_schema = api.model(
    "GPS",
    {
        "gps_source": fields.Float(  # TODO: TBD with partners
            required=False,
            description="Source of gps position. E.g. Beacon",
            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 <insert units>",
            example="",
        ),
        "longitude": fields.Float(
            required=False,
            description="Longitude in <insert units>",
            example="",
        ),
        "depth": fields.Float(
            required=False,
            description="",
            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",
        ),
        "discuss_other_fields": fields.String(
            required=False,
            description="",  # we can track the version of the AE plan?
            example="",
        ),
    }
)
platform_status_message_schema = api.model(
    "PlatformStatusMessage",
    {
        "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",
        ),
        "version": fields.Integer(
            required=False,
            description="",  # we can track the version of the AE plan?
            example="",
        ),
        "platform_state": fields.String(
            # TODO: Define dictionary with potential STATES of each platform
            required=True,
            description="Current state executed by platform. E.g. "
            + "STOP, IDLE, ABORT.",
            example="IDLE",
        ),
        "mission_track_number": fields.Integer(
            required=False,
            description=(
                "Track number - stage in mission (e.g. "
                + "4 --> Waypoint 3 to Waypoint 4)"
            ),
            example=4,
        ),
        "range_to_go": fields.Float(
            required=False,
            description="Estimated distance to reach next waypoint",
            example=124.3,
        ),
        "c2_health_status": fields.String(
            required=False,
            description="Health status extracted by respective platform's C2 "
            + "if any diagnosis available checks on sensors",
            example="Warning",
        ),
        "gps_data": fields.Nested(gps_schema),
        "battery_voltage": fields.Float(
            required=True,
            description="Volts",
            example=23.0,
        ),
        "battery_current": fields.Float(
            required=False,
            description="Amps",
            example=1.2,
        ),
        "battery_current_per_hour": fields.Float(
            required=False,
            description="Amp-Hours",
            example=1.2,
        ),
        "battery_wattage": fields.Float(
            required=False,
            description="Watts",
            example=23.0,
        ),
        "battery_wattage_per_hour": fields.Float(
            required=False,
            description="Watt-Hours",
            example=23.0,
        ),
    },
)

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"],
}

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

ns1 = api.namespace("message", description="Message Header Format")
@ns1.route("/header")
class MessageHeader(Resource):
    @ns1.response(200, "Success", message_header_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"
)

@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("/ecosub")
class MissionPlanEcosub(Resource):
    @ns4.response(200, "Success", {})
    def get(self):
        pass
@ns4.route("/reav")
class MissionPlanReav(Resource):
    @ns4.response(200, "Success", {})
    def get(self):
        pass
@ns4.route("/autosubhover")
class MissionPlanAutosubHover(Resource):
    @ns4.response(200, "Success", {})
    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()