Commit b21a71af authored by Laurens Oostwegel's avatar Laurens Oostwegel Committed by Laurens Oostwegel
Browse files

Add possibility to set and get main material in API

parent 5140208c
Pipeline #52371 passed with stage
in 3 minutes and 38 seconds
%% Cell type:markdown id:c65dfc7c tags:
## Introduction
This API communicates with the loss-calculator. The API can be used to run the loss-calculator, add manual damage assessments and to update a mission planning. A scenario is saved on the serverside that can be updated by anyone, only by referencing the `scenario` name. The database can be downloaded via the API too. A complete list of functions that are explained in this Jupyter Notebook file are:
- `get_scenarios`
- `create_scenario`
- `run_scenario`
- `create_damage_assessment_report`
- `update_mission_planning`
- `get_scenario_database`
%% Cell type:code id:5105b219 tags:
``` python
import requests
# General params
base_url = "https://losscalculator.openbuildingmap.org/"
headers = {"Content-Type": "application/json", "Accept": "text/plain"}
# Scenario name that is referenced throughout the process
# This scenario is only added once (in create_scenario).
scenario = "Athens"
```
%% Cell type:markdown id:1f7bcfea tags:
### Scenarios
When an earthquake happens, a scenario can be created using a QuakeML file. For this, the post request `create_scenario` can be used. A separate Spatialite database will be created from the source exposure model database. Once a scenario is created, it can be referenced to update the created database.
#### Get all scenarios
First: let's see which scenarios are created in the API already, before we create a new one. This can be achieved using the `get_scenario` request.
The input for the GET request is:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/get_scenarios`)
- `headers`: API headers
%% Cell type:code id:7993523c tags:
``` python
r = requests.get(
url=base_url + "get_scenarios",
headers=headers
)
# Print all scenarios
print(r.content)
```
%% Cell type:markdown id:51aae79e tags:
#### Create a scenario
If your scenario is not in the scenario list, you can create a new scenario, using the `create_scenario` post request. The input of the POST request is:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/create_scenario`)
- `headers`: API headers
- `data`: QuakeML file of the requested area and earthquake.
- `params`:
- `scenario`: Scenario name identifier
- `description`: Description of the scenario
- `exposure_model_config` (Optional): The exposure model configuration that should be used. This configuration can be found inside the API and contains the taxonomy mapping and fragility curves of the exposure model. For now this defaults to the only option that is available: `Europe`.
%% Cell type:code id:8928e4cd tags:
``` python
bbox = {
"lon_min": 23.64759,
"lat_min": 37.97848,
"lon_max": 23.68521,
"lat_max": 38.01400,
}
# Create scenario
r = requests.post(
url=base_url + "create_scenario",
headers=headers,
params={
"scenario": scenario,
"description": "Test model of Athens",
},
json=bbox,
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:markdown id:3aa8e93a tags:
#### Run a scenario with the loss-calculator
When a scenario is created, the loss-calculator can be run with post request `run_scenario`. The input is:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/run_scenario`)
- `headers`: API headers
- `data`: QuakeML file of the scenario earthquake.
- `params`:
- `scenario`: Scenario name identifier
- `assessment_source`: Name of the assessment source. The `assessment_source` is a (unique) identifier of the damage assessment. For example, in the case of the loss-calculator, it could be `loss-calculator_*identifier*`.
It is possible to run the scenario multiple times with different QuakeML files. IMPORTANT: the assessment_source should be a unique name, otherwise it will resolve in an error.
%% Cell type:code id:b5ca8633 tags:
``` python
# Assessment source names that should be different for each request
loss_assessment_source = "loss-calculator"
# QuakeML input file
QuakeML = open("athens.quakeml", "rb")
# Run scenario
requests.post(
url=base_url + "run_scenario",
headers={"Content-Type": "application/text"},
data=QuakeML,
params={"scenario": scenario, "assessment_source": loss_assessment_source},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:markdown id:5a07e893 tags:
#### Create damage report
After a manual damage assessment, a damage report can be submitted to the loss-calculator API. The input file is the damage report JSON file. The damage report includes a `dateTimeStamp`; a `damageScale` with a name and all the included damage states; a list of `osmBuildingIds` that include the probability of the `damageValues` for each damage state and an optional comment.
An example damage report input file:
```json
{
"dateTimeStamp": "87890954331",
"damageScales": {"Binary":
{"states": ["no damage", "damage"] },
"Multi":
{"states": ["no damage", "slight damage","..."]}
},
"assessmentSources": ["MM", "automatic"],
"osmBuildingIds": {
"317581202": {
"damages":
[
{
"Binary": [0.1, 0.9],
"assessment_source": "automatic",
"status": "finished"
},
{
"Multi": [0.1,0.3,0.5],
"assessment_source": "MM",
"status": "pending"
}
]
"comment": "optional comment"
},
},
"90237856": {
"damageValues": [
0.5,
0.5
]
}
}
}
```
The POST request inputs are:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/create_damage_assessment_report`)
- `headers`: API headers
- `files`:
- `damage_report`: the damage report JSON file
- `params`:
- `scenario`: Scenario name identifier
- `assessment_source`: Name of the manual assessment source
Again, it is important to note that the assessment_source name needs to be unique.
%% Cell type:code id:bc0bc7a2 tags:
``` python
# Assessment source names that should be different for each request
assessment_source = "damage_report"
manual_assessment_source = "damage_report"
# Create damage assessment report
requests.post(
r = requests.post(
url=base_url + "create_damage_assessment_report",
params={"scenario": scenario, "assessment_source": manual_assessment_source},
files={"damage_report": open("damage_report.json", "rb")},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:code id:ce2343b2 tags:
``` python
r = requests.get(
url=base_url + "get_damage_assessment_report",
params={"scenario": scenario},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:code id:d1276b79 tags:
``` python
status = "finished"
r = requests.get(
url=base_url + "get_damage_assessment_status",
params={"scenario": scenario, "status": status},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:code id:f5812c13 tags:
``` python
r = requests.get(
url=base_url + "get_building_main_material",
params={"scenario": scenario},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:markdown id:2e2815a5 tags:
#### Update mission planning
The mission planning has little to do with the loss-calculator itself, but is a tool to update the status of buildings in a mission (for example: drone mapping) to a certain state. The states now are: `0`: unvisited; `1`: being visited; `2`: visited.
The input of the POST reuqest are:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/update_mission_planning_status`)
- `headers`: API headers
- `data`: A list of OSM building IDs that need to be updated
- `params`:
- `scenario`: Scenario name identifier
- `status`: Status the OSM buildings need to be updated with (`0`, `1` or `2`).
%% Cell type:code id:606b5ebb tags:
``` python
# Update mission planning
updated_osm_ids = [82482673, 209090749, 216092203, 216092204]
updated_status = 1
requests.post(
url=base_url + "update_mission_planning_status",
data=json.dumps(updated_osm_ids),
params={"scenario": scenario, "status": updated_status},
)
# If the request resolves in an error, printing the content will give you some clues about what went wrong.
print(r.content)
```
%% Cell type:markdown id:a7cb0e7d tags:
#### Get the resulting exposure database
The last step is to get the resulting exposure database. This can be done after every intermediary step, or at the end.
The input of the GET request is:
- `url`: API URL (`https://losscalculator.openbuildingmap.org/get_scenario_database`)
- `headers`: API headers
- `params`:
- `scenario`: Scenario name identifier.
%% Cell type:code id:dd45b6e6 tags:
``` python
# Spatialite output file
db_output_file = "athens.db"
# Get Spatialite (SQLite) database output
output = requests.get(
url=base_url + "get_scenario_database",
params={"scenario": scenario},
)
with open(db_output_file, "wb") as f:
f.write(output.content)
```
......
......@@ -85,6 +85,27 @@ def main():
files={"damage_report": open("damage_report.json", "rb")},
)
# Get full damage assessment report (including materials)
output = requests.get(
url=base_url + "get_damage_assessment_report",
params={"scenario": scenario},
)
print(output.content)
# Get damage assessments report of one status
output = requests.get(
url=base_url + "get_damage_assessment_status",
params={"scenario": scenario, "status": "pending"},
)
print(output.content)
# Get main material and probability
output = requests.get(
url=base_url + "get_building_main_material",
params={"scenario": scenario},
)
print(output.content)
# Get Spatialite (SQLite) database output
output = requests.get(
url=base_url + "get_scenario_database",
......
......@@ -49,7 +49,11 @@
"status": "pending",
"comment": "optional comment"
}
]
],
"material": {
"mainMaterial": "Concrete",
"mainMaterialProbability": 0.5
}
},
"454951165": {
"damage": [
......@@ -75,7 +79,11 @@
"assessmentSourceMethod": "micro_mapping",
"status": "pending"
}
]
],
"material": {
"mainMaterial": "Wood",
"mainMaterialProbability": 0.6
}
}
}
}
......@@ -207,10 +207,14 @@ class LosscalculatorApi(FastAPI):
"""
if not self.scenario_exists(scenario_name):
return None
raise HTTPException(
status_code=500, detail=f"Scenario {scenario_name} does not exist"
)
if not self.exposure_database_exists(scenario_name):
return None
raise HTTPException(
status_code=500, detail=f"Database of scenario {scenario_name} does not exist"
)
return SpatiaLiteExposure(
database_filepath=self.get_scenario_config(scenario_name)[
......@@ -492,7 +496,13 @@ async def create_exposure_database(
)
# Add status column to table Entity for mission planning
sql_statement = "ALTER TABLE Entity ADD status INTEGER;"
sql_statement = "ALTER TABLE Entity ADD status INTEGER"
exposure_model.cursor.execute(sql_statement)
# Add material column to table Entity for material probabilities
sql_statement = "ALTER TABLE Entity ADD main_material VARCHAR"
exposure_model.cursor.execute(sql_statement)
sql_statement = "ALTER TABLE Entity ADD main_material_probability VARCHAR"
exposure_model.cursor.execute(sql_statement)
# Add status column to table Assessment for assessment reports
......@@ -621,16 +631,7 @@ async def update_mission_planning_status(
OSM ids that should be updated with the status code
"""
if not app.scenario_exists(scenario):
logger.debug(f"Scenario {scenario} does not exist")
raise HTTPException(status_code=500, detail=f"Scenario {scenario} does not exist")
logger.info(osm_ids)
db = app.get_exposure_model_database(scenario)
if db is None:
raise HTTPException(
status_code=500, detail=f"Database of scenario {scenario} does not exist"
)
logger.debug("Connecting to database...")
db.connect(init_spatial_metadata=False)
......@@ -671,13 +672,6 @@ async def create_damage_assessment_report(
# Connect to scenario database
db = app.get_exposure_model_database(scenario)
# Raise error if the database is not returned
if db is None:
logger.debug(f"Database of scenario {scenario} does not exist")
raise HTTPException(
status_code=500, detail=f"Database of scenario {scenario} does not exist"
)
# Connect to the database
logger.debug("Connecting to database...")
db.connect(init_spatial_metadata=False)
......@@ -736,8 +730,21 @@ async def create_damage_assessment_report(
# Retrieve entity ID using the OSM ID.
entity_id = db.get_entity_id(osm_id=int(osm_id))
# Retrieve main material & probability
material = vals.get("material", None)
if material:
main_material = material["mainMaterial"]
main_probability = material["mainMaterialProbability"]
sql_statement = f"""
UPDATE Entity
SET main_material='{main_material}',
main_material_probability='{main_probability}'
WHERE id = {entity_id}
"""
db.cursor.execute(sql_statement)
# Iterate through each damage assessment of the building
for damage_assessment in vals["damage"]:
for damage_assessment in vals.get("damage", []):
assessment_source_method = damage_assessment["assessmentSourceMethod"]
assessment_source_id = assessment_source_ids[assessment_source_method]
damage_scale = damage_assessment["damageScale"]
......@@ -802,11 +809,66 @@ async def create_damage_assessment_report(
return {}
@app.get("/get_damage_assessment_report")
async def get_damage_assessment_report(scenario: str):
"""
Get all damage assessments in the exposure database. Returns the ID, Quadkey, geometry, main
material and main material probability of the entity and the ID, status and timestamp of the
damage assessment.
Args:
scenario (str):
Name of the scenario.
"""
# Connect to the scenario database
db = app.get_exposure_model_database(scenario)
# Connect to the database
logger.debug("Connecting to database...")
db.connect(init_spatial_metadata=False)
sql_statement = """
SELECT Entity.osm_id, Entity.quadkey, ST_AsText(Entity.geom),
Entity.main_material, Entity.main_material_probability,
Assessment.id, Assessment.status, Assessment.date
FROM Assessment
INNER JOIN Entity
ON Assessment.entity_id = Entity.id
INNER JOIN AssessmentSource
ON Assessment.assessment_source_id = AssessmentSource.id
WHERE Assessment.status IS NOT NULL
ORDER BY Entity.quadkey, Entity.osm_id, Assessment.date DESC
"""
db.cursor.execute(sql_statement)
result = db.cursor.fetchall()
db.close()
cols = [
"osmBuildingIds",
"quadkey",
"geometry",
"mainMaterial",
"mainMaterialProbability",
"assessmentId",
"status",
"timestamp",
]
if len(result) == 0:
return {}
else:
return {
"entities": [{var: attr for var, attr in zip(cols, entity)} for entity in result]
}
@app.get("/get_damage_assessment_status")
async def get_damage_assessment_status(scenario: str, status: str = None):
"""
Get all damage assessments in the exposure database. Returns the ID, Quadkey,
geometry of the entity and the ID, status and timestamp of the damage assessment.
Get all damage assessments in the exposure database. Returns the ID, Quadkey, geometry of
the entity and the ID, status and timestamp of the damage assessment.
Args:
scenario (str):
......@@ -819,13 +881,6 @@ async def get_damage_assessment_status(scenario: str, status: str = None):
# Connect to the scenario database
db = app.get_exposure_model_database(scenario)
# Raise error if the database is not returned
if db is None:
logger.debug(f"Database of scenario {scenario} does not exist")
raise HTTPException(
status_code=500, detail=f"Database of scenario {scenario} does not exist"
)
# Connect to the database
logger.debug("Connecting to database...")
db.connect(init_spatial_metadata=False)
......@@ -838,36 +893,74 @@ async def get_damage_assessment_status(scenario: str, status: str = None):
sql_statement = """
SELECT Entity.osm_id, Entity.quadkey, ST_AsText(Entity.geom),
Assessment.status, Assessment.date
Assessment.id, Assessment.status, Assessment.date
FROM Assessment
INNER JOIN Entity
ON Assessment.entity_id = Entity.id
INNER JOIN AssessmentSource
ON Assessment.assessment_source_id = AssessmentSource.id
ORDER BY Entity.quadkey, Entity.osm_id, Assessment.date DESC
"""
if status_id:
sql_statement += f"WHERE status = {status_id}"
sql_statement += f"WHERE Assessment.status = {status_id} "
sql_statement += "ORDER BY Entity.quadkey, Entity.osm_id, Assessment.date DESC"
db.cursor.execute(sql_statement)
result = db.cursor.fetchall()
db.close()
cols = [
"osmBuildingIds",
"quadkey",
"geometry",
"assessmentId",
"status",
"timestamp",
]
if len(result) == 0:
return {}
else:
return {
"entities": list(
map(
lambda osm_id, quadkey, geom, assessment_id, status, timestamp: {
"osm_id": osm_id,
"quadkey": quadkey,
"geom": geom,
"assessment_id": assessment_id,
"status": status,
"timestamp": timestamp,
},
result,
)
)
"entities": [{var: attr for var, attr in zip(cols, entity)} for entity in result]
}
@app.get("/get_building_main_material")
async def get_building_main_material(scenario: str):
"""
Get all materials in the exposure database. Returns the ID, Quadkey, geometry, main
material and main material probability of all entities that have a main material assigned.
Args:
scenario (str):
Name of the scenario.
"""
# Connect to the scenario database
db = app.get_exposure_model_database(scenario)
# Connect to the database
logger.debug("Connecting to database...")
db.connect(init_spatial_metadata=False)
# Get entity materials
sql_statement = """
SELECT osm_id, quadkey, ST_AsText(geom), main_material,
main_material_probability
FROM Entity
WHERE main_material IS NOT NULL
"""
db.cursor.execute(sql_statement)
result = db.cursor.fetchall()
# closing connection
db.close()
cols = ["osmBuildingIds", "quadkey", "geometry", "mainMaterial", "mainMaterialProbability"]
if len(result) == 0:
return {}
else:
return {
"entities": [{var: attr for var, attr in zip(cols, entity)} for entity in result]
}
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment