src.app

The main web application that serves the Digital Twin to the web through a Rest API.

Attributes

`app`
`WWW_HOST`
`WWW_PORT`
`SWAGGER_URL`
`API_URL`
`swagger_ui_blueprint`
`gunicorn_logger`

Functions

`check_celery_alive`(→ Callable[Ellipsis, flask.Response])	Function decorator to check if the Celery workers are running and return INTERNAL_SERVER_ERROR if they are down.
`index`(→ flask.Response)	Ping this endpoint to check that the flask app is running
`health_check`(→ flask.Response)	Ping this endpoint to check that the server is up and running
`get_status`(→ flask.Response)	Retrieves status of a particular Celery backend task.
`remove_task`(→ flask.Response)	Deletes and stops a particular Celery backend task.
`generate_model`(→ flask.Response)	Generates a flood model for a given area.
`get_valid_flood_model_parameters`(→ flask.Response)	Get information on valid flood model parameters based on the valid values in the database.
`create_wkt_from_coords`(→ str)	Takes two points and creates a wkt bbox string from them
`get_depth_at_point`(→ flask.Response)	Finds the depths and times at a particular point for a given completed model output task.
`retrieve_building_flood_status`(→ flask.Response)	Retrieves information on building flood status, for a given flood model output id.
`serve_model_output`(→ flask.Response)	Serve the specified model output as a raw file.
`refresh_lidar_data_sources`()	Updates LiDAR data sources to the most recent.
`valid_coordinates`(→ bool)	Validates coordinates are in the valid range of WGS84

Module Contents

src.app.app

src.app.WWW_HOST

src.app.WWW_PORT

src.app.check_celery_alive(f: Callable[Ellipsis, flask.Response]) → Callable[Ellipsis, flask.Response]

Function decorator to check if the Celery workers are running and return INTERNAL_SERVER_ERROR if they are down.

Parameters:: f (Callable[..., Response]) – The view function that is being decorated
Returns:: Response is SERVICE_UNAVAILABLE if the celery workers are down, otherwise continue to function f
Return type:: Callable[…, Response]

src.app.SWAGGER_URL = '/swagger'

src.app.API_URL = '/static/api_documentation.yml'

src.app.swagger_ui_blueprint

src.app.index() → flask.Response

Ping this endpoint to check that the flask app is running Supported methods: GET

Returns:: The HTTP Response. Expect OK if health check is successful
Return type:: Response

src.app.health_check() → flask.Response

Ping this endpoint to check that the server is up and running Supported methods: GET

Returns:: The HTTP Response. Expect OK if health check is successful
Return type:: Response

src.app.get_status(task_id: str) → flask.Response

Retrieves status of a particular Celery backend task. Supported methods: GET

Parameters:: task_id (str) – The id of the Celery task to retrieve status from
Returns:: JSON response containing taskStatus
Return type:: Response

src.app.remove_task(task_id: str) → flask.Response

Deletes and stops a particular Celery backend task. Supported methods: DELETE

Parameters:: task_id (str) – The id of the Celery task to remove
Returns:: ACCEPTED is the expected response
Return type:: Response

src.app.generate_model() → flask.Response

Generates a flood model for a given area. Supported methods: POST

Returns:: ACCEPTED is the expected response. Response body contains Celery taskId
Return type:: Response

src.app.get_valid_flood_model_parameters() → flask.Response

Get information on valid flood model parameters based on the valid values in the database.

Returns:: JSON response describing the valid flood model parameters.
Return type:: Response

src.app.create_wkt_from_coords(lat1: float, lng1: float, lat2: float, lng2: float) → str

Takes two points and creates a wkt bbox string from them

Parameters:

lat1 (float) – latitude of first point
lng1 (float) – longitude of first point
lat2 (float) – latitude of second point
lng2 (float) – longitude of second point

Returns:

bbox in wkt form generated from the two coordinates

Return type:

str

src.app.get_depth_at_point(task_id: str) → flask.Response

Finds the depths and times at a particular point for a given completed model output task. Supported methods: GET Required query param values: “lat”: float, “lng”: float

Parameters:: task_id (str) – The id of the completed task for generating a flood model.
Returns:: Returns JSON response in the form {“depth”: Array<number>, “time”: Array<number>} representing the values for the given point.
Return type:: Response

src.app.retrieve_building_flood_status(model_id: int) → flask.Response

Retrieves information on building flood status, for a given flood model output id. It is recommended to use the geoserver API if it is possible, since this is a proxy around that.

Parameters:: model_id (int) – The ID of the flood output model to be queried
Returns:: Returns GeoJSON building layer for the area of the flood model output. Has a property “is_flooded” to designate if a building is flooded in that scenario or not
Return type:: Response

src.app.serve_model_output(model_id: int) → flask.Response

Serve the specified model output as a raw file.

Parameters:: model_id (int) – The ID of the model output to be served.
Returns:: HTTP Response containing the model output file.
Return type:: Response

src.app.refresh_lidar_data_sources()

Updates LiDAR data sources to the most recent. Web-scrapes OpenTopography metadata to update the datasets table containing links to LiDAR data sources. Takes a long time to run but needs to be run periodically so that the datasets are up to date. Supported methods: POST

Returns:: ACCEPTED is the expected response. Response body contains Celery taskId
Return type:: Response

src.app.valid_coordinates(latitude: float, longitude: float) → bool

Validates coordinates are in the valid range of WGS84 (-90 < latitude <= 90) and (-180 < longitude <= 180)

Parameters:

latitude (float) – The latitude part of the coordinate
longitude (float) – The longitude part of the coordinate

Returns:

True if both latitude and longitude are within their valid ranges.

Return type:

bool

src.app.gunicorn_logger