+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
18 """Logic for FMTM project routes."""
+
+
+
+
+
+
+
25 from asyncio
import gather
+
26 from concurrent.futures
import ThreadPoolExecutor, wait
+
27 from importlib.resources
import files
as pkg_files
+
28 from io
import BytesIO
+
29 from typing
import List, Optional, Union
+
+
+
+
+
+
35 import shapely.wkb
as wkblib
+
36 import sozipfile.sozipfile
as zipfile
+
+
38 from asgiref.sync
import async_to_sync
+
39 from fastapi
import File, HTTPException, UploadFile
+
40 from fastapi.concurrency
import run_in_threadpool
+
41 from fmtm_splitter.splitter
import split_by_sql, split_by_square
+
42 from geoalchemy2.shape
import from_shape, to_shape
+
43 from geojson.feature
import Feature, FeatureCollection
+
44 from loguru
import logger
as log
+
45 from osm_fieldwork.basemapper
import create_basemap_file
+
46 from osm_fieldwork.data_models
import data_models_path
+
47 from osm_fieldwork.filter_data
import FilterData
+
48 from osm_fieldwork.json2osm
import json2osm
+
49 from osm_fieldwork.OdkCentral
import OdkAppUser
+
50 from osm_fieldwork.xlsforms
import xlsforms_path
+
51 from osm_rawdata.postgres
import PostgresClient
+
52 from shapely
import to_geojson, wkt
+
53 from shapely.geometry
import (
+
+
+
+
57 from shapely.ops
import unary_union
+
58 from sqlalchemy
import and_, column, func, inspect, select, table, text
+
59 from sqlalchemy.dialects.postgresql
import insert
+
60 from sqlalchemy.orm
import Session
+
+
+
+
64 from app.db import db_models
+
+
+
+
+
69 from app.s3 import add_obj_to_bucket, get_obj_from_bucket
+
+
+
+
73 QR_CODES_DIR =
"QR_codes/"
+
74 TASK_GEOJSON_DIR =
"geojson/"
+
75 TILESDIR =
"/opt/tiles"
+
+
+
78 async
def get_projects(
+
+
+
+
+
83 hashtags: List[str] =
None,
+
+
+
86 """Get all projects."""
+
+
+
89 filters.append(db_models.DbProject.author_id == user_id)
+
+
+
92 filters.append(db_models.DbProject.hashtags.op(
"&&")(hashtags))
+
+
+
95 filters.append(db_models.DbProject.project_name_prefix.ilike(f
"%{search}%"))
+
+
+
+
99 db.query(db_models.DbProject)
+
100 .filter(and_(*filters))
+
101 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
106 project_count = db.query(db_models.DbProject).filter(and_(*filters)).count()
+
+
+
+
110 db.query(db_models.DbProject)
+
111 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
116 project_count = db.query(db_models.DbProject).count()
+
117 return project_count, await convert_to_app_projects(db_projects)
+
+
+
120 async
def get_project_summaries(
+
+
+
+
+
125 hashtags: str =
None,
+
+
+
128 """Get project summary details for main page."""
+
129 project_count, db_projects = await get_projects(
+
130 db, user_id, skip, limit, hashtags, search
+
+
132 return project_count, await convert_to_project_summaries(db_projects)
+
+
+
135 async
def get_project(db: Session, project_id: int):
+
136 """Get a single project."""
+
+
138 db.query(db_models.DbProject)
+
139 .filter(db_models.DbProject.id == project_id)
+
+
+
+
+
+
145 async
def get_project_by_id(db: Session, project_id: int):
+
146 """Get a single project by id."""
+
+
148 db.query(db_models.DbProject)
+
149 .filter(db_models.DbProject.id == project_id)
+
+
+
152 return await convert_to_app_project(db_project)
+
+
+
155 async
def get_project_info_by_id(db: Session, project_id: int):
+
156 """Get the project info only by id."""
+
+
158 db.query(db_models.DbProjectInfo)
+
159 .filter(db_models.DbProjectInfo.project_id == project_id)
+
160 .order_by(db_models.DbProjectInfo.project_id)
+
+
+
163 return await convert_to_app_project_info(db_project_info)
+
+
+
166 async
def delete_one_project(db: Session, db_project: db_models.DbProject) ->
None:
+
167 """Delete a project by id."""
+
+
169 project_id = db_project.id
+
170 db.delete(db_project)
+
+
172 log.info(f
"Deleted project with ID: {project_id}")
+
173 except Exception
as e:
+
+
175 raise HTTPException(e)
from e
+
+
+
178 async
def partial_update_project_info(
+
179 db: Session, project_metadata: project_schemas.ProjectUpdate, project_id
+
+
181 """Partial project update for PATCH."""
+
+
183 db_project = await get_project_by_id(db, project_id)
+
+
+
+
+
188 status_code=428, detail=f
"Project with id {project_id} does not exist"
+
+
+
+
192 db_project_info = await get_project_info_by_id(db, project_id)
+
+
+
195 if project_metadata.name:
+
196 db_project.project_name_prefix = project_metadata.name
+
197 db_project_info.name = project_metadata.name
+
198 if project_metadata.description:
+
199 db_project_info.description = project_metadata.description
+
200 if project_metadata.short_description:
+
201 db_project_info.short_description = project_metadata.short_description
+
+
+
204 db.refresh(db_project)
+
+
206 return await convert_to_app_project(db_project)
+
+
+
209 async
def update_project_info(
+
210 db: Session, project_metadata: project_schemas.ProjectUpload, project_id
+
+
212 """Full project update for PUT."""
+
213 user = project_metadata.author
+
214 project_info = project_metadata.project_info
+
+
+
+
218 raise HTTPException(
"No user passed in")
+
+
220 raise HTTPException(
"No project info passed in")
+
+
+
223 db_user = await user_crud.get_user(db, user.id)
+
+
+
226 status_code=400, detail=f
"User {user.username} does not exist"
+
+
+
+
230 db_project = await get_project_by_id(db, project_id)
+
+
+
233 status_code=428, detail=f
"Project with id {project_id} does not exist"
+
+
+
+
237 project_info = project_metadata.project_info
+
+
+
240 db_project.author = db_user
+
241 db_project.project_name_prefix = project_info.name
+
+
+
244 db_project_info = await get_project_info_by_id(db, project_id)
+
+
+
247 db_project_info.name = project_info.name
+
248 db_project_info.short_description = project_info.short_description
+
249 db_project_info.description = project_info.description
+
+
+
252 db.refresh(db_project)
+
+
254 return await convert_to_app_project(db_project)
+
+
+
257 async
def create_project_with_project_info(
+
258 db: Session, project_metadata: project_schemas.ProjectUpload, odk_project_id: int
+
+
260 """Create a new project, including all associated info."""
+
+
+
263 project_user = project_metadata.author
+
264 project_info = project_metadata.project_info
+
265 xform_title = project_metadata.xform_title
+
266 odk_credentials = project_metadata.odk_central
+
267 hashtags = project_metadata.hashtags
+
268 organisation_id = project_metadata.organisation_id
+
269 task_split_type = project_metadata.task_split_type
+
270 task_split_dimension = project_metadata.task_split_dimension
+
271 task_num_buildings = project_metadata.task_num_buildings
+
272 data_extract_type = project_metadata.task_num_buildings
+
+
+
+
276 raise HTTPException(
"User details are missing")
+
+
278 raise HTTPException(
"Project info is missing")
+
279 if not odk_project_id:
+
280 raise HTTPException(
"ODK Central project id is missing")
+
+
+
283 "Creating project in FMTM database with vars: "
+
284 f
"project_user: {project_user} | "
+
285 f
"project_info: {project_info} | "
+
286 f
"xform_title: {xform_title} | "
+
287 f
"hashtags: {hashtags}| "
+
288 f
"organisation_id: {organisation_id}"
+
+
+
+
+
293 url = odk_credentials.odk_central_url
+
294 user = odk_credentials.odk_central_user
+
295 pw = odk_credentials.odk_central_password
+
+
+
298 log.debug(
"ODKCentral connection variables not set in function")
+
299 log.debug(
"Attempting extraction from environment variables")
+
300 url = settings.ODK_CENTRAL_URL
+
301 user = settings.ODK_CENTRAL_USER
+
302 pw = settings.ODK_CENTRAL_PASSWD
+
+
+
+
+
307 db_user = await user_crud.get_user(db, project_user.id)
+
+
+
310 status_code=400, detail=f
"User {project_user.username} does not exist"
+
+
+
+
+
+
316 lambda hashtag: hashtag
if hashtag.startswith(
"#")
else f
"#{hashtag}",
+
+
+
+
+
+
+
+
324 db_project = db_models.DbProject(
+
+
326 odkid=odk_project_id,
+
327 project_name_prefix=project_info.name,
+
328 xform_title=xform_title,
+
+
330 odk_central_user=user,
+
331 odk_central_password=pw,
+
+
333 organisation_id=organisation_id,
+
334 task_split_type=task_split_type,
+
335 task_split_dimension=task_split_dimension,
+
336 task_num_buildings=task_num_buildings,
+
337 data_extract_type=data_extract_type,
+
+
+
+
+
+
+
344 db_project_info = db_models.DbProjectInfo(
+
+
346 name=project_info.name,
+
347 short_description=project_info.short_description,
+
348 description=project_info.description,
+
+
350 db.add(db_project_info)
+
+
+
353 db.refresh(db_project)
+
+
355 return await convert_to_app_project(db_project)
+
+
+
358 async
def upload_xlsform(
+
+
+
+
+
+
364 """Upload a custom XLSForm from the user."""
+
+
+
+
+
+
+
+
+
+
374 ins = insert(forms).values(title=name, xls=xlsform, category=category)
+
375 sql = ins.on_conflict_do_update(
+
376 constraint=
"xlsforms_title_key",
+
377 set_=dict(title=name, xls=xlsform, category=category),
+
+
+
+
+
382 except Exception
as e:
+
+
384 raise HTTPException(status=400, detail={
"message": str(e)})
from e
+
+
+
387 async
def update_multi_polygon_project_boundary(
+
+
+
+
+
392 """Update the boundary for a project & update tasks.
+
+
394 TODO requires refactoring, as it has too large of
+
395 a scope. It should update a project boundary only, then manage
+
396 tasks in another function.
+
+
398 This function receives the project_id and boundary as a parameter
+
399 and creates a task for each polygon in the database.
+
400 This function also creates a project outline from the multiple
+
+
+
+
404 if isinstance(boundary, str):
+
405 boundary = json.loads(boundary)
+
+
+
408 db_project = await get_project_by_id(db, project_id)
+
+
410 log.error(f
"Project {project_id} doesn't exist!")
+
+
+
+
414 if boundary[
"type"] ==
"Feature":
+
415 polygons = [boundary]
+
+
417 polygons = boundary[
"features"]
+
418 log.debug(f
"Processing {len(polygons)} task geometries")
+
419 for polygon
in polygons:
+
+
421 if polygon[
"geometry"][
"type"] ==
"MultiPolygon":
+
422 log.debug(
"Converting MultiPolygon to Polygon")
+
423 polygon[
"geometry"][
"type"] =
"Polygon"
+
424 polygon[
"geometry"][
"coordinates"] = polygon[
"geometry"][
"coordinates"][
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
439 db_task = db_models.DbTask(
+
440 project_id=project_id,
+
441 outline=wkblib.dumps(shape(polygon[
"geometry"]), hex=
True),
+
442 project_task_index=1,
+
+
+
+
+
+
448 db_task.project_task_name = str(db_task.id)
+
+
450 "Created database task | "
+
451 f
"Project ID {project_id} | "
+
452 f
"Task ID {db_task.project_task_name}"
+
+
+
+
+
+
458 f
"""SELECT ST_AsText(ST_ConvexHull(ST_Collect(outline)))
+
+
460 WHERE project_id={project_id};"""
+
+
+
463 log.debug(
"Generating project outline from tasks")
+
464 result = db.execute(query)
+
465 data = result.fetchone()
+
+
467 await update_project_location_info(db_project, data[0])
+
+
+
470 db.refresh(db_project)
+
471 log.debug(
"COMPLETE: creating project boundary, based on task boundaries")
+
+
+
474 except Exception
as e:
+
+
476 raise HTTPException(e)
from e
+
+
+
479 async
def preview_split_by_square(boundary: str, meters: int):
+
480 """Preview split by square for a project boundary.
+
+
482 Use a lambda function to remove the "z" dimension from each
+
483 coordinate in the feature's geometry.
+
+
+
486 def remove_z_dimension(coord):
+
487 """Remove z dimension from geojson."""
+
488 return coord.pop()
if len(coord) == 3
else None
+
+
490 """ Check if the boundary is a Feature or a FeatureCollection """
+
491 if boundary[
"type"] ==
"Feature":
+
492 features = [boundary]
+
493 elif boundary[
"type"] ==
"FeatureCollection":
+
494 features = boundary[
"features"]
+
495 elif boundary[
"type"] ==
"Polygon":
+
+
+
+
+
500 "geometry": boundary,
+
+
+
+
+
505 status_code=400, detail=f
"Invalid GeoJSON type: {boundary['type']}"
+
+
+
+
+
+
511 for feature
in features:
+
512 list(map(remove_z_dimension, feature[
"geometry"][
"coordinates"][0]))
+
513 if feature[
"geometry"][
"type"] ==
"MultiPolygon":
+
514 multi_polygons.append(Polygon(feature[
"geometry"][
"coordinates"][0][0]))
+
+
+
+
518 boundary = multi_polygons[0]
+
519 for geom
in multi_polygons[1:]:
+
520 boundary = boundary.union(geom)
+
+
522 return await run_in_threadpool(
+
523 lambda: split_by_square(
+
+
+
+
+
+
+
530 async
def get_data_extract_from_osm_rawdata(
+
+
+
+
534 """Get data extract using OSM RawData module.
+
+
536 Filters by a specific category.
+
+
+
+
540 aoi_content = await aoi.read()
+
541 boundary = json.loads(aoi_content)
+
+
+
+
+
+
547 config_path = f
"{data_models_path}/{category}.yaml"
+
+
549 if boundary[
"type"] ==
"FeatureCollection":
+
+
+
552 shape(feature[
"geometry"])
for feature
in boundary[
"features"]
+
+
554 updated_geometry = unary_union(geometries)
+
+
556 updated_geometry = shape(boundary[
"geometry"])
+
+
+
559 merged_polygon = updated_geometry.convex_hull
+
+
+
+
+
564 "geometry": to_geojson(merged_polygon),
+
+
+
+
+
569 pg = PostgresClient(
"underpass", config_path)
+
570 data_extract = pg.execQuery(boundary)
+
+
572 except Exception
as e:
+
+
574 raise HTTPException(status_code=400, detail=str(e))
from e
+
+
+
577 async
def get_data_extract_url(
+
+
579 aoi: Union[FeatureCollection, Feature, dict],
+
580 project_id: Optional[int] =
None,
+
+
582 """Request an extract from raw-data-api and extract the file contents.
+
+
584 - The query is posted to raw-data-api and job initiated for fetching the extract.
+
585 - The status of the job is polled every few seconds, until 'SUCCESS' is returned.
+
586 - The resulting flatgeobuf file is streamed in the frontend.
+
+
+
589 str: the URL for the flatgeobuf data extract.
+
+
+
592 db_project = await get_project_by_id(db, project_id)
+
+
594 log.error(f
"Project {project_id} doesn't exist!")
+
+
+
+
598 fgb_url = db_project.data_extract_type
+
+
+
+
+
+
+
+
+
+
+
+
+
+
612 "join_or": {
"building": [],
"highway": [],
"waterway": []}
+
+
+
+
+
+
618 if geom_type := aoi.get(
"type") ==
"FeatureCollection":
+
+
+
621 shape(feature.get(
"geometry"))
for feature
in aoi.get(
"features", [])
+
+
623 merged_geom = unary_union(geometries)
+
624 elif geom_type == aoi.get(
"type") ==
"Feature":
+
625 merged_geom = shape(aoi.get(
"geometry"))
+
+
627 merged_geom = shape(aoi)
+
+
629 query[
"geometry"] = json.loads(to_geojson(merged_geom.convex_hull))
+
+
+
+
633 query[
"fileName"] =
"fmtm-extract"
+
+
635 query[
"outputType"] =
"fgb"
+
+
637 query[
"bind_zip"] =
False
+
+
+
+
641 log.debug(f
"Query for raw data api: {query}")
+
642 base_url = settings.UNDERPASS_API_URL
+
643 query_url = f
"{base_url}/snapshot/"
+
644 headers = {
"accept":
"application/json",
"Content-Type":
"application/json"}
+
+
+
+
648 result = requests.post(query_url, data=json.dumps(query), headers=headers)
+
649 result.raise_for_status()
+
650 except requests.exceptions.HTTPError:
+
651 error_dict = result.json()
+
652 error_dict[
"status_code"] = result.status_code
+
653 log.error(f
"Failed to get extract from raw data api: {error_dict}")
+
+
+
656 task_id = result.json()[
"task_id"]
+
+
+
659 task_url = f
"{base_url}/tasks/status/{task_id}"
+
+
661 result = requests.get(task_url, headers=headers)
+
662 if result.json()[
"status"] ==
"PENDING":
+
+
+
665 elif result.json()[
"status"] ==
"SUCCESS":
+
+
+
668 fgb_url = result.json()[
"result"][
"download_url"]
+
+
+
+
672 async
def split_geojson_into_tasks(
+
+
674 project_geojson: Union[dict, FeatureCollection],
+
675 extract_geojson: Union[dict, FeatureCollection],
+
676 no_of_buildings: int,
+
+
678 """Splits a project into tasks.
+
+
+
681 db (Session): A database session.
+
682 project_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
+
684 extract_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
685 boundary osm data extract (features).
+
686 no_of_buildings (int): The number of buildings to include in each task.
+
+
+
689 Any: A GeoJSON object containing the tasks for the specified project.
+
+
691 log.debug(
"STARTED task splitting using provided boundary and data extract")
+
692 features = await run_in_threadpool(
+
693 lambda: split_by_sql(
+
+
+
696 num_buildings=no_of_buildings,
+
697 osm_extract=extract_geojson,
+
+
+
700 log.debug(
"COMPLETE task splitting")
+
+
+
+
704 async
def update_project_boundary(
+
705 db: Session, project_id: int, boundary: str, meters: int
+
+
707 """Update the boundary for a project and update tasks."""
+
+
709 db_project = await get_project_by_id(db, project_id)
+
+
711 log.error(f
"Project {project_id} doesn't exist!")
+
+
+
+
+
716 def remove_z_dimension(coord):
+
717 """Remove the z dimension from a geojson."""
+
718 return coord.pop()
if len(coord) == 3
else None
+
+
720 """ Check if the boundary is a Feature or a FeatureCollection """
+
721 if boundary[
"type"] ==
"Feature":
+
722 features = [boundary]
+
723 elif boundary[
"type"] ==
"FeatureCollection":
+
724 features = boundary[
"features"]
+
725 elif boundary[
"type"] ==
"Polygon":
+
+
+
+
+
730 "geometry": boundary,
+
+
+
+
+
+
736 status_code=400, detail=f
"Invalid GeoJSON type: {boundary['type']}"
+
+
+
739 """ Apply the lambda function to each coordinate in its geometry """
+
+
741 for feature
in features:
+
742 list(map(remove_z_dimension, feature[
"geometry"][
"coordinates"][0]))
+
743 if feature[
"geometry"][
"type"] ==
"MultiPolygon":
+
744 multi_polygons.append(Polygon(feature[
"geometry"][
"coordinates"][0][0]))
+
+
746 """Update the boundary polygon on the database."""
+
+
748 outline = multi_polygons[0]
+
749 for geom
in multi_polygons[1:]:
+
750 outline = outline.union(geom)
+
+
752 outline = shape(features[0][
"geometry"])
+
+
754 await update_project_location_info(db_project, outline.wkt)
+
+
+
757 db.refresh(db_project)
+
758 log.debug(
"Finished updating project boundary")
+
+
760 log.debug(
"Splitting tasks")
+
761 tasks = split_by_square(
+
+
+
+
765 for poly
in tasks[
"features"]:
+
+
767 task_id = str(poly.get(
"properties", {}).get(
"id")
or poly.get(
"id"))
+
768 db_task = db_models.DbTask(
+
769 project_id=project_id,
+
770 project_task_name=task_id,
+
771 outline=wkblib.dumps(shape(poly[
"geometry"]), hex=
True),
+
+
+
+
775 project_task_index=1,
+
+
+
+
+
+
+
+
+
+
+
786 async
def update_project_with_zip(
+
+
+
789 project_name_prefix: str,
+
790 task_type_prefix: str,
+
791 uploaded_zip: UploadFile,
+
+
793 """Update a project from a zip file.
+
+
795 TODO ensure that logged in user is user who created this project,
+
796 return 403 (forbidden) if not authorized.
+
+
+
799 if uploaded_zip.content_type
not in [
+
+
801 "application/zip-compressed",
+
802 "application/x-zip-compressed",
+
+
+
+
806 detail=f
"File must be a zip. Uploaded file was {uploaded_zip.content_type}",
+
+
+
809 with zipfile.ZipFile(io.BytesIO(uploaded_zip.file.read()),
"r")
as zip:
+
+
811 bad_file = zip.testzip()
+
+
+
814 status_code=400, detail=f
"Zip contained a bad file: {bad_file}"
+
+
+
+
818 listed_files = zip.namelist()
+
+
820 if QR_CODES_DIR
not in listed_files:
+
+
+
823 detail=f
"Zip must contain directory named {QR_CODES_DIR}",
+
+
+
826 if TASK_GEOJSON_DIR
not in listed_files:
+
+
+
829 detail=f
"Zip must contain directory named {TASK_GEOJSON_DIR}",
+
+
+
832 outline_filename = f
"{project_name_prefix}.geojson"
+
833 if outline_filename
not in listed_files:
+
+
+
+
837 f
"Zip must contain file named '{outline_filename}' "
+
838 "that contains a FeatureCollection outlining the project"
+
+
+
+
842 task_outlines_filename = f
"{project_name_prefix}_polygons.geojson"
+
843 if task_outlines_filename
not in listed_files:
+
+
+
+
847 f
"Zip must contain file named '{task_outlines_filename}' "
+
848 "that contains a FeatureCollection where each Feature "
+
+
+
+
+
+
854 db_project = await get_project_by_id(db, project_id)
+
+
+
857 status_code=428, detail=f
"Project with id {project_id} does not exist"
+
+
+
+
861 db_project.project_name_prefix = project_name_prefix
+
862 db_project.task_type_prefix = task_type_prefix
+
+
+
865 outline_shape = await get_outline_from_geojson_file_in_zip(
+
866 zip, outline_filename, f
"Could not generate Shape from {outline_filename}"
+
+
868 await update_project_location_info(db_project, outline_shape.wkt)
+
+
+
871 project_tasks_feature_collection = await get_json_from_zip(
+
+
873 task_outlines_filename,
+
874 f
"Could not generate FeatureCollection from {task_outlines_filename}",
+
+
+
+
+
+
880 db_project.total_tasks = len(project_tasks_feature_collection[
"features"])
+
881 for feature
in project_tasks_feature_collection[
"features"]:
+
882 task_name = feature[
"properties"][
"task"]
+
+
+
+
886 f
"{project_name_prefix}_{task_type_prefix}__{task_name}.png"
+
+
888 db_qr = await get_dbqrcode_from_file(
+
+
890 QR_CODES_DIR + qr_filename,
+
+
892 f
"QRCode for task {task_name} does not exist. "
+
893 f
"File should be in {qr_filename}"
+
+
+
+
+
+
899 task_outline_shape = await get_shape_from_json_str(
+
+
901 f
"Could not create task outline for {task_name} using {feature}",
+
+
+
+
905 task_geojson_filename = (
+
906 f
"{project_name_prefix}_{task_type_prefix}__{task_name}.geojson"
+
+
908 task_geojson = await get_json_from_zip(
+
+
910 TASK_GEOJSON_DIR + task_geojson_filename,
+
911 f
"Geojson for task {task_name} does not exist",
+
+
+
+
+
+
917 task = db_models.DbTask(
+
918 project_id=project_id,
+
919 project_task_index=feature[
"properties"][
"fid"],
+
920 project_task_name=task_name,
+
+
+
923 outline=task_outline_shape.wkt,
+
+
925 initial_feature_count=len(task_geojson[
"features"]),
+
+
+
+
+
930 task_count = task_count + 1
+
931 db_project.last_updated = timestamp()
+
+
+
+
935 db.refresh(db_project)
+
+
+
+
+
+
941 except HTTPException
as e:
+
+
+
+
+
946 except Exception
as e:
+
+
+
+
950 f
"{task_count} tasks were created before the "
+
951 f
"following error was thrown: {e}, on feature: {feature}"
+
+
+
+
+
+
+
+
+
+
961 async
def read_xlsforms(
+
+
+
+
965 """Read the list of XLSForms from the disk."""
+
+
967 package_name =
"osm_fieldwork"
+
968 package_files = pkg_files(package_name)
+
969 for xls
in os.listdir(directory):
+
970 if xls.endswith(
".xls")
or xls.endswith(
".xlsx"):
+
971 file_name = xls.split(
".")[0]
+
972 yaml_file_name = f
"data_models/{file_name}.yaml"
+
973 if package_files.joinpath(yaml_file_name).is_file():
+
+
+
+
+
978 inspect(db_models.DbXForm)
+
+
980 "xlsforms", column(
"title"), column(
"xls"), column(
"xml"), column(
"id")
+
+
+
+
+
985 for xlsform
in xlsforms:
+
986 infile = f
"{directory}/{xlsform}"
+
987 if os.path.getsize(infile) <= 0:
+
988 log.warning(f
"{infile} is empty!")
+
+
990 xls = open(infile,
"rb")
+
991 name = xlsform.split(
".")[0]
+
+
+
+
995 ins = insert(forms).values(title=name, xls=data)
+
996 sql = ins.on_conflict_do_update(
+
997 constraint=
"xlsforms_title_key", set_=dict(title=name, xls=data)
+
+
+
+
+
+
+
+
1005 async
def get_odk_id_for_project(db: Session, project_id: int):
+
1006 """Get the odk project id for the fmtm project id."""
+
+
+
+
+
+
1012 where = f
"id={project_id}"
+
1013 sql = select(project).where(text(where))
+
+
1015 result = db.execute(sql)
+
+
+
1018 if result.rowcount != 1:
+
1019 log.warning(str(sql))
+
+
1021 project_info = result.first()
+
1022 return project_info.odkid
+
+
+
1025 async
def upload_custom_data_extract(
+
+
+
+
+
1030 """Uploads custom data extracts to S3.
+
+
+
1033 db (Session): SQLAlchemy database session.
+
1034 project_id (int): The ID of the project.
+
1035 geojson_str (str): The custom data extracts contents.
+
+
+
1038 str: URL to fgb file in S3.
+
+
1040 project = await get_project(db, project_id)
+
1041 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
+
1044 raise HTTPException(status_code=404, detail=
"Project not found")
+
+
1046 log.debug(
"Parsing geojson file")
+
1047 geojson_parsed = geojson.loads(geojson_str)
+
1048 if isinstance(geojson_parsed, FeatureCollection):
+
1049 log.debug(
"Already in FeatureCollection format, skipping reparse")
+
1050 featcol = geojson_parsed
+
1051 elif isinstance(geojson_parsed, Feature):
+
1052 log.debug(
"Converting Feature to FeatureCollection")
+
1053 featcol = FeatureCollection(geojson_parsed)
+
+
1055 log.debug(
"Converting geometry to FeatureCollection")
+
1056 featcol = FeatureCollection[Feature(geometry=geojson_parsed)]
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
1072 feature_type = featcol.get(
"features", [])[-1].get(
"geometry", {}).get(
"type")
+
1073 if feature_type
not in [
"Polygon",
"Polyline"]:
+
+
1075 "Extract does not contain valid geometry types, from 'Polygon' "
+
+
+
+
1079 raise HTTPException(status_code=404, detail=msg)
+
1080 features_filtered = [
+
+
1082 for feature
in featcol.get(
"features", [])
+
1083 if feature.get(
"geometry", {}).get(
"type",
"") == feature_type
+
+
1085 featcol_filtered = FeatureCollection(features_filtered)
+
+
+
1088 "Generating fgb object from geojson with "
+
1089 f
"{len(featcol_filtered.get('features', []))} features"
+
+
1091 fgb_obj = BytesIO(await geojson_to_flatgeobuf(db, featcol_filtered))
+
1092 s3_fgb_path = f
"/{project.organisation_id}/{project_id}/custom_extract.fgb"
+
+
1094 log.debug(f
"Uploading fgb to S3 path: {s3_fgb_path}")
+
+
1096 settings.S3_BUCKET_NAME,
+
+
+
1099 content_type=
"application/octet-stream",
+
+
+
+
1103 s3_fgb_url = f
"{settings.S3_DOWNLOAD_ROOT}/{settings.S3_BUCKET_NAME}{s3_fgb_path}"
+
1104 log.debug(f
"Commiting extract S3 path to database: {s3_fgb_url}")
+
1105 project.data_extract_type = s3_fgb_url
+
+
+
+
+
+
1111 def flatten_dict(d, parent_key="", sep="_"):
+
1112 """Recursively flattens a nested dictionary into a single-level dictionary.
+
+
+
1115 d (dict): The input dictionary.
+
1116 parent_key (str): The parent key (used for recursion).
+
1117 sep (str): The separator character to use in flattened keys.
+
+
+
1120 dict: The flattened dictionary.
+
+
+
1123 for k, v
in d.items():
+
1124 new_key = f
"{parent_key}{sep}{k}" if parent_key
else k
+
1125 if isinstance(v, dict):
+
1126 items.update(flatten_dict(v, new_key, sep=sep))
+
+
+
+
+
+
+
1133 def generate_task_files(
+
+
+
+
+
+
1139 odk_credentials: project_schemas.ODKCentral,
+
+
1141 """Generate all files for a task."""
+
1142 project_log = log.bind(task=
"create_project", project_id=project_id)
+
+
1144 project_log.info(f
"Generating files for task {task_id}")
+
+
1146 get_project_sync = async_to_sync(get_project)
+
1147 project = get_project_sync(db, project_id)
+
+
1149 odk_id = project.odkid
+
1150 project_name = project.project_name_prefix
+
1151 category = project.xform_title
+
1152 name = f
"{project_name}_{category}_{task_id}"
+
+
+
1155 project_log.info(f
"Creating odkcentral app user for task {task_id}")
+
1156 appuser = central_crud.create_appuser(odk_id, name, odk_credentials)
+
+
+
+
1160 project_log.error(
"Couldn't create appuser for project")
+
+
+
+
1164 project_log.info(f
"Creating qr code for task {task_id}")
+
1165 create_qr_sync = async_to_sync(create_qrcode)
+
1166 qr_code = create_qr_sync(
+
+
+
1169 appuser.json()[
"token"],
+
+
1171 odk_credentials.odk_central_url,
+
+
+
1174 get_task_sync = async_to_sync(tasks_crud.get_task)
+
1175 task = get_task_sync(db, task_id)
+
1176 task.qr_code_id = qr_code[
"qr_code_id"]
+
+
+
+
+
1181 xform = f
"/tmp/{name}.xml"
+
1182 extracts = f
"/tmp/{name}.geojson"
+
+
+
1185 xform_id = f
"{name}".split(
"_")[2]
+
+
+
+
+
+
+
1192 SET task_id={task_id}
+
+
+
+
1196 WHERE project_id={project_id}
+
1197 AND ST_IsValid(geometry)
+
1198 AND ST_IsValid('{task.outline}'::Geometry)
+
1199 AND ST_Contains('{task.outline}'::Geometry, ST_Centroid(geometry))
+
+
+
+
1203 result = db.execute(query)
+
+
+
+
1207 f
"""SELECT jsonb_build_object(
+
1208 'type', 'FeatureCollection',
+
1209 'features', jsonb_agg(feature)
+
+
+
1212 SELECT jsonb_build_object(
+
+
+
1215 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
1216 'properties', properties
+
+
+
1219 WHERE project_id={project_id} and task_id={task_id}
+
+
+
+
1223 result = db.execute(query)
+
+
1225 features = result.fetchone()[0]
+
+
1227 upload_media =
False if features[
"features"]
is None else True
+
+
+
+
1231 with open(extracts,
"w")
as jsonfile:
+
1232 jsonfile.truncate(0)
+
1233 geojson.dump(features, jsonfile)
+
+
+
1236 f
"Generating xform for task: {task_id} "
+
1237 f
"using xform: {xform} | form_type: {form_type}"
+
+
1239 outfile = central_crud.generate_updated_xform(xlsform, xform, form_type)
+
+
+
1242 project_log.info(f
"Uploading media in {task_id}")
+
1243 result = central_crud.create_odk_xform(
+
1244 odk_id, task_id, outfile, odk_credentials,
False, upload_media
+
+
+
+
1248 project_log.info(f
"Updating role for app user in task {task_id}")
+
+
+
+
+
1253 url = odk_credentials.odk_central_url
+
1254 user = odk_credentials.odk_central_user
+
1255 pw = odk_credentials.odk_central_password
+
+
+
1258 log.debug(
"ODKCentral connection variables not set in function")
+
1259 log.debug(
"Attempting extraction from environment variables")
+
1260 url = settings.ODK_CENTRAL_URL
+
1261 user = settings.ODK_CENTRAL_USER
+
1262 pw = settings.ODK_CENTRAL_PASSWD
+
+
1264 odk_app = OdkAppUser(url, user, pw)
+
+
+
1267 projectId=odk_id, xform=xform_id, actorId=appuser.json()[
"id"]
+
+
1269 except Exception
as e:
+
+
+
1272 project.extract_completed_count += 1
+
+
+
+
+
+
+
+
1280 def generate_appuser_files(
+
+
+
1283 extract_polygon: bool,
+
1284 custom_xls_form: str,
+
1285 extracts_contents: str,
+
+
+
1288 background_task_id: Optional[uuid.UUID] =
None,
+
+
1290 """Generate the files for a project.
+
+
1292 QR code, new XForm, and the OSM data extract.
+
+
+
1295 - db: the database session
+
1296 - project_id: Project ID
+
1297 - extract_polygon: boolean to determine if we should extract the polygon
+
1298 - custom_xls_form: the xls file to upload if we have a custom form
+
1299 - extracts_contents: the custom data extract
+
1300 - category: the category of the project
+
1301 - form_type: weather the form is xls, xlsx or xml
+
1302 - background_task_id: the task_id of the background task running this function.
+
+
+
1305 project_log = log.bind(task=
"create_project", project_id=project_id)
+
+
1307 project_log.info(f
"Starting generate_appuser_files for project {project_id}")
+
+
+
+
+
1312 column(
"project_name_prefix"),
+
1313 column(
"xform_title"),
+
+
1315 column(
"odk_central_url"),
+
1316 column(
"odk_central_user"),
+
1317 column(
"odk_central_password"),
+
+
+
+
1321 where = f
"id={project_id}"
+
+
1323 project.c.project_name_prefix,
+
1324 project.c.xform_title,
+
+
1326 project.c.odk_central_url,
+
1327 project.c.odk_central_user,
+
1328 project.c.odk_central_password,
+
1329 geoalchemy2.functions.ST_AsGeoJSON(project.c.outline).label(
"outline"),
+
1330 ).where(text(where))
+
1331 result = db.execute(sql)
+
+
+
1334 if result.rowcount != 1:
+
1335 log.warning(str(sql))
+
1336 if result.rowcount < 1:
+
1337 raise HTTPException(status_code=400, detail=
"Project not found")
+
+
1339 raise HTTPException(status_code=400, detail=
"Multiple projects found")
+
+
1341 one = result.first()
+
+
+
+
+
1346 "odk_central_url": one.odk_central_url,
+
1347 "odk_central_user": one.odk_central_user,
+
1348 "odk_central_password": one.odk_central_password,
+
+
+
1351 odk_credentials = project_schemas.ODKCentral(**odk_credentials)
+
+
1353 xform_title = one.xform_title
if one.xform_title
else None
+
+
1355 category = xform_title
+
+
1357 xlsform = f
"/tmp/{category}.{form_type}"
+
1358 contents = custom_xls_form
+
1359 with open(xlsform,
"wb")
as f:
+
+
+
1362 xlsform = f
"{xlsforms_path}/{xform_title}.xls"
+
+
+
1365 if extracts_contents
is not None:
+
1366 project_log.info(
"Uploading data extracts")
+
1367 upload_extract_sync = async_to_sync(upload_custom_data_extract)
+
1368 upload_extract_sync(db, project_id, extracts_contents)
+
+
+
+
1372 db.query(db_models.DbProject)
+
1373 .filter(db_models.DbProject.id == project_id)
+
+
+
1376 config_file_contents = project.form_config_file
+
+
1378 project_log.info(
"Extracting Data from OSM")
+
+
1380 config_path =
"/tmp/config.yaml"
+
1381 if config_file_contents:
+
1382 with open(config_path,
"w", encoding=
"utf-8")
as config_file_handle:
+
1383 config_file_handle.write(config_file_contents.decode(
"utf-8"))
+
+
1385 config_path = f
"{data_models_path}/{category}.yaml"
+
+
+
1388 pg = PostgresClient(
"underpass", config_path)
+
1389 outline = json.loads(one.outline)
+
1390 boundary = {
"type":
"Feature",
"properties": {},
"geometry": outline}
+
1391 data_extract = pg.execQuery(boundary)
+
1392 filter = FilterData(xlsform)
+
+
1394 updated_data_extract = {
"type":
"FeatureCollection",
"features": []}
+
1395 filtered_data_extract = (
+
1396 filter.cleanData(data_extract)
+
+
1398 else updated_data_extract
+
+
+
+
1402 feature_mappings = []
+
+
1404 for feature
in filtered_data_extract[
"features"]:
+
+
+
1407 feature[
"properties"][
"title"] =
""
+
+
1409 feature_shape = shape(feature[
"geometry"])
+
+
+
+
1413 if extract_polygon
and (
+
1414 not shape(outline).contains(shape(feature_shape.centroid))
+
+
+
+
1418 wkb_element = from_shape(feature_shape, srid=4326)
+
+
1420 "project_id": project_id,
+
1421 "category_title": category,
+
1422 "geometry": wkb_element,
+
1423 "properties": feature[
"properties"],
+
+
1425 updated_data_extract[
"features"].append(feature)
+
1426 feature_mappings.append(feature_mapping)
+
+
1428 db.bulk_insert_mappings(db_models.DbFeatures, feature_mappings)
+
+
+
+
1432 get_task_id_list_sync = async_to_sync(tasks_crud.get_task_id_list)
+
1433 task_list = get_task_id_list_sync(db, project_id)
+
+
+
1436 def wrap_generate_task_files(task):
+
1437 """Func to wrap and return errors from thread.
+
+
1439 Also passes it's own database session for thread safety.
+
1440 If we pass a single db session to multiple threads,
+
1441 there may be inconsistencies or errors.
+
+
+
1444 generate_task_files(
+
+
+
+
+
+
+
+
1452 except Exception
as e:
+
1453 log.exception(str(e))
+
+
+
1456 with ThreadPoolExecutor()
as executor:
+
+
+
1459 executor.submit(wrap_generate_task_files, task)
+
1460 for task
in task_list
+
+
+
+
+
1465 if background_task_id:
+
+
1467 update_bg_task_sync = async_to_sync(
+
1468 update_background_task_status_in_database
+
+
1470 update_bg_task_sync(db, background_task_id, 4)
+
+
1472 except Exception
as e:
+
+
+
1475 if background_task_id:
+
+
1477 update_bg_task_sync = async_to_sync(
+
1478 update_background_task_status_in_database
+
+
1480 update_bg_task_sync(db, background_task_id, 2, str(e))
+
+
+
+
+
+
1486 async
def create_qrcode(
+
+
+
+
+
1491 odk_central_url: str =
None,
+
+
1493 """Create a QR code for a task."""
+
+
1495 log.debug(f
"Generating base64 encoded QR settings for token: {token}")
+
1496 qrcode_data = await central_crud.create_qrcode(
+
1497 odk_id, token, project_name, odk_central_url
+
+
+
1500 log.debug(
"Generating QR code from base64 settings")
+
1501 qrcode = segno.make(qrcode_data, micro=
False)
+
+
1503 log.debug(
"Saving to buffer and decoding")
+
1504 buffer = io.BytesIO()
+
1505 qrcode.save(buffer, kind=
"png", scale=5)
+
1506 qrcode_binary = buffer.getvalue()
+
+
1508 log.debug(f
"Writing QR code to database for token {token}")
+
1509 qrdb = db_models.DbQrCode(image=qrcode_binary)
+
+
+
1512 codes = table(
"qr_code", column(
"id"))
+
1513 sql = select(sqlalchemy.func.count(codes.c.id))
+
1514 result = db.execute(sql)
+
1515 rows = result.fetchone()[0]
+
1516 return {
"data": qrcode,
"id": rows + 1,
"qr_code_id": qrdb.id}
+
+
+
1519 async
def get_project_geometry(db: Session, project_id: int):
+
1520 """Retrieves the geometry of a project.
+
+
+
1523 db (Session): The database session.
+
1524 project_id (int): The ID of the project.
+
+
+
1527 str: A geojson of the project outline.
+
+
1529 projects = table(
"projects", column(
"outline"), column(
"id"))
+
1530 where = f
"projects.id={project_id}"
+
1531 sql = select(geoalchemy2.functions.ST_AsGeoJSON(projects.c.outline)).where(
+
+
+
1534 result = db.execute(sql)
+
+
1536 if result.rowcount != 1:
+
1537 log.warning(str(sql))
+
+
1539 row = eval(result.first()[0])
+
1540 return json.dumps(row)
+
+
+
1543 async
def get_task_geometry(db: Session, project_id: int):
+
1544 """Retrieves the geometry of tasks associated with a project.
+
+
+
1547 db (Session): The database session.
+
1548 project_id (int): The ID of the project.
+
+
+
1551 str: A geojson of the task boundaries
+
+
1553 db_tasks = await tasks_crud.get_tasks(db, project_id,
None)
+
+
1555 for task
in db_tasks:
+
1556 geom = to_shape(task.outline)
+
+
1558 geometry = geom.__geo_interface__
+
+
+
+
1562 feature = {
"type":
"Feature",
"geometry": geometry,
"properties": properties}
+
1563 features.append(feature)
+
+
1565 feature_collection = {
"type":
"FeatureCollection",
"features": features}
+
1566 return json.dumps(feature_collection)
+
+
+
1569 async
def get_project_features_geojson(db: Session, project_id: int):
+
1570 """Get a geojson of all features for a task."""
+
+
1572 db.query(db_models.DbFeatures)
+
1573 .filter(db_models.DbFeatures.project_id == project_id)
+
+
+
+
+
1578 f
"""SELECT jsonb_build_object(
+
1579 'type', 'FeatureCollection',
+
1580 'features', jsonb_agg(feature)
+
+
+
1583 SELECT jsonb_build_object(
+
+
+
1586 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
1587 'properties', properties
+
+
+
1590 WHERE project_id={project_id}
+
+
+
+
+
1595 result = db.execute(query)
+
1596 features = result.fetchone()[0]
+
+
+
1599 task_feature_mapping = {feat.id: feat.task_id
for feat
in db_features}
+
+
1601 for feature
in features[
"features"]:
+
1602 if (feat_id := feature[
"id"])
in task_feature_mapping:
+
1603 feature[
"properties"][
"task_id"] = task_feature_mapping[feat_id]
+
+
+
+
+
1608 async
def get_json_from_zip(zip, filename: str, error_detail: str):
+
1609 """Extract json file from zip."""
+
+
1611 with zip.open(filename)
as file:
+
+
1613 return json.loads(data)
+
1614 except Exception
as e:
+
+
1616 raise HTTPException(
+
1617 status_code=400, detail=f
"{error_detail} ----- Error: {e}"
+
+
+
+
1621 async
def get_outline_from_geojson_file_in_zip(
+
1622 zip, filename: str, error_detail: str, feature_index: int = 0
+
+
1624 """Parse geojson outline within a zip."""
+
+
1626 with zip.open(filename)
as file:
+
+
1628 json_dump = json.loads(data)
+
1629 check_crs(json_dump)
+
1630 feature_collection = FeatureCollection(json_dump)
+
1631 feature = feature_collection[
"features"][feature_index]
+
1632 geom = feature[
"geometry"]
+
1633 shape_from_geom = shape(geom)
+
1634 return shape_from_geom
+
1635 except Exception
as e:
+
+
1637 raise HTTPException(
+
+
1639 detail=f
"{error_detail} ----- Error: {e} ----",
+
+
+
+
1643 async
def get_shape_from_json_str(feature: str, error_detail: str):
+
1644 """Parse geojson outline within a zip to shapely geom."""
+
+
1646 geom = feature[
"geometry"]
+
+
1648 except Exception
as e:
+
+
1650 raise HTTPException(
+
+
1652 detail=f
"{error_detail} ----- Error: {e} ---- Json: {feature}",
+
+
+
+
1656 async
def get_dbqrcode_from_file(zip, qr_filename: str, error_detail: str):
+
1657 """Get qr code from database during import."""
+
+
1659 with zip.open(qr_filename)
as qr_file:
+
1660 binary_qrcode = qr_file.read()
+
+
1662 return db_models.DbQrCode(
+
1663 filename=qr_filename,
+
1664 image=binary_qrcode,
+
+
+
1667 raise HTTPException(
+
1668 status_code=400, detail=f
"{qr_filename} is an empty file"
+
+
1670 except Exception
as e:
+
+
1672 raise HTTPException(
+
1673 status_code=400, detail=f
"{error_detail} ----- Error: {e}"
+
+
+
+
+
+
+
+
+
+
+
1684 async
def convert_to_app_project(db_project: db_models.DbProject):
+
1685 """Legacy function to convert db models --> Pydantic.
+
+
1687 TODO refactor to use Pydantic model methods instead.
+
+
+
1690 log.debug(
"convert_to_app_project called, but no project provided")
+
+
+
1693 log.debug(
"Converting db project to app project")
+
1694 app_project: project_schemas.Project = db_project
+
+
1696 if db_project.outline:
+
1697 log.debug(
"Converting project outline to geojson")
+
1698 app_project.outline_geojson = geometry_to_geojson(
+
1699 db_project.outline, {
"id": db_project.id}, db_project.id
+
+
+
1702 app_project.project_tasks = db_project.tasks
+
+
+
+
+
1707 async
def convert_to_app_project_info(db_project_info: db_models.DbProjectInfo):
+
1708 """Legacy function to convert db models --> Pydantic.
+
+
1710 TODO refactor to use Pydantic model methods instead.
+
+
+
1713 app_project_info: project_schemas.ProjectInfo = db_project_info
+
1714 return app_project_info
+
+
+
+
+
1719 async
def convert_to_app_projects(
+
1720 db_projects: List[db_models.DbProject],
+
1721 ) -> List[project_schemas.ProjectOut]:
+
1722 """Legacy function to convert db models --> Pydantic.
+
+
1724 TODO refactor to use Pydantic model methods instead.
+
+
1726 if db_projects
and len(db_projects) > 0:
+
+
1728 async
def convert_project(project):
+
1729 return await convert_to_app_project(project)
+
+
1731 app_projects = await gather(
+
1732 *[convert_project(project)
for project
in db_projects]
+
+
1734 return [project
for project
in app_projects
if project
is not None]
+
+
+
+
+
1739 async
def convert_to_project_summary(db_project: db_models.DbProject):
+
1740 """Legacy function to convert db models --> Pydantic.
+
+
1742 TODO refactor to use Pydantic model methods instead.
+
+
+
1745 summary: project_schemas.ProjectSummary = db_project
+
+
1747 if db_project.project_info:
+
1748 summary.location_str = db_project.location_str
+
1749 summary.title = db_project.project_info.name
+
1750 summary.description = db_project.project_info.short_description
+
+
1752 summary.num_contributors = (
+
1753 db_project.tasks_mapped + db_project.tasks_validated
+
+
1755 summary.organisation_logo = (
+
1756 db_project.organisation.logo
if db_project.organisation
else None
+
+
+
+
+
+
+
+
1764 async
def convert_to_project_summaries(
+
1765 db_projects: List[db_models.DbProject],
+
1766 ) -> List[project_schemas.ProjectSummary]:
+
1767 """Legacy function to convert db models --> Pydantic.
+
+
1769 TODO refactor to use Pydantic model methods instead.
+
+
1771 if db_projects
and len(db_projects) > 0:
+
+
1773 async
def convert_summary(project):
+
1774 return await convert_to_project_summary(project)
+
+
1776 project_summaries = await gather(
+
1777 *[convert_summary(project)
for project
in db_projects]
+
+
1779 return [summary
for summary
in project_summaries
if summary
is not None]
+
+
+
+
+
1784 async
def convert_to_project_feature(db_project_feature: db_models.DbFeatures):
+
1785 """Legacy function to convert db models --> Pydantic.
+
+
1787 TODO refactor to use Pydantic model methods instead.
+
+
1789 if db_project_feature:
+
1790 app_project_feature: project_schemas.Feature = db_project_feature
+
+
1792 if db_project_feature.geometry:
+
1793 app_project_feature.geometry = geometry_to_geojson(
+
1794 db_project_feature.geometry,
+
1795 db_project_feature.properties,
+
1796 db_project_feature.id,
+
+
+
1799 return app_project_feature
+
+
+
+
+
1804 async
def convert_to_project_features(
+
1805 db_project_features: List[db_models.DbFeatures],
+
1806 ) -> List[project_schemas.Feature]:
+
1807 """Legacy function to convert db models --> Pydantic.
+
+
1809 TODO refactor to use Pydantic model methods instead.
+
+
1811 if db_project_features
and len(db_project_features) > 0:
+
+
1813 async
def convert_feature(project_feature):
+
1814 return await convert_to_project_feature(project_feature)
+
+
1816 app_project_features = await gather(
+
1817 *[convert_feature(feature)
for feature
in db_project_features]
+
+
1819 return [feature
for feature
in app_project_features
if feature
is not None]
+
+
+
+
+
1824 async
def get_project_features(db: Session, project_id: int, task_id: int =
None):
+
1825 """Get features from database for a project."""
+
+
+
1828 db.query(db_models.DbFeatures)
+
1829 .filter(db_models.DbFeatures.project_id == project_id)
+
1830 .filter(db_models.DbFeatures.task_id == task_id)
+
+
+
+
+
1835 db.query(db_models.DbFeatures)
+
1836 .filter(db_models.DbFeatures.project_id == project_id)
+
+
+
1839 return await convert_to_project_features(features)
+
+
+
1842 async
def get_background_task_status(task_id: uuid.UUID, db: Session):
+
1843 """Get the status of a background task."""
+
+
1845 db.query(db_models.BackgroundTasks)
+
1846 .filter(db_models.BackgroundTasks.id == str(task_id))
+
+
+
+
1850 log.warning(f
"No background task with found with UUID: {task_id}")
+
1851 raise HTTPException(status_code=404, detail=
"Task not found")
+
1852 return task.status, task.message
+
+
+
1855 async
def insert_background_task_into_database(
+
1856 db: Session, name: str =
None, project_id: str =
None
+
+
1858 """Inserts a new task into the database.
+
+
+
1861 db (Session): database session
+
1862 name (str): name of the task.
+
1863 project_id (str): associated project id
+
+
+
1866 task_id (uuid.uuid4): The background task uuid for tracking.
+
+
1868 task_id = uuid.uuid4()
+
+
1870 task = db_models.BackgroundTasks(
+
1871 id=str(task_id), name=name, status=1, project_id=project_id
+
+
+
+
+
+
+
+
+
+
1881 async
def update_background_task_status_in_database(
+
1882 db: Session, task_id: uuid.UUID, status: int, message: str =
None
+
+
1884 """Updates the status of a task in the database.
+
+
+
1887 db (Session): database session.
+
1888 task_id (uuid.UUID): uuid of the task.
+
1889 status (int): status of the task.
+
1890 message (str): optional message to add to the db task.
+
+
+
+
+
1895 db.query(db_models.BackgroundTasks).filter(
+
1896 db_models.BackgroundTasks.id == str(task_id)
+
+
+
1899 db_models.BackgroundTasks.status: status,
+
1900 db_models.BackgroundTasks.message: message,
+
+
+
+
+
+
+
+
1908 async
def update_project_form(
+
1909 db: Session, project_id: int, form_type: str, form: UploadFile = File(
None)
+
+
1911 """Upload a new custom XLSForm for a project."""
+
1912 project = await get_project(db, project_id)
+
1913 category = project.xform_title
+
1914 project_title = project.project_name_prefix
+
1915 odk_id = project.odkid
+
+
+
1918 odk_credentials = project_schemas.ODKCentral(
+
1919 odk_central_url=project.odk_central_url,
+
1920 odk_central_user=project.odk_central_user,
+
1921 odk_central_password=project.odk_central_password,
+
+
+
+
1925 xlsform = f
"/tmp/custom_form.{form_type}"
+
1926 contents = await form.read()
+
1927 with open(xlsform,
"wb")
as f:
+
+
+
1930 xlsform = f
"{xlsforms_path}/{category}.xls"
+
+
1932 db.query(db_models.DbFeatures).filter(
+
1933 db_models.DbFeatures.project_id == project_id
+
+
+
+
+
1938 pg = PostgresClient(
"underpass")
+
+
1940 f
"/tmp/{project_title}_{category}.geojson"
+
+
+
1943 extract_polygon =
True if project.data_extract_type ==
"polygon" else False
+
+
1945 project = table(
"projects", column(
"outline"))
+
+
+
+
1949 geoalchemy2.functions.ST_AsGeoJSON(project.c.outline).label(
"outline"),
+
1950 ).where(text(f
"id={project_id}"))
+
1951 result = db.execute(sql)
+
1952 project_outline = result.first()
+
+
1954 final_outline = json.loads(project_outline.outline)
+
+
1956 outline_geojson = pg.getFeatures(
+
1957 boundary=final_outline,
+
+
1959 polygon=extract_polygon,
+
1960 xlsfile=f
"{category}.xls",
+
+
+
+
1964 updated_outline_geojson = {
"type":
"FeatureCollection",
"features": []}
+
+
+
1967 feature_mappings = []
+
+
1969 for feature
in outline_geojson[
"features"]:
+
+
+
1972 feature[
"properties"][
"title"] =
""
+
+
1974 feature_shape = shape(feature[
"geometry"])
+
+
+
+
+
+
+
+
+
+
+
1985 wkb_element = from_shape(feature_shape, srid=4326)
+
+
1987 "project_id": project_id,
+
1988 "category_title": category,
+
1989 "geometry": wkb_element,
+
1990 "properties": feature[
"properties"],
+
+
1992 updated_outline_geojson[
"features"].append(feature)
+
1993 feature_mappings.append(feature_mapping)
+
+
+
1996 db_feature = db_models.DbFeatures(
+
1997 project_id=project_id,
+
1998 category_title=category,
+
1999 geometry=wkb_element,
+
2000 properties=feature[
"properties"],
+
+
+
+
+
2005 tasks_list = await tasks_crud.get_task_id_list(db, project_id)
+
+
2007 for task
in tasks_list:
+
2008 task_obj = await tasks_crud.get_task(db, task)
+
+
+
+
+
+
+
+
+
+
+
2019 WHERE project_id={project_id} and
+
2020 ST_Intersects(geometry, '{task_obj.outline}'::Geometry)
+
+
+
+
2024 result = db.execute(query)
+
+
+
+
2028 f
"""SELECT jsonb_build_object(
+
2029 'type', 'FeatureCollection',
+
2030 'features', jsonb_agg(feature)
+
+
+
2033 SELECT jsonb_build_object(
+
+
+
2036 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
2037 'properties', properties
+
+
+
2040 WHERE project_id={project_id} and task_id={task}
+
+
+
+
2044 result = db.execute(query)
+
2045 features = result.fetchone()[0]
+
+
2047 xform = f
"/tmp/{project_title}_{category}_{task}.xml"
+
+
2049 extracts = f
"/tmp/{project_title}_{category}_{task}.geojson"
+
+
+
+
2053 with open(extracts,
"w")
as jsonfile:
+
2054 jsonfile.truncate(0)
+
2055 geojson.dump(features, jsonfile)
+
+
2057 outfile = central_crud.generate_updated_xform(xlsform, xform, form_type)
+
+
+
2060 result = central_crud.create_odk_xform(
+
2061 odk_id, task, xform, odk_credentials,
True,
True,
False
+
+
+
+
+
+
2067 async
def get_extracted_data_from_db(db: Session, project_id: int, outfile: str):
+
2068 """Get the geojson of those features for this project."""
+
+
2070 f
"""SELECT jsonb_build_object(
+
2071 'type', 'FeatureCollection',
+
2072 'features', jsonb_agg(feature)
+
+
+
2075 SELECT jsonb_build_object(
+
+
+
2078 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
2079 'properties', properties
+
+
+
2082 WHERE project_id={project_id}
+
+
+
+
2086 result = db.execute(query)
+
2087 features = result.fetchone()[0]
+
+
+
+
2091 with open(outfile,
"w")
as jsonfile:
+
2092 jsonfile.truncate(0)
+
2093 geojson.dump(features, jsonfile)
+
+
+
+
2097 def get_project_tiles(
+
+
+
2100 background_task_id: uuid.UUID,
+
+
2102 output_format: str =
"mbtiles",
+
+
+
2105 """Get the tiles for a project.
+
+
+
2108 db (Session): SQLAlchemy db session.
+
2109 project_id (int): ID of project to create tiles for.
+
2110 background_task_id (uuid.UUID): UUID of background task to track.
+
2111 source (str): Tile source ("esri", "bing", "topo", "google", "oam").
+
2112 output_format (str, optional): Default "mbtiles".
+
2113 Other options: "pmtiles", "sqlite3".
+
2114 tms (str, optional): Default None. Custom TMS provider URL.
+
+
+
2117 tiles_path_id = uuid.uuid4()
+
2118 tiles_dir = f
"{TILESDIR}/{tiles_path_id}"
+
2119 outfile = f
"{tiles_dir}/{project_id}_{source}tiles.{output_format}"
+
+
2121 tile_path_instance = db_models.DbTilesPath(
+
2122 project_id=project_id,
+
2123 background_task_id=str(background_task_id),
+
+
+
+
+
+
+
2130 db.add(tile_path_instance)
+
+
+
+
2134 log.debug(f
"Getting bbox for project: {project_id}")
+
+
2136 f
"""SELECT ST_XMin(ST_Envelope(outline)) AS min_lon,
+
2137 ST_YMin(ST_Envelope(outline)) AS min_lat,
+
2138 ST_XMax(ST_Envelope(outline)) AS max_lon,
+
2139 ST_YMax(ST_Envelope(outline)) AS max_lat
+
+
2141 WHERE id = {project_id};"""
+
+
+
2144 result = db.execute(query)
+
2145 project_bbox = result.fetchone()
+
2146 log.debug(f
"Extracted project bbox: {project_bbox}")
+
+
+
2149 min_lon, min_lat, max_lon, max_lat = project_bbox
+
+
2151 log.error(f
"Failed to get bbox from project: {project_id}")
+
+
+
2154 "Creating basemap with params: "
+
2155 f
"boundary={min_lon},{min_lat},{max_lon},{max_lat} | "
+
2156 f
"outfile={outfile} | "
+
+
2158 f
"outdir={tiles_dir} | "
+
2159 f
"source={source} | "
+
+
+
+
2163 create_basemap_file(
+
2164 boundary=f
"{min_lon},{min_lat},{max_lon},{max_lat}",
+
+
+
+
+
+
+
+
2172 log.info(f
"Basemap created for project ID {project_id}: {outfile}")
+
+
2174 tile_path_instance.status = 4
+
+
+
+
2178 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
2179 update_bg_task_sync(db, background_task_id, 4)
+
+
2181 log.info(f
"Tiles generation process completed for project id {project_id}")
+
+
2183 except Exception
as e:
+
2184 log.exception(str(e))
+
2185 log.error(f
"Tiles generation process failed for project id {project_id}")
+
+
2187 tile_path_instance.status = 2
+
+
+
+
2191 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
2192 update_bg_task_sync(db, background_task_id, 2, str(e))
+
+
+
2195 async
def get_mbtiles_list(db: Session, project_id: int):
+
2196 """List mbtiles in database for a project."""
+
+
+
+
2200 db_models.DbTilesPath.id,
+
2201 db_models.DbTilesPath.project_id,
+
2202 db_models.DbTilesPath.status,
+
2203 db_models.DbTilesPath.tile_source,
+
+
2205 .filter(db_models.DbTilesPath.project_id == str(project_id))
+
+
+
+
2209 processed_tiles_list = [
+
+
+
2212 "project_id": x.project_id,
+
2213 "status": x.status.name,
+
2214 "tile_source": x.tile_source,
+
+
+
+
+
2219 return processed_tiles_list
+
+
2221 except Exception
as e:
+
+
2223 raise HTTPException(status_code=400, detail=str(e))
from e
+
+
+
2226 async
def convert_geojson_to_osm(geojson_file: str):
+
2227 """Convert a GeoJSON file to OSM format."""
+
2228 return json2osm(geojson_file)
+
+
+
2231 async
def get_address_from_lat_lon(latitude, longitude):
+
2232 """Get address using Nominatim, using lat,lon."""
+
2233 base_url =
"https://nominatim.openstreetmap.org/reverse"
+
+
+
+
+
+
+
+
2241 headers = {
"Accept-Language":
"en"}
+
+
2243 response = requests.get(base_url, params=params, headers=headers)
+
2244 data = response.json()
+
2245 address = data[
"address"][
"country"]
+
+
2247 if response.status_code == 200:
+
2248 if "city" in data[
"address"]:
+
2249 city = data[
"address"][
"city"]
+
2250 address = f
"{city}" +
"," + address
+
+
+
2253 return "Address not found."
+
+
+
2256 async
def update_project_location_info(
+
2257 db_project: sqlalchemy.orm.declarative_base, project_boundary: str
+
+
2259 """Update project boundary, centroid, address.
+
+
+
2262 db_project(sqlalchemy.orm.declarative_base): The project database record.
+
2263 project_boundary(str): WKT string geometry.
+
+
2265 db_project.outline = project_boundary
+
2266 centroid = (wkt.loads(project_boundary)).centroid.wkt
+
2267 db_project.centroid = centroid
+
2268 geometry = wkt.loads(centroid)
+
2269 longitude, latitude = geometry.x, geometry.y
+
2270 address = await get_address_from_lat_lon(latitude, longitude)
+
2271 db_project.location_str = address
if address
is not None else ""
+
+
+
2274 def check_crs(input_geojson: Union[dict, FeatureCollection]):
+
2275 """Validate CRS is valid for a geojson."""
+
2276 log.debug(
"validating coordinate reference system")
+
+
2278 def is_valid_crs(crs_name):
+
+
2280 "urn:ogc:def:crs:OGC:1.3:CRS84",
+
2281 "urn:ogc:def:crs:EPSG::4326",
+
+
+
2284 return crs_name
in valid_crs_list
+
+
2286 def is_valid_coordinate(coord):
+
+
+
2289 return -180 <= coord[0] <= 180
and -90 <= coord[1] <= 90
+
+
+
2292 "ERROR: Unsupported coordinate system, it is recommended to use a "
+
2293 "GeoJSON file in WGS84(EPSG 4326) standard."
+
+
2295 if "crs" in input_geojson:
+
2296 crs = input_geojson.get(
"crs", {}).get(
"properties", {}).get(
"name")
+
2297 if not is_valid_crs(crs):
+
2298 log.error(error_message)
+
2299 raise HTTPException(status_code=400, detail=error_message)
+
+
+
2302 if input_geojson_type := input_geojson.get(
"type") ==
"FeatureCollection":
+
2303 features = input_geojson.get(
"features", [])
+
+
2305 features[-1].get(
"geometry", {}).get(
"coordinates", [])
if features
else []
+
+
2307 elif input_geojson_type := input_geojson.get(
"type") ==
"Feature":
+
2308 coordinates = input_geojson.get(
"geometry", {}).get(
"coordinates", [])
+
+
+
2311 features[0].get(
"geometry", {}).get(
"type")
+
2312 if input_geojson_type ==
"FeatureCollection" and features
+
2313 else input_geojson.get(
"geometry", {}).get(
"type",
"")
+
+
2315 if geometry_type ==
"MultiPolygon":
+
2316 first_coordinate = coordinates[0][0]
if coordinates
and coordinates[0]
else None
+
2317 elif geometry_type ==
"Point":
+
2318 first_coordinate = coordinates
if coordinates
else None
+
+
2320 elif geometry_type ==
"LineString":
+
2321 first_coordinate = coordinates[0]
if coordinates
else None
+
+
+
2324 first_coordinate = coordinates[0][0]
if coordinates
else None
+
+
2326 if not is_valid_coordinate(first_coordinate):
+
2327 log.error(error_message)
+
2328 raise HTTPException(status_code=400, detail=error_message)
+
+
+
2331 async
def get_tasks_count(db: Session, project_id: int):
+
2332 """Get number of tasks for a project."""
+
+
2334 db.query(db_models.DbProject)
+
2335 .filter(db_models.DbProject.id == project_id)
+
+
+
2338 task_count = len(db_task.tasks)
+
+
+
+
2342 async
def get_pagination(page: int, count: int, results_per_page: int, total: int):
+
2343 """Pagination result for splash page."""
+
2344 total_pages = (count + results_per_page - 1) // results_per_page
+
2345 has_next = (page * results_per_page) < count
+
+
+
2348 pagination = project_schemas.PaginationInfo(
+
+
+
2351 next_num=page + 1
if has_next
else None,
+
+
+
2354 prev_num=page - 1
if has_prev
else None,
+
2355 per_page=results_per_page,
+
+
+
+
+
+
+
2362 async
def get_dashboard_detail(project_id: int, db: Session):
+
2363 """Get project details for project dashboard."""
+
2364 project = await get_project(db, project_id)
+
2365 db_organization = await organization_crud.get_organisation_by_id(
+
2366 db, project.organisation_id
+
+
+
2369 s3_project_path = f
"/{project.organisation_id}/{project_id}"
+
2370 s3_submission_path = f
"/{s3_project_path}/submission.zip"
+
2371 s3_submission_meta_path = f
"/{s3_project_path}/submissions.meta.json"
+
+
+
2374 submission = get_obj_from_bucket(settings.S3_BUCKET_NAME, s3_submission_path)
+
2375 with zipfile.ZipFile(submission,
"r")
as zip_ref:
+
2376 with zip_ref.open(
"submissions.json")
as file_in_zip:
+
2377 content = file_in_zip.read()
+
2378 content = json.loads(content)
+
2379 project.total_submission = len(content)
+
2380 submission_meta = get_obj_from_bucket(
+
2381 settings.S3_BUCKET_NAME, s3_submission_meta_path
+
+
2383 project.last_active = (json.loads(submission_meta.getvalue()))[
+
+
+
+
2387 project.total_submission = 0
+
+
+
+
2391 db.query(db_models.DbTaskHistory.user_id)
+
+
2393 db_models.DbTaskHistory.project_id == project_id,
+
2394 db_models.DbTaskHistory.user_id.isnot(
None),
+
+
+
+
+
+
2400 project.total_tasks = await tasks_crud.get_task_count_in_project(db, project_id)
+
2401 project.organization, project.organization_logo = (
+
2402 db_organization.name,
+
2403 db_organization.logo,
+
+
2405 project.total_contributors = contributors
+
+
+
+
+
2410 async
def get_project_users(db: Session, project_id: int):
+
2411 """Get the users and their contributions for a project.
+
+
+
2414 db (Session): The database session.
+
2415 project_id (int): The ID of the project.
+
+
+
2418 List[Dict[str, Union[str, int]]]: A list of dictionaries containing
+
2419 the username and the number of contributions made by each user
+
2420 for the specified project.
+
+
+
2423 db.query(db_models.DbTaskHistory)
+
2424 .filter(db_models.DbTaskHistory.project_id == project_id)
+
+
+
+
2428 user.user_id
for user
in contributors
if user.user_id
is not None
+
+
+
+
2432 for user_id
in unique_user_ids:
+
2433 contributions = count_user_contributions(db, user_id, project_id)
+
2434 db_user = await user_crud.get_user(db, user_id)
+
2435 response.append({
"user": db_user.username,
"contributions": contributions})
+
+
2437 response = sorted(response, key=
lambda x: x[
"contributions"], reverse=
True)
+
+
+
+
2441 def count_user_contributions(db: Session, user_id: int, project_id: int) -> int:
+
2442 """Count contributions for a specific user.
+
+
+
2445 db (Session): The database session.
+
2446 user_id (int): The ID of the user.
+
2447 project_id (int): The ID of the project.
+
+
+
2450 int: The number of contributions made by the user for the specified
+
+
+
2453 contributions_count = (
+
2454 db.query(func.count(db_models.DbTaskHistory.user_id))
+
+
2456 db_models.DbTaskHistory.user_id == user_id,
+
2457 db_models.DbTaskHistory.project_id == project_id,
+
+
+
+
+
2462 return contributions_count
+
+
+
+
+
+
+
+
+
+
+
First off, We are really glad you're reading this, because we need
+volunteer developers to help improve the Field Mapping Tasking Manager (FMTM)!
+
Thank you¶