"""
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()