+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
18 """Logic for FMTM project routes."""
+
+
+
+
22 from asyncio
import gather
+
23 from io
import BytesIO
+
24 from pathlib
import Path
+
25 from typing
import List, Optional, Union
+
+
+
+
+
30 import shapely.wkb
as wkblib
+
31 from asgiref.sync
import async_to_sync
+
32 from fastapi
import HTTPException, Response
+
33 from fastapi.concurrency
import run_in_threadpool
+
34 from fmtm_splitter.splitter
import split_by_sql, split_by_square
+
35 from geoalchemy2.shape
import to_shape
+
36 from geojson.feature
import Feature, FeatureCollection
+
37 from loguru
import logger
as log
+
38 from osm_fieldwork.basemapper
import create_basemap_file
+
39 from osm_fieldwork.OdkCentral
import OdkAppUser
+
40 from osm_fieldwork.xlsforms
import entities_registration, xlsforms_path
+
41 from osm_rawdata.postgres
import PostgresClient
+
42 from shapely.geometry
import shape
+
43 from sqlalchemy
import and_, column, func, select, table, text
+
44 from sqlalchemy.orm
import Session
+
+
+
+
48 from app.db import db_models
+
+
+
51 flatgeobuf_to_geojson,
+
52 geojson_to_flatgeobuf,
+
+
54 get_featcol_main_geom_type,
+
+
56 parse_and_filter_geojson,
+
57 split_geojson_by_task_areas,
+
58 task_geojson_dict_to_entity_values,
+
+
60 from app.models.enums import HTTPStatus, ProjectRole, ProjectVisibility, XLSFormType
+
+
62 from app.s3 import add_obj_to_bucket
+
+
+
+
66 TILESDIR =
"/opt/tiles"
+
+
+
69 async
def get_projects(
+
+
+
+
73 user_id: Optional[int] =
None,
+
74 hashtags: Optional[List[str]] =
None,
+
75 search: Optional[str] =
None,
+
+
77 """Get all projects, or a filtered subset."""
+
+
+
+
81 filters.append(db_models.DbProject.author_id == user_id)
+
+
+
84 filters.append(db_models.DbProject.hashtags.op(
"&&")(hashtags))
+
+
+
+
88 db_models.DbProject.project_info.has(
+
89 db_models.DbProjectInfo.name.ilike(f
"%{search}%")
+
+
+
+
+
+
95 db.query(db_models.DbProject)
+
96 .filter(and_(*filters))
+
97 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
102 project_count = db.query(db_models.DbProject).filter(and_(*filters)).count()
+
+
+
+
106 db.query(db_models.DbProject)
+
+
108 db_models.DbProject.visibility
+
109 == ProjectVisibility.PUBLIC
+
+
111 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
116 project_count = db.query(db_models.DbProject).count()
+
+
118 filtered_projects = await convert_to_app_projects(db_projects)
+
119 return project_count, filtered_projects
+
+
+
122 async
def get_projects_featcol(
+
+
124 bbox: Optional[str] =
None,
+
125 ) -> geojson.FeatureCollection:
+
126 """Get all projects, or a filtered subset."""
+
+
128 """AND ST_Intersects(
+
129 p.outline, ST_MakeEnvelope(:minx, :miny, :maxx, :maxy, 4326)
+
+
+
+
+
+
+
+
137 minx, miny, maxx, maxy = map(float, bbox.split(
","))
+
138 bbox_params = {
"minx": minx,
"miny": miny,
"maxx": maxx,
"maxy": maxy}
+
+
+
+
142 SELECT jsonb_build_object(
+
143 'type', 'FeatureCollection',
+
144 'features', COALESCE(jsonb_agg(feature), '[]'::jsonb)
+
+
+
147 SELECT jsonb_build_object(
+
+
+
150 'geometry', ST_AsGeoJSON(p.outline)::jsonb,
+
151 'properties', jsonb_build_object(
+
+
+
154 'percentValidated', 0,
+
155 'created', p.created,
+
156 'link', concat('https://', :domain, '/project/', p.id)
+
+
+
+
160 LEFT JOIN project_info pi ON p.id = pi.project_id
+
161 WHERE p.visibility = 'PUBLIC'
+
+
+
+
+
+
167 result = db.execute(query, {
"domain": settings.FMTM_DOMAIN, **bbox_params})
+
168 featcol = result.scalar_one()
+
+
+
+
+
173 async
def get_project_summaries(
+
+
+
+
177 user_id: Optional[int] =
None,
+
178 hashtags: Optional[List[str]] =
None,
+
179 search: Optional[str] =
None,
+
+
181 """Get project summary details for main page."""
+
182 project_count, db_projects = await get_projects(
+
183 db, skip, limit, user_id, hashtags, search
+
+
185 return project_count, await convert_to_project_summaries(db, db_projects)
+
+
+
188 async
def get_project(db: Session, project_id: int):
+
189 """Get a single project."""
+
+
191 db.query(db_models.DbProject)
+
192 .filter(db_models.DbProject.id == project_id)
+
+
+
+
+
197 status_code=HTTPStatus.NOT_FOUND,
+
198 detail=f
"Project with id {project_id} does not exist",
+
+
+
+
+
203 async
def get_project_by_id(db: Session, project_id: int):
+
204 """Get a single project by id."""
+
+
206 db.query(db_models.DbProject)
+
207 .filter(db_models.DbProject.id == project_id)
+
+
+
210 return await convert_to_app_project(db_project)
+
+
+
213 async
def get_project_info_by_id(db: Session, project_id: int):
+
214 """Get the project info only by id."""
+
+
216 db.query(db_models.DbProjectInfo)
+
217 .filter(db_models.DbProjectInfo.project_id == project_id)
+
218 .order_by(db_models.DbProjectInfo.project_id)
+
+
+
221 return await convert_to_app_project_info(db_project_info)
+
+
+
224 async
def delete_one_project(db: Session, db_project: db_models.DbProject) ->
None:
+
225 """Delete a project by id."""
+
+
227 project_id = db_project.id
+
228 db.delete(db_project)
+
+
230 log.info(f
"Deleted project with ID: {project_id}")
+
231 except Exception
as e:
+
+
233 raise HTTPException(e)
from e
+
+
+
236 async
def partial_update_project_info(
+
+
238 project_metadata: project_schemas.ProjectPartialUpdate,
+
239 db_project: db_models.DbProject,
+
+
241 """Partial project update for PATCH."""
+
+
243 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
246 if db_project
and db_project_info:
+
247 if project_metadata.name:
+
248 db_project.project_name_prefix = project_metadata.project_name_prefix
+
249 db_project_info.name = project_metadata.name
+
250 if project_metadata.description:
+
251 db_project_info.description = project_metadata.description
+
252 if project_metadata.short_description:
+
253 db_project_info.short_description = project_metadata.short_description
+
254 if project_metadata.per_task_instructions:
+
255 db_project_info.per_task_instructions = (
+
256 project_metadata.per_task_instructions
+
+
258 if project_metadata.hashtags:
+
259 db_project.hashtags = project_metadata.hashtags
+
+
+
262 db.refresh(db_project)
+
+
264 return await convert_to_app_project(db_project)
+
+
+
267 async
def update_project_info(
+
+
269 project_metadata: project_schemas.ProjectUpdate,
+
270 db_project: db_models.DbProject,
+
271 db_user: db_models.DbUser,
+
+
273 """Full project update for PUT."""
+
274 project_info = project_metadata.project_info
+
+
+
+
278 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
279 detail=
"No project info passed in",
+
+
+
+
283 project_info = project_metadata.project_info
+
+
+
286 db_project.author_id = db_user.id
+
287 db_project.project_name_prefix = project_metadata.project_name_prefix
+
+
+
290 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
293 if db_project
and db_project_info:
+
294 db_project_info.name = project_info.name
+
295 db_project_info.short_description = project_info.short_description
+
296 db_project_info.description = project_info.description
+
+
+
299 db.refresh(db_project)
+
+
301 return await convert_to_app_project(db_project)
+
+
+
304 async
def create_project_with_project_info(
+
+
306 project_metadata: project_schemas.ProjectUpload,
+
+
308 current_user: db_models.DbUser,
+
+
310 """Create a new project, including all associated info."""
+
311 if not odk_project_id:
+
+
313 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
314 detail=
"ODK Central project id is missing",
+
+
+
+
318 "Creating project in FMTM database with vars: "
+
319 f
"project_user: {current_user.username} | "
+
320 f
"project_name: {project_metadata.project_info.name} | "
+
321 f
"xform_category: {project_metadata.xform_category} | "
+
322 f
"hashtags: {project_metadata.hashtags} | "
+
323 f
"organisation_id: {project_metadata.organisation_id}"
+
+
+
+
327 db_project = db_models.DbProject(
+
328 author_id=current_user.id,
+
329 odkid=odk_project_id,
+
330 **project_metadata.model_dump(exclude=[
"project_info",
"outline_geojson"]),
+
+
+
+
+
335 db_project_info = db_models.DbProjectInfo(
+
+
337 name=project_metadata.project_info.name,
+
338 description=project_metadata.project_info.description,
+
339 short_description=project_metadata.project_info.short_description,
+
340 per_task_instructions=project_metadata.project_info.per_task_instructions,
+
+
342 db.add(db_project_info)
+
+
+
345 db.refresh(db_project)
+
+
+
348 generated_project_id = db_project.id
+
349 if db_project.hashtags:
+
350 db_project.hashtags = db_project.hashtags + [
+
351 f
"{settings.FMTM_DOMAIN}-{generated_project_id}"
+
+
+
+
355 return await convert_to_app_project(db_project)
+
+
+
358 async
def create_tasks_from_geojson(
+
+
+
+
+
363 """Create tasks for a project, from provided task boundaries."""
+
+
365 if isinstance(boundaries, str):
+
366 boundaries = json.loads(boundaries)
+
+
+
369 if boundaries[
"type"] ==
"Feature":
+
370 polygons = [boundaries]
+
+
372 polygons = boundaries[
"features"]
+
373 log.debug(f
"Processing {len(polygons)} task geometries")
+
374 for index, polygon
in enumerate(polygons):
+
+
376 if polygon[
"geometry"][
"type"] ==
"MultiPolygon":
+
377 log.debug(
"Converting MultiPolygon to Polygon")
+
378 polygon[
"geometry"][
"type"] =
"Polygon"
+
379 polygon[
"geometry"][
"coordinates"] = polygon[
"geometry"][
"coordinates"][
+
+
+
+
383 db_task = db_models.DbTask(
+
384 project_id=project_id,
+
385 outline=wkblib.dumps(shape(polygon[
"geometry"]), hex=
True),
+
386 project_task_index=index + 1,
+
+
+
+
390 "Created database task | "
+
391 f
"Project ID {project_id} | "
+
392 f
"Task index {index}"
+
+
+
+
+
+
398 log.debug(
"COMPLETE: creating project boundary, based on task boundaries")
+
+
+
401 except Exception
as e:
+
+
403 raise HTTPException(e)
from e
+
+
+
406 async
def preview_split_by_square(boundary: str, meters: int):
+
407 """Preview split by square for a project boundary.
+
+
409 Use a lambda function to remove the "z" dimension from each
+
410 coordinate in the feature's geometry.
+
+
412 boundary = merge_multipolygon(boundary)
+
+
414 return await run_in_threadpool(
+
415 lambda: split_by_square(
+
+
+
+
+
+
+
422 async
def generate_data_extract(
+
423 aoi: Union[FeatureCollection, Feature, dict],
+
424 extract_config: Optional[BytesIO] =
None,
+
+
426 """Request a new data extract in flatgeobuf format.
+
+
+
+
+
431 aoi (Union[FeatureCollection, Feature, dict]):
+
432 Area of interest for data extraction.
+
433 extract_config (Optional[BytesIO], optional):
+
434 Configuration for data extraction. Defaults to None.
+
+
+
+
438 When necessary parameters are missing or data extraction fails.
+
+
+
+
442 URL for the flatgeobuf data extract.
+
+
444 if not extract_config:
+
+
446 status_code=HTTPStatus.BAD_REQUEST,
+
447 detail=
"To generate a new data extract a form_category must be specified.",
+
+
+
+
+
+
453 auth_token=settings.RAW_DATA_API_AUTH_TOKEN
+
454 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
457 fgb_url = pg.execQuery(
+
+
+
+
461 f
"fmtm/{settings.FMTM_DOMAIN}/data_extract"
+
462 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
+
+
467 "useStWithin":
False,
+
468 "fgb_wrap_geoms":
True,
+
+
+
+
+
473 msg =
"Could not get download URL for data extract. Did the API change?"
+
+
+
476 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
+
483 async
def split_geojson_into_tasks(
+
+
485 project_geojson: Union[dict, FeatureCollection],
+
486 no_of_buildings: int,
+
487 extract_geojson: Optional[FeatureCollection] =
None,
+
+
489 """Splits a project into tasks.
+
+
+
492 db (Session): A database session.
+
493 project_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
+
495 extract_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
496 boundary osm data extract (features).
+
497 extract_geojson (FeatureCollection): A GeoJSON of the project
+
498 boundary osm data extract (features).
+
499 If not included, an extract is generated automatically.
+
500 no_of_buildings (int): The number of buildings to include in each task.
+
+
+
503 Any: A GeoJSON object containing the tasks for the specified project.
+
+
505 log.debug(
"STARTED task splitting using provided boundary and data extract")
+
506 features = await run_in_threadpool(
+
507 lambda: split_by_sql(
+
+
+
510 num_buildings=no_of_buildings,
+
511 osm_extract=extract_geojson,
+
+
+
514 log.debug(
"COMPLETE task splitting")
+
+
+
+
+
+
+
+
+
523 async
def read_and_insert_xlsforms(db, directory) -> None:
+
524 """Read the list of XLSForms from the disk and insert to DB."""
+
+
526 for xls_type
in XLSFormType:
+
527 file_name = xls_type.name
+
528 category = xls_type.value
+
529 file_path = Path(directory) / f
"{file_name}.xls"
+
+
531 if not file_path.exists():
+
532 log.warning(f
"{file_path} does not exist!")
+
+
+
535 if file_path.stat().st_size == 0:
+
536 log.warning(f
"{file_path} is empty!")
+
+
+
539 with open(file_path,
"rb")
as xls:
+
+
+
+
+
+
545 INSERT INTO xlsforms (title, xls)
+
546 VALUES (:title, :xls)
+
547 ON CONFLICT (title) DO UPDATE SET
+
548 title = EXCLUDED.title, xls = EXCLUDED.xls
+
+
+
551 db.execute(insert_query, {
"title": category,
"xls": data})
+
+
553 log.info(f
"XLSForm for '{category}' present in the database")
+
+
555 except Exception
as e:
+
+
557 f
"Failed to insert or update {category} in the database. Error: {e}"
+
+
+
560 existing_db_forms = set(
+
561 title
for (title,)
in db.query(db_models.DbXLSForm.title).all()
+
+
563 required_forms = set(xls_type.value
for xls_type
in XLSFormType)
+
+
565 for title
in existing_db_forms - required_forms:
+
+
+
568 DELETE FROM xlsforms WHERE title = :title
+
+
+
571 db.execute(delete_query, {
"title": title})
+
+
+
574 f
"Deleted {title} from the database as it was not present in XLSFormType."
+
+
+
+
578 async
def get_odk_id_for_project(db: Session, project_id: int):
+
579 """Get the odk project id for the fmtm project id."""
+
+
+
+
+
+
585 where = f
"id={project_id}"
+
586 sql = select(project).where(text(where))
+
+
588 result = db.execute(sql)
+
+
+
591 if result.rowcount != 1:
+
592 log.warning(str(sql))
+
+
594 project_info = result.first()
+
595 return project_info.odkid
+
+
+
598 async
def get_or_set_data_extract_url(
+
+
+
+
+
603 """Get or set the data extract URL for a project.
+
+
+
606 db (Session): SQLAlchemy database session.
+
607 project_id (int): The ID of the project.
+
608 url (str): URL to the streamable flatgeobuf data extract.
+
609 If not passed, a new extract is generated.
+
+
+
612 str: URL to fgb file in S3.
+
+
614 db_project = await get_project_by_id(db, project_id)
+
+
616 msg = f
"Project ({project_id}) not found"
+
+
618 raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=msg)
+
+
+
+
+
623 existing_url = db_project.data_extract_url
+
+
+
+
627 f
"No data extract exists for project ({project_id}). "
+
628 "To generate one, call 'projects/generate-data-extract/'"
+
+
+
631 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
+
+
+
636 extract_type =
"polygon"
+
+
638 await update_data_extract_url_in_db(db, db_project, url, extract_type)
+
+
+
+
+
643 async
def update_data_extract_url_in_db(
+
644 db: Session, project: db_models.DbProject, url: str, extract_type: str
+
+
646 """Update the data extract params in the database for a project."""
+
647 log.debug(f
"Setting data extract URL for project ({project.id}): {url}")
+
648 project.data_extract_url = url
+
649 project.data_extract_type = extract_type
+
+
+
+
653 async
def upload_custom_extract_to_s3(
+
+
+
+
657 data_extract_type: str,
+
+
659 """Uploads custom data extracts to S3.
+
+
+
662 db (Session): SQLAlchemy database session.
+
663 project_id (int): The ID of the project.
+
664 fgb_content (bytes): Content of read flatgeobuf file.
+
665 data_extract_type (str): centroid/polygon/line for database.
+
+
+
668 str: URL to fgb file in S3.
+
+
670 project = await get_project(db, project_id)
+
671 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
+
674 raise HTTPException(status_code=404, detail=
"Project not found")
+
+
676 fgb_obj = BytesIO(fgb_content)
+
677 s3_fgb_path = f
"{project.organisation_id}/{project_id}/custom_extract.fgb"
+
+
679 log.debug(f
"Uploading fgb to S3 path: /{s3_fgb_path}")
+
+
681 settings.S3_BUCKET_NAME,
+
+
+
684 content_type=
"application/octet-stream",
+
+
+
+
+
689 f
"{settings.S3_DOWNLOAD_ROOT}/{settings.S3_BUCKET_NAME}/{s3_fgb_path}"
+
+
+
692 await update_data_extract_url_in_db(db, project, s3_fgb_full_url, data_extract_type)
+
+
694 return s3_fgb_full_url
+
+
+
697 async
def upload_custom_fgb_extract(
+
+
+
+
+
702 """Upload a flatgeobuf data extract.
+
+
704 FIXME how can we validate this has the required fields for geojson conversion?
+
+
+
+
+
+
+
+
+
+
714 db (Session): SQLAlchemy database session.
+
715 project_id (int): The ID of the project.
+
716 fgb_content (bytes): Content of read flatgeobuf file.
+
+
+
719 str: URL to fgb file in S3.
+
+
721 featcol = await flatgeobuf_to_geojson(db, fgb_content)
+
+
+
724 msg = f
"Failed extracting geojson from flatgeobuf for project ({project_id})"
+
+
726 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
728 data_extract_type = await get_data_extract_type(featcol)
+
+
730 return await upload_custom_extract_to_s3(
+
+
+
+
+
+
+
+
738 async
def get_data_extract_type(featcol: FeatureCollection) -> str:
+
739 """Determine predominant geometry type for extract."""
+
740 geom_type = get_featcol_main_geom_type(featcol)
+
741 if geom_type
not in [
"Polygon",
"Polyline",
"Point"]:
+
+
743 "Extract does not contain valid geometry types, from 'Polygon' "
+
744 ", 'Polyline' and 'Point'."
+
+
+
747 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
749 "Polygon":
"polygon",
+
+
+
+
753 data_extract_type = geom_name_map.get(geom_type,
"polygon")
+
+
755 return data_extract_type
+
+
+
758 async
def upload_custom_geojson_extract(
+
+
+
761 geojson_raw: Union[str, bytes],
+
+
763 """Upload a geojson data extract.
+
+
+
766 db (Session): SQLAlchemy database session.
+
767 project_id (int): The ID of the project.
+
768 geojson_raw (str): The custom data extracts contents.
+
+
+
771 str: URL to fgb file in S3.
+
+
773 project = await get_project(db, project_id)
+
774 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
776 featcol_filtered = parse_and_filter_geojson(geojson_raw)
+
777 if not featcol_filtered:
+
+
779 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
780 detail=
"Could not process geojson input",
+
+
+
783 await check_crs(featcol_filtered)
+
+
785 data_extract_type = await get_data_extract_type(featcol_filtered)
+
+
+
788 "Generating fgb object from geojson with "
+
789 f
"{len(featcol_filtered.get('features', []))} features"
+
+
791 fgb_data = await geojson_to_flatgeobuf(db, featcol_filtered)
+
+
+
794 msg = f
"Failed converting geojson to flatgeobuf for project ({project_id})"
+
+
796 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
798 return await upload_custom_extract_to_s3(
+
799 db, project_id, fgb_data, data_extract_type
+
+
+
+
803 def flatten_dict(d, parent_key="", sep="_"):
+
804 """Recursively flattens a nested dictionary into a single-level dictionary.
+
+
+
807 d (dict): The input dictionary.
+
808 parent_key (str): The parent key (used for recursion).
+
809 sep (str): The separator character to use in flattened keys.
+
+
+
812 dict: The flattened dictionary.
+
+
+
815 for k, v
in d.items():
+
816 new_key = f
"{parent_key}{sep}{k}" if parent_key
else k
+
817 if isinstance(v, dict):
+
818 items.update(flatten_dict(v, new_key, sep=sep))
+
+
+
+
+
+
824 async
def generate_odk_central_project_content(
+
825 project: db_models.DbProject,
+
826 odk_credentials: project_schemas.ODKCentralDecrypted,
+
+
+
+
+
+
+
833 """Populate the project in ODK Central with XForm, Appuser, Permissions."""
+
834 project_odk_id = project.odkid
+
+
836 appuser_name =
"fmtm_user"
+
+
838 f
"Creating ODK appuser ({appuser_name}) for ODK project ({project_odk_id})"
+
+
840 appuser = OdkAppUser(
+
841 odk_credentials.odk_central_url,
+
842 odk_credentials.odk_central_user,
+
843 odk_credentials.odk_central_password,
+
+
845 appuser_json = appuser.create(project_odk_id, appuser_name)
+
+
+
+
849 msg = f
"Couldn't create appuser {appuser_name} for project"
+
+
+
852 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
854 if not (appuser_token := appuser_json.get(
"token")):
+
855 msg = f
"Couldn't get token for appuser {appuser_name}"
+
+
+
858 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
860 appuser_id = appuser_json.get(
"id")
+
+
+
+
864 with open(entities_registration,
"rb")
as f:
+
865 registration_xlsform = BytesIO(f.read())
+
866 registration_xform = await central_crud.read_and_test_xform(
+
867 registration_xlsform,
"xls", return_form_data=
True
+
+
+
870 log.info(
"Uploading Entity registration XForm to ODK Central")
+
871 central_crud.create_odk_xform(
+
+
+
+
+
+
+
878 xform = await central_crud.read_and_test_xform(
+
879 xlsform, form_file_ext, return_form_data=
True
+
+
+
882 updated_xform = await central_crud.modify_xform_xml(
+
+
+
+
+
+
888 log.info(
"Uploading survey XForm to ODK Central")
+
889 xform_id = central_crud.create_odk_xform(
+
+
+
+
+
+
895 log.info(
"Updating XForm role for appuser in ODK Central")
+
+
897 response = appuser.updateRole(
+
898 projectId=project_odk_id,
+
+
+
+
+
+
904 json_data = response.json()
+
+
906 except json.decoder.JSONDecodeError:
+
+
908 "Could not parse response json during appuser update. "
+
909 f
"status_code={response.status_code}"
+
+
+
912 msg = f
"Failed to update appuser for formId: ({xform_id})"
+
+
+
915 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
+
+
+
920 project_id, odk_form_id, category
+
+
+
923 :project_id, :xform_id, :category
+
+
+
+
+
+
929 {
"project_id": project.id,
"xform_id": xform_id,
"category": form_category},
+
+
+
932 odk_url = odk_credentials.odk_central_url
+
933 return encrypt_value(f
"{odk_url}/v1/key/{appuser_token}/projects/{project_odk_id}")
+
+
+
936 async
def generate_project_files(
+
+
938 project: db_models.DbProject,
+
939 custom_form: Optional[BytesIO],
+
+
941 background_task_id: Optional[uuid.UUID] =
None,
+
+
943 """Generate the files for a project.
+
+
945 QR code (appuser), ODK XForm, ODK Entities from OSM data extract.
+
+
+
948 db (Session): the database session.
+
949 project (DbProject): FMTM database project.
+
950 custom_form (BytesIO): the xls file to upload if we have a custom form
+
951 form_file_ext (str): weather the form is xls, xlsx or xml
+
952 background_task_id (uuid): the task_id of the background task.
+
+
+
955 project_id = project.id
+
956 form_category = project.xform_category
+
957 log.info(f
"Starting generate_project_files for project {project_id}")
+
958 odk_credentials = await project_deps.get_odk_credentials(db, project_id)
+
+
+
961 log.debug(
"User provided custom XLSForm")
+
962 xlsform = custom_form
+
+
964 log.debug(f
"Using default XLSForm for category: '{form_category}'")
+
+
966 form_filename = XLSFormType(form_category).name
+
967 xlsform_path = f
"{xlsforms_path}/{form_filename}.xls"
+
968 with open(xlsform_path,
"rb")
as f:
+
969 xlsform = BytesIO(f.read())
+
+
+
972 log.debug(
"Getting data extract geojson from flatgeobuf")
+
973 feature_collection = await get_project_features_geojson(db, project)
+
+
+
976 log.debug(
"Splitting data extract per task area")
+
977 task_extract_dict = await split_geojson_by_task_areas(
+
978 db, feature_collection, project_id
+
+
+
981 if not task_extract_dict:
+
982 log.warning(f
"Project ({project_id}) failed splitting tasks")
+
+
984 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
985 detail=
"Failed splitting extract by tasks.",
+
+
+
+
989 project_odk_id = project.odkid
+
+
991 encrypted_odk_token = await generate_odk_central_project_content(
+
+
+
+
+
+
997 len(task_extract_dict.keys()),
+
+
+
+
1001 f
"Setting odk token for FMTM project ({project_id}) "
+
1002 f
"ODK project {project_odk_id}"
+
+
1004 project.odk_token = encrypted_odk_token
+
+
1006 for task
in project.tasks:
+
+
1008 task_features = task_extract_dict.get(task.project_task_index, {})
+
1009 feature_count = len(task_features.get(
"features", []))
+
1010 task.feature_count = feature_count
+
+
1012 f
"{feature_count} features added for task {task.project_task_index}"
+
+
+
+
+
+
+
1019 entities_data_dict = await task_geojson_dict_to_entity_values(task_extract_dict)
+
+
+
+
1023 async
with central_deps.get_odk_entity(odk_credentials)
as odk_central:
+
1024 entities = await odk_central.createEntities(
+
+
+
+
+
+
1030 log.debug(f
"Wrote {len(entities)} entities for project ({project_id})")
+
+
1032 log.debug(f
"No entities uploaded for project ({project_id})")
+
+
1034 if background_task_id:
+
+
1036 await update_background_task_status_in_database(
+
1037 db, background_task_id, 4
+
+
+
1040 except Exception
as e:
+
+
+
1043 if background_task_id:
+
+
1045 await update_background_task_status_in_database(
+
1046 db, background_task_id, 2, str(e)
+
+
+
+
+
+
+
1053 async
def get_project_geometry(db: Session, project_id: int):
+
1054 """Retrieves the geometry of a project.
+
+
+
1057 db (Session): The database session.
+
1058 project_id (int): The ID of the project.
+
+
+
1061 str: A geojson of the project outline.
+
+
1063 projects = table(
"projects", column(
"outline"), column(
"id"))
+
1064 where = f
"projects.id={project_id}"
+
1065 sql = select(geoalchemy2.functions.ST_AsGeoJSON(projects.c.outline)).where(
+
+
+
1068 result = db.execute(sql)
+
+
1070 if result.rowcount != 1:
+
1071 log.warning(str(sql))
+
+
1073 row = eval(result.first()[0])
+
1074 return json.dumps(row)
+
+
+
1077 async
def get_task_geometry(db: Session, project_id: int):
+
1078 """Retrieves the geometry of tasks associated with a project.
+
+
+
1081 db (Session): The database session.
+
1082 project_id (int): The ID of the project.
+
+
+
1085 str: A geojson of the task boundaries
+
+
1087 db_tasks = await tasks_crud.get_tasks(db, project_id,
None)
+
+
1089 for task
in db_tasks:
+
1090 geom = to_shape(task.outline)
+
+
1092 geometry = geom.__geo_interface__
+
+
+
+
1096 feature = {
"type":
"Feature",
"geometry": geometry,
"properties": properties}
+
1097 features.append(feature)
+
+
1099 feature_collection = {
"type":
"FeatureCollection",
"features": features}
+
1100 return json.dumps(feature_collection)
+
+
+
1103 async
def get_project_features_geojson(
+
+
1105 project: Union[db_models.DbProject, int],
+
1106 task_id: Optional[int] =
None,
+
1107 ) -> FeatureCollection:
+
1108 """Get a geojson of all features for a task."""
+
1109 if isinstance(project, int):
+
1110 db_project = await get_project(db, project)
+
+
1112 db_project = project
+
1113 project_id = db_project.id
+
+
1115 data_extract_url = db_project.data_extract_url
+
+
1117 if not data_extract_url:
+
1118 raise HTTPException(
+
1119 status_code=HTTPStatus.NOT_FOUND,
+
1120 detail=f
"No data extract exists for project ({project_id})",
+
+
+
+
1124 data_extract_url = data_extract_url.replace(
+
1125 settings.S3_DOWNLOAD_ROOT,
+
1126 settings.S3_ENDPOINT,
+
+
+
1129 with requests.get(data_extract_url)
as response:
+
+
1131 msg = f
"Download failed for data extract, project ({project_id})"
+
+
1133 raise HTTPException(
+
1134 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
1137 log.debug(
"Converting download flatgeobuf to geojson")
+
1138 data_extract_geojson = await flatgeobuf_to_geojson(db, response.content)
+
+
1140 if not data_extract_geojson:
+
1141 msg =
"Failed to convert flatgeobuf --> geojson for " f
"project ({project_id})"
+
+
1143 raise HTTPException(
+
1144 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
1150 split_extract_dict = await split_geojson_by_task_areas(
+
1151 db, data_extract_geojson, project_id
+
+
1153 if not split_extract_dict:
+
1154 raise HTTPException(
+
1155 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
1156 detail=(f
"Failed to extract geojson for task ({task_id})"),
+
+
1158 return split_extract_dict[task_id]
+
+
1160 return data_extract_geojson
+
+
+
1163 async
def get_json_from_zip(zip, filename: str, error_detail: str):
+
1164 """Extract json file from zip."""
+
+
1166 with zip.open(filename)
as file:
+
+
1168 return json.loads(data)
+
1169 except Exception
as e:
+
+
1171 raise HTTPException(
+
1172 status_code=400, detail=f
"{error_detail} ----- Error: {e}"
+
+
+
+
1176 async
def get_outline_from_geojson_file_in_zip(
+
1177 zip, filename: str, error_detail: str, feature_index: int = 0
+
+
1179 """Parse geojson outline within a zip."""
+
+
1181 with zip.open(filename)
as file:
+
+
1183 json_dump = json.loads(data)
+
1184 await check_crs(json_dump)
+
1185 feature_collection = FeatureCollection(json_dump)
+
1186 feature = feature_collection[
"features"][feature_index]
+
1187 geom = feature[
"geometry"]
+
1188 shape_from_geom = shape(geom)
+
1189 return shape_from_geom
+
1190 except Exception
as e:
+
+
1192 raise HTTPException(
+
+
1194 detail=f
"{error_detail} ----- Error: {e} ----",
+
+
+
+
1198 async
def get_shape_from_json_str(feature: str, error_detail: str):
+
1199 """Parse geojson outline within a zip to shapely geom."""
+
+
1201 geom = feature[
"geometry"]
+
+
1203 except Exception
as e:
+
+
1205 raise HTTPException(
+
+
1207 detail=f
"{error_detail} ----- Error: {e} ---- Json: {feature}",
+
+
+
+
+
+
+
+
+
+
+
1218 async
def convert_to_app_project(db_project: db_models.DbProject):
+
1219 """Legacy function to convert db models --> Pydantic.
+
+
1221 TODO refactor to use Pydantic model methods instead.
+
+
+
1224 log.debug(
"convert_to_app_project called, but no project provided")
+
+
+
1227 app_project = db_project
+
+
1229 if db_project.outline:
+
1230 app_project.outline_geojson = geometry_to_geojson(
+
1231 db_project.outline, {
"id": db_project.id}, db_project.id
+
+
+
1234 app_project.tasks = db_project.tasks
+
+
+
+
+
1239 async
def convert_to_app_project_info(db_project_info: db_models.DbProjectInfo):
+
1240 """Legacy function to convert db models --> Pydantic.
+
+
1242 TODO refactor to use Pydantic model methods instead.
+
+
+
1245 app_project_info: project_schemas.ProjectInfo = db_project_info
+
1246 return app_project_info
+
+
+
+
+
1251 async
def convert_to_app_projects(
+
1252 db_projects: List[db_models.DbProject],
+
1253 ) -> List[project_schemas.ProjectOut]:
+
1254 """Legacy function to convert db models --> Pydantic.
+
+
1256 TODO refactor to use Pydantic model methods instead.
+
+
1258 if db_projects
and len(db_projects) > 0:
+
+
1260 async
def convert_project(project):
+
1261 return await convert_to_app_project(project)
+
+
1263 app_projects = await gather(
+
1264 *[convert_project(project)
for project
in db_projects]
+
+
1266 return [project
for project
in app_projects
if project
is not None]
+
+
+
+
+
1271 async
def convert_to_project_summary(db: Session, db_project: db_models.DbProject):
+
1272 """Legacy function to convert db models --> Pydantic.
+
+
1274 TODO refactor to use Pydantic model methods instead.
+
+
+
1277 summary: project_schemas.ProjectSummary = db_project
+
+
1279 if db_project.project_info:
+
1280 summary.title = db_project.project_info.name
+
1281 summary.description = db_project.project_info.short_description
+
1282 summary.num_contributors = (
+
1283 db.query(db_models.DbTaskHistory.user_id)
+
+
1285 db_models.DbTaskHistory.project_id == db_project.id,
+
1286 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
1291 summary.organisation_logo = (
+
1292 db_project.organisation.logo
if db_project.organisation
else None
+
+
+
+
+
+
+
+
1300 async
def convert_to_project_summaries(
+
+
1302 db_projects: List[db_models.DbProject],
+
1303 ) -> List[project_schemas.ProjectSummary]:
+
1304 """Legacy function to convert db models --> Pydantic.
+
+
1306 TODO refactor to use Pydantic model methods instead.
+
+
1308 if db_projects
and len(db_projects) > 0:
+
+
1310 async
def convert_summary(db, project):
+
1311 return await convert_to_project_summary(db, project)
+
+
1313 project_summaries = await gather(
+
1314 *[convert_summary(db, project)
for project
in db_projects]
+
+
1316 return [summary
for summary
in project_summaries
if summary
is not None]
+
+
+
+
+
1321 async
def get_background_task_status(task_id: uuid.UUID, db: Session):
+
1322 """Get the status of a background task."""
+
+
1324 db.query(db_models.BackgroundTasks)
+
1325 .filter(db_models.BackgroundTasks.id == str(task_id))
+
+
+
+
1329 log.warning(f
"No background task with found with UUID: {task_id}")
+
1330 raise HTTPException(status_code=404, detail=
"Task not found")
+
1331 return task.status, task.message
+
+
+
1334 async
def insert_background_task_into_database(
+
1335 db: Session, name: Optional[str] =
None, project_id: Optional[int] =
None
+
+
1337 """Inserts a new task into the database.
+
+
+
1340 db (Session): database session
+
1341 name (int): name of the task.
+
1342 project_id (str): associated project id
+
+
+
1345 task_id (uuid.uuid4): The background task uuid for tracking.
+
+
1347 task_id = uuid.uuid4()
+
+
1349 task = db_models.BackgroundTasks(
+
1350 id=str(task_id), name=name, status=1, project_id=project_id
+
+
+
+
+
+
+
+
+
+
1360 async
def update_background_task_status_in_database(
+
1361 db: Session, task_id: uuid.UUID, status: int, message: str =
None
+
+
1363 """Updates the status of a task in the database.
+
+
+
1366 db (Session): database session.
+
1367 task_id (uuid.UUID): uuid of the task.
+
1368 status (int): status of the task.
+
1369 message (str): optional message to add to the db task.
+
+
+
+
+
1374 db.query(db_models.BackgroundTasks).filter(
+
1375 db_models.BackgroundTasks.id == str(task_id)
+
+
+
1378 db_models.BackgroundTasks.status: status,
+
1379 db_models.BackgroundTasks.message: message,
+
+
+
+
+
+
+
+
1387 async
def get_extracted_data_from_db(db: Session, project_id: int, outfile: str):
+
1388 """Get the geojson of those features for this project."""
+
+
1390 f
"""SELECT jsonb_build_object(
+
1391 'type', 'FeatureCollection',
+
1392 'features', jsonb_agg(feature)
+
+
+
1395 SELECT jsonb_build_object(
+
+
+
1398 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
1399 'properties', properties
+
+
+
1402 WHERE project_id={project_id}
+
+
+
+
1406 result = db.execute(query)
+
1407 features = result.fetchone()[0]
+
+
+
+
1411 with open(outfile,
"w")
as jsonfile:
+
1412 jsonfile.truncate(0)
+
1413 geojson.dump(features, jsonfile)
+
+
+
+
1417 def get_project_tiles(
+
+
+
1420 background_task_id: uuid.UUID,
+
+
1422 output_format: str =
"mbtiles",
+
+
+
1425 """Get the tiles for a project.
+
+
+
1428 db (Session): SQLAlchemy db session.
+
1429 project_id (int): ID of project to create tiles for.
+
1430 background_task_id (uuid.UUID): UUID of background task to track.
+
1431 source (str): Tile source ("esri", "bing", "google", "custom" (tms)).
+
1432 output_format (str, optional): Default "mbtiles".
+
1433 Other options: "pmtiles", "sqlite3".
+
1434 tms (str, optional): Default None. Custom TMS provider URL.
+
+
+
1437 tiles_dir = f
"{TILESDIR}/{project_id}"
+
1438 outfile = f
"{tiles_dir}/{project_id}_{source}tiles.{output_format}"
+
+
1440 tile_path_instance = db_models.DbTilesPath(
+
1441 project_id=project_id,
+
1442 background_task_id=str(background_task_id),
+
+
+
+
+
+
+
1449 db.add(tile_path_instance)
+
+
+
+
1453 log.debug(f
"Getting bbox for project: {project_id}")
+
+
1455 f
"""SELECT ST_XMin(ST_Envelope(outline)) AS min_lon,
+
1456 ST_YMin(ST_Envelope(outline)) AS min_lat,
+
1457 ST_XMax(ST_Envelope(outline)) AS max_lon,
+
1458 ST_YMax(ST_Envelope(outline)) AS max_lat
+
+
1460 WHERE id = {project_id};"""
+
+
+
1463 result = db.execute(query)
+
1464 project_bbox = result.fetchone()
+
1465 log.debug(f
"Extracted project bbox: {project_bbox}")
+
+
+
1468 min_lon, min_lat, max_lon, max_lat = project_bbox
+
+
1470 log.error(f
"Failed to get bbox from project: {project_id}")
+
+
+
1473 "Creating basemap with params: "
+
1474 f
"boundary={min_lon},{min_lat},{max_lon},{max_lat} | "
+
1475 f
"outfile={outfile} | "
+
+
1477 f
"outdir={tiles_dir} | "
+
1478 f
"source={source} | "
+
+
+
+
+
1483 create_basemap_file(
+
1484 boundary=f
"{min_lon},{min_lat},{max_lon},{max_lat}",
+
+
+
+
+
+
+
+
+
1493 log.info(f
"Basemap created for project ID {project_id}: {outfile}")
+
+
1495 tile_path_instance.status = 4
+
+
+
+
1499 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1500 update_bg_task_sync(db, background_task_id, 4)
+
+
1502 log.info(f
"Tiles generation process completed for project id {project_id}")
+
+
1504 except Exception
as e:
+
1505 log.exception(str(e))
+
1506 log.error(f
"Tiles generation process failed for project id {project_id}")
+
+
1508 tile_path_instance.status = 2
+
+
+
+
1512 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1513 update_bg_task_sync(db, background_task_id, 2, str(e))
+
+
+
1516 async
def get_mbtiles_list(db: Session, project_id: int):
+
1517 """List mbtiles in database for a project."""
+
+
+
+
1521 db_models.DbTilesPath.id,
+
1522 db_models.DbTilesPath.project_id,
+
1523 db_models.DbTilesPath.status,
+
1524 db_models.DbTilesPath.path,
+
1525 db_models.DbTilesPath.tile_source,
+
+
1527 .filter(db_models.DbTilesPath.project_id == str(project_id))
+
+
+
+
1531 processed_tiles_list = [
+
+
+
1534 "project_id": x.project_id,
+
1535 "status": x.status.name,
+
1536 "source": x.tile_source,
+
1537 "format": Path(x.path).suffix[1:],
+
+
+
+
+
1542 return processed_tiles_list
+
+
1544 except Exception
as e:
+
+
1546 raise HTTPException(status_code=400, detail=str(e))
from e
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
1568 async
def get_pagination(page: int, count: int, results_per_page: int, total: int):
+
1569 """Pagination result for splash page."""
+
1570 total_pages = (count + results_per_page - 1) // results_per_page
+
1571 has_next = (page * results_per_page) < count
+
+
+
1574 pagination = project_schemas.PaginationInfo(
+
+
+
1577 next_num=page + 1
if has_next
else None,
+
+
+
1580 prev_num=page - 1
if has_prev
else None,
+
1581 per_page=results_per_page,
+
+
+
+
+
+
+
1588 async
def get_dashboard_detail(
+
1589 project: db_models.DbProject, db_organisation: db_models.DbOrganisation, db: Session
+
+
1591 """Get project details for project dashboard."""
+
1592 odk_central = await project_deps.get_odk_credentials(db, project.id)
+
1593 xform = central_crud.get_odk_form(odk_central)
+
1594 db_xform = await project_deps.get_project_xform(db, project.id)
+
+
1596 submission_meta_data = xform.getFullDetails(project.odkid, db_xform.odk_form_id)
+
1597 project.total_submission = submission_meta_data.get(
"submissions", 0)
+
1598 project.last_active = submission_meta_data.get(
"lastSubmission")
+
+
+
1601 db.query(db_models.DbTaskHistory.user_id)
+
+
1603 db_models.DbTaskHistory.project_id == project.id,
+
1604 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
+
1610 project.total_tasks = await tasks_crud.get_task_count_in_project(db, project.id)
+
1611 project.organisation_name, project.organisation_logo = (
+
1612 db_organisation.name,
+
1613 db_organisation.logo,
+
+
1615 project.total_contributors = contributors
+
+
+
+
+
1620 async
def get_project_users(db: Session, project_id: int):
+
1621 """Get the users and their contributions for a project.
+
+
+
1624 db (Session): The database session.
+
1625 project_id (int): The ID of the project.
+
+
+
1628 List[Dict[str, Union[str, int]]]: A list of dictionaries containing
+
1629 the username and the number of contributions made by each user
+
1630 for the specified project.
+
+
+
1633 db.query(db_models.DbTaskHistory)
+
1634 .filter(db_models.DbTaskHistory.project_id == project_id)
+
+
+
+
1638 user.user_id
for user
in contributors
if user.user_id
is not None
+
+
+
+
1642 for user_id
in unique_user_ids:
+
1643 contributions = count_user_contributions(db, user_id, project_id)
+
1644 db_user = await user_crud.get_user(db, user_id)
+
1645 response.append({
"user": db_user.username,
"contributions": contributions})
+
+
1647 response = sorted(response, key=
lambda x: x[
"contributions"], reverse=
True)
+
+
+
+
1651 def count_user_contributions(db: Session, user_id: int, project_id: int) -> int:
+
1652 """Count contributions for a specific user.
+
+
+
1655 db (Session): The database session.
+
1656 user_id (int): The ID of the user.
+
1657 project_id (int): The ID of the project.
+
+
+
1660 int: The number of contributions made by the user for the specified
+
+
+
1663 contributions_count = (
+
1664 db.query(func.count(db_models.DbTaskHistory.user_id))
+
+
1666 db_models.DbTaskHistory.user_id == user_id,
+
1667 db_models.DbTaskHistory.project_id == project_id,
+
+
+
+
+
1672 return contributions_count
+
+
+
1675 async
def add_project_admin(
+
1676 db: Session, user: db_models.DbUser, project: db_models.DbProject
+
+
1678 """Adds a user as an admin to the specified organisation.
+
+
+
1681 db (Session): The database session.
+
1682 user (int): The ID of the user to be added as an admin.
+
1683 project (DbOrganisation): The Project model instance.
+
+
+
1686 Response: The HTTP response with status code 200.
+
+
1688 new_user_role = db_models.DbUserRoles(
+
+
1690 project_id=project.id,
+
1691 role=ProjectRole.PROJECT_MANAGER,
+
+
+
+
1695 project.roles.append(new_user_role)
+
+
+
1698 return Response(status_code=HTTPStatus.OK)
+
+
+
+
+
+
+
+
+
+
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¶