+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
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
+
+
+
+
29 import shapely.wkb
as wkblib
+
30 from asgiref.sync
import async_to_sync
+
31 from fastapi
import HTTPException, Response
+
32 from fastapi.concurrency
import run_in_threadpool
+
33 from fmtm_splitter.splitter
import split_by_sql, split_by_square
+
34 from geoalchemy2.shape
import to_shape
+
35 from geojson.feature
import Feature, FeatureCollection
+
36 from loguru
import logger
as log
+
37 from osm_fieldwork.basemapper
import create_basemap_file
+
38 from osm_fieldwork.xlsforms
import entities_registration, xlsforms_path
+
39 from osm_rawdata.postgres
import PostgresClient
+
40 from shapely.geometry
import shape
+
41 from sqlalchemy
import and_, column, func, select, table, text
+
42 from sqlalchemy.orm
import Session
+
+
+
+
46 from app.db import db_models
+
+
+
49 featcol_keep_dominant_geom_type,
+
50 featcol_to_flatgeobuf,
+
51 flatgeobuf_to_featcol,
+
52 get_featcol_dominant_geom_type,
+
+
54 parse_geojson_file_to_featcol,
+
55 split_geojson_by_task_areas,
+
56 task_geojson_dict_to_entity_values,
+
+
+
59 from app.models.enums import HTTPStatus, ProjectRole, ProjectVisibility, XLSFormType
+
+
61 from app.s3 import add_obj_to_bucket
+
+
+
64 TILESDIR =
"/opt/tiles"
+
+
+
67 async
def get_projects(
+
+
+
+
71 user_id: Optional[int] =
None,
+
72 hashtags: Optional[List[str]] =
None,
+
73 search: Optional[str] =
None,
+
+
75 """Get all projects, or a filtered subset."""
+
+
+
+
79 filters.append(db_models.DbProject.author_id == user_id)
+
+
+
82 filters.append(db_models.DbProject.hashtags.op(
"&&")(hashtags))
+
+
+
+
86 db_models.DbProject.project_info.has(
+
87 db_models.DbProjectInfo.name.ilike(f
"%{search}%")
+
+
+
+
+
+
93 db.query(db_models.DbProject)
+
94 .filter(and_(*filters))
+
95 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
100 project_count = db.query(db_models.DbProject).filter(and_(*filters)).count()
+
+
+
+
104 db.query(db_models.DbProject)
+
+
106 db_models.DbProject.visibility
+
107 == ProjectVisibility.PUBLIC
+
+
109 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
114 project_count = db.query(db_models.DbProject).count()
+
+
116 filtered_projects = await convert_to_app_projects(db_projects)
+
117 return project_count, filtered_projects
+
+
+
120 async
def get_projects_featcol(
+
+
122 bbox: Optional[str] =
None,
+
123 ) -> geojson.FeatureCollection:
+
124 """Get all projects, or a filtered subset."""
+
+
126 """AND ST_Intersects(
+
127 p.outline, ST_MakeEnvelope(:minx, :miny, :maxx, :maxy, 4326)
+
+
+
+
+
+
+
+
135 minx, miny, maxx, maxy = map(float, bbox.split(
","))
+
136 bbox_params = {
"minx": minx,
"miny": miny,
"maxx": maxx,
"maxy": maxy}
+
+
+
+
140 SELECT jsonb_build_object(
+
141 'type', 'FeatureCollection',
+
142 'features', COALESCE(jsonb_agg(feature), '[]'::jsonb)
+
+
+
145 SELECT jsonb_build_object(
+
+
+
148 'geometry', ST_AsGeoJSON(p.outline)::jsonb,
+
149 'properties', jsonb_build_object(
+
+
+
152 'percentValidated', 0,
+
153 'created', p.created,
+
154 'link', concat('https://', :domain, '/project/', p.id)
+
+
+
+
158 LEFT JOIN project_info pi ON p.id = pi.project_id
+
159 WHERE p.visibility = 'PUBLIC'
+
+
+
+
+
+
165 result = db.execute(query, {
"domain": settings.FMTM_DOMAIN, **bbox_params})
+
166 featcol = result.scalar_one()
+
+
+
+
+
171 async
def get_project_summaries(
+
+
+
+
175 user_id: Optional[int] =
None,
+
176 hashtags: Optional[List[str]] =
None,
+
177 search: Optional[str] =
None,
+
+
179 """Get project summary details for main page."""
+
180 project_count, db_projects = await get_projects(
+
181 db, skip, limit, user_id, hashtags, search
+
+
183 return project_count, await convert_to_project_summaries(db, db_projects)
+
+
+
186 async
def get_project(db: Session, project_id: int):
+
187 """Get a single project."""
+
+
189 db.query(db_models.DbProject)
+
190 .filter(db_models.DbProject.id == project_id)
+
+
+
+
+
195 status_code=HTTPStatus.NOT_FOUND,
+
196 detail=f
"Project with id {project_id} does not exist",
+
+
+
+
+
201 async
def get_project_by_id(db: Session, project_id: int):
+
202 """Get a single project by id."""
+
+
204 db.query(db_models.DbProject)
+
205 .filter(db_models.DbProject.id == project_id)
+
+
+
208 return await convert_to_app_project(db_project)
+
+
+
211 async
def get_project_info_by_id(db: Session, project_id: int):
+
212 """Get the project info only by id."""
+
+
214 db.query(db_models.DbProjectInfo)
+
215 .filter(db_models.DbProjectInfo.project_id == project_id)
+
216 .order_by(db_models.DbProjectInfo.project_id)
+
+
+
219 return await convert_to_app_project_info(db_project_info)
+
+
+
222 async
def delete_one_project(db: Session, db_project: db_models.DbProject) ->
None:
+
223 """Delete a project by id."""
+
+
225 project_id = db_project.id
+
226 db.delete(db_project)
+
+
228 log.info(f
"Deleted project with ID: {project_id}")
+
229 except Exception
as e:
+
+
231 raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=e)
from e
+
+
+
234 async
def partial_update_project_info(
+
+
236 project_metadata: project_schemas.ProjectPartialUpdate,
+
237 db_project: db_models.DbProject,
+
+
239 """Partial project update for PATCH."""
+
+
241 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
244 if db_project
and db_project_info:
+
245 if project_metadata.name:
+
246 db_project.project_name_prefix = project_metadata.project_name_prefix
+
247 db_project_info.name = project_metadata.name
+
248 if project_metadata.description:
+
249 db_project_info.description = project_metadata.description
+
250 if project_metadata.short_description:
+
251 db_project_info.short_description = project_metadata.short_description
+
252 if project_metadata.per_task_instructions:
+
253 db_project_info.per_task_instructions = (
+
254 project_metadata.per_task_instructions
+
+
256 if project_metadata.hashtags:
+
257 db_project.hashtags = project_metadata.hashtags
+
+
+
260 db.refresh(db_project)
+
+
262 return await convert_to_app_project(db_project)
+
+
+
265 async
def update_project_with_project_info(
+
+
267 project_metadata: project_schemas.ProjectUpdate,
+
268 db_project: db_models.DbProject,
+
269 db_user: db_models.DbUser,
+
+
271 """Full project update for PUT."""
+
272 project_info = project_metadata.project_info
+
+
+
+
276 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
277 detail=
"No project info passed in",
+
+
+
280 for key, value
in project_metadata.model_dump(
+
281 exclude=[
"project_info",
"outline_geojson"]
+
+
283 setattr(db_project, key, value)
+
+
285 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
288 for key, value
in project_info.model_dump(exclude_unset=
True).items():
+
289 setattr(db_project_info, key, value)
+
+
+
292 db.refresh(db_project)
+
293 db.refresh(db_project_info)
+
+
295 return await convert_to_app_project(db_project)
+
+
+
298 async
def create_project_with_project_info(
+
+
300 project_metadata: project_schemas.ProjectUpload,
+
+
302 current_user: db_models.DbUser,
+
+
304 """Create a new project, including all associated info."""
+
305 if not odk_project_id:
+
+
307 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
308 detail=
"ODK Central project id is missing",
+
+
+
+
312 "Creating project in FMTM database with vars: "
+
313 f
"project_user: {current_user.username} | "
+
314 f
"project_name: {project_metadata.project_info.name} | "
+
315 f
"xform_category: {project_metadata.xform_category} | "
+
316 f
"hashtags: {project_metadata.hashtags} | "
+
317 f
"organisation_id: {project_metadata.organisation_id}"
+
+
+
+
321 db_project = db_models.DbProject(
+
322 author_id=current_user.id,
+
323 odkid=odk_project_id,
+
324 **project_metadata.model_dump(exclude=[
"project_info",
"outline_geojson"]),
+
+
+
+
+
329 db_project_info = db_models.DbProjectInfo(
+
+
331 name=project_metadata.project_info.name,
+
332 description=project_metadata.project_info.description,
+
333 short_description=project_metadata.project_info.short_description,
+
334 per_task_instructions=project_metadata.project_info.per_task_instructions,
+
+
336 db.add(db_project_info)
+
+
+
339 db.refresh(db_project)
+
+
+
342 generated_project_id = db_project.id
+
343 if db_project.hashtags:
+
344 db_project.hashtags = db_project.hashtags + [
+
345 f
"#{settings.FMTM_DOMAIN}-{generated_project_id}"
+
+
+
+
349 return await convert_to_app_project(db_project)
+
+
+
352 async
def create_tasks_from_geojson(
+
+
+
+
+
357 """Create tasks for a project, from provided task boundaries."""
+
+
359 if isinstance(boundaries, str):
+
360 boundaries = json.loads(boundaries)
+
+
+
363 if boundaries[
"type"] ==
"Feature":
+
364 polygons = [boundaries]
+
+
366 polygons = boundaries[
"features"]
+
367 log.debug(f
"Processing {len(polygons)} task geometries")
+
368 for index, polygon
in enumerate(polygons):
+
+
370 if polygon[
"geometry"][
"type"] ==
"MultiPolygon":
+
371 log.debug(
"Converting MultiPolygon to Polygon")
+
372 polygon[
"geometry"][
"type"] =
"Polygon"
+
373 polygon[
"geometry"][
"coordinates"] = polygon[
"geometry"][
"coordinates"][
+
+
+
+
377 db_task = db_models.DbTask(
+
378 project_id=project_id,
+
379 outline=wkblib.dumps(shape(polygon[
"geometry"]), hex=
True),
+
380 project_task_index=index + 1,
+
+
+
+
384 "Created database task | "
+
385 f
"Project ID {project_id} | "
+
386 f
"Task index {index}"
+
+
+
+
+
+
392 log.debug(
"COMPLETE: creating project boundary, based on task boundaries")
+
+
+
395 except Exception
as e:
+
+
397 raise HTTPException(HTTPStatus.UNPROCESSABLE_ENTITY, detail=e)
from e
+
+
+
400 async
def preview_split_by_square(
+
401 boundary: FeatureCollection,
+
+
403 extract_geojson: Optional[FeatureCollection] =
None,
+
+
405 """Preview split by square for a project boundary.
+
+
407 Use a lambda function to remove the "z" dimension from each
+
408 coordinate in the feature's geometry.
+
+
410 boundary = merge_polygons(boundary)
+
+
412 return await run_in_threadpool(
+
413 lambda: split_by_square(
+
+
415 osm_extract=extract_geojson,
+
+
+
+
+
+
421 async
def generate_data_extract(
+
422 aoi: Union[FeatureCollection, Feature, dict],
+
423 extract_config: Optional[BytesIO] =
None,
+
+
425 """Request a new data extract in flatgeobuf format.
+
+
+
+
+
430 aoi (Union[FeatureCollection, Feature, dict]):
+
431 Area of interest for data extraction.
+
432 extract_config (Optional[BytesIO], optional):
+
433 Configuration for data extraction. Defaults to None.
+
+
+
+
437 When necessary parameters are missing or data extraction fails.
+
+
+
+
441 URL for the flatgeobuf data extract.
+
+
443 if not extract_config:
+
+
445 status_code=HTTPStatus.BAD_REQUEST,
+
446 detail=
"To generate a new data extract a form_category must be specified.",
+
+
+
+
+
+
452 auth_token=settings.RAW_DATA_API_AUTH_TOKEN
+
453 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
456 fgb_url = pg.execQuery(
+
+
+
+
460 f
"fmtm/{settings.FMTM_DOMAIN}/data_extract"
+
461 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
+
+
466 "useStWithin":
False,
+
467 "fgb_wrap_geoms":
True,
+
+
+
+
+
472 msg =
"Could not get download URL for data extract. Did the API change?"
+
+
+
475 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
+
482 async
def split_geojson_into_tasks(
+
+
484 project_geojson: FeatureCollection,
+
485 no_of_buildings: int,
+
486 extract_geojson: Optional[FeatureCollection] =
None,
+
+
488 """Splits a project into tasks.
+
+
+
491 db (Session): A database session.
+
492 project_geojson (FeatureCollection): A FeatureCollection containing a
+
493 single project boundary geometry.
+
494 extract_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
495 boundary osm data extract (features).
+
496 extract_geojson (FeatureCollection): A GeoJSON of the project
+
497 boundary osm data extract (features).
+
498 If not included, an extract is generated automatically.
+
499 no_of_buildings (int): The number of buildings to include in each task.
+
+
+
502 Any: A GeoJSON object containing the tasks for the specified project.
+
+
504 log.debug(
"STARTED task splitting using provided boundary and data extract")
+
505 features = await run_in_threadpool(
+
506 lambda: split_by_sql(
+
+
+
509 num_buildings=no_of_buildings,
+
510 osm_extract=extract_geojson,
+
+
+
513 log.debug(
"COMPLETE task splitting")
+
+
+
+
+
+
+
+
+
522 async
def read_and_insert_xlsforms(db, directory) -> None:
+
523 """Read the list of XLSForms from the disk and insert to DB."""
+
+
525 for xls_type
in XLSFormType:
+
526 file_name = xls_type.name
+
527 category = xls_type.value
+
528 file_path = Path(directory) / f
"{file_name}.xls"
+
+
530 if not file_path.exists():
+
531 log.warning(f
"{file_path} does not exist!")
+
+
+
534 if file_path.stat().st_size == 0:
+
535 log.warning(f
"{file_path} is empty!")
+
+
+
538 with open(file_path,
"rb")
as xls:
+
+
+
+
+
+
544 INSERT INTO xlsforms (title, xls)
+
545 VALUES (:title, :xls)
+
546 ON CONFLICT (title) DO UPDATE SET
+
547 title = EXCLUDED.title, xls = EXCLUDED.xls
+
+
+
550 db.execute(insert_query, {
"title": category,
"xls": data})
+
+
552 log.info(f
"XLSForm for '{category}' present in the database")
+
+
554 except Exception
as e:
+
+
556 f
"Failed to insert or update {category} in the database. Error: {e}"
+
+
+
559 existing_db_forms = {
+
560 title
for (title,)
in db.query(db_models.DbXLSForm.title).all()
+
+
562 required_forms = {xls_type.value
for xls_type
in XLSFormType}
+
+
564 for title
in existing_db_forms - required_forms:
+
+
+
567 DELETE FROM xlsforms WHERE title = :title
+
+
+
570 db.execute(delete_query, {
"title": title})
+
+
+
573 f
"Deleted {title} from the database as it was not present in XLSFormType."
+
+
+
+
577 async
def get_odk_id_for_project(db: Session, project_id: int):
+
578 """Get the odk project id for the fmtm project id."""
+
+
+
+
+
+
584 where = f
"id={project_id}"
+
585 sql = select(project).where(text(where))
+
+
587 result = db.execute(sql)
+
+
+
590 if result.rowcount != 1:
+
591 log.warning(str(sql))
+
+
593 project_info = result.first()
+
594 return project_info.odkid
+
+
+
597 async
def get_or_set_data_extract_url(
+
+
+
+
+
602 """Get or set the data extract URL for a project.
+
+
+
605 db (Session): SQLAlchemy database session.
+
606 project_id (int): The ID of the project.
+
607 url (str): URL to the streamable flatgeobuf data extract.
+
608 If not passed, a new extract is generated.
+
+
+
611 str: URL to fgb file in S3.
+
+
613 db_project = await get_project_by_id(db, project_id)
+
+
615 msg = f
"Project ({project_id}) not found"
+
+
617 raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=msg)
+
+
+
+
+
622 existing_url = db_project.data_extract_url
+
+
+
+
626 f
"No data extract exists for project ({project_id}). "
+
627 "To generate one, call 'projects/generate-data-extract/'"
+
+
+
630 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
+
+
+
635 extract_type =
"polygon"
+
+
637 await update_data_extract_url_in_db(db, db_project, url, extract_type)
+
+
+
+
+
642 async
def update_data_extract_url_in_db(
+
643 db: Session, project: db_models.DbProject, url: str, extract_type: str
+
+
645 """Update the data extract params in the database for a project."""
+
646 log.debug(f
"Setting data extract URL for project ({project.id}): {url}")
+
647 project.data_extract_url = url
+
648 project.data_extract_type = extract_type
+
+
+
+
652 async
def upload_custom_extract_to_s3(
+
+
+
+
656 data_extract_type: str,
+
+
658 """Uploads custom data extracts to S3.
+
+
+
661 db (Session): SQLAlchemy database session.
+
662 project_id (int): The ID of the project.
+
663 fgb_content (bytes): Content of read flatgeobuf file.
+
664 data_extract_type (str): centroid/polygon/line for database.
+
+
+
667 str: URL to fgb file in S3.
+
+
669 project = await get_project(db, project_id)
+
670 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
+
673 raise HTTPException(status_code=404, detail=
"Project not found")
+
+
675 fgb_obj = BytesIO(fgb_content)
+
676 s3_fgb_path = f
"{project.organisation_id}/{project_id}/custom_extract.fgb"
+
+
678 log.debug(f
"Uploading fgb to S3 path: /{s3_fgb_path}")
+
+
680 settings.S3_BUCKET_NAME,
+
+
+
683 content_type=
"application/octet-stream",
+
+
+
+
+
688 f
"{settings.S3_DOWNLOAD_ROOT}/{settings.S3_BUCKET_NAME}/{s3_fgb_path}"
+
+
+
691 await update_data_extract_url_in_db(db, project, s3_fgb_full_url, data_extract_type)
+
+
693 return s3_fgb_full_url
+
+
+
696 async
def upload_custom_fgb_extract(
+
+
+
+
+
701 """Upload a flatgeobuf data extract.
+
+
703 FIXME how can we validate this has the required fields for geojson conversion?
+
+
+
+
+
+
+
+
+
+
713 db (Session): SQLAlchemy database session.
+
714 project_id (int): The ID of the project.
+
715 fgb_content (bytes): Content of read flatgeobuf file.
+
+
+
718 str: URL to fgb file in S3.
+
+
720 featcol = await flatgeobuf_to_featcol(db, fgb_content)
+
+
+
723 msg = f
"Failed extracting geojson from flatgeobuf for project ({project_id})"
+
+
725 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
727 data_extract_type = await get_data_extract_type(featcol)
+
+
729 return await upload_custom_extract_to_s3(
+
+
+
+
+
+
+
+
737 async
def get_data_extract_type(featcol: FeatureCollection) -> str:
+
738 """Determine predominant geometry type for extract."""
+
739 geom_type = get_featcol_dominant_geom_type(featcol)
+
740 if geom_type
not in [
"Polygon",
"Polyline",
"Point"]:
+
+
742 "Extract does not contain valid geometry types, from 'Polygon' "
+
743 ", 'Polyline' and 'Point'."
+
+
+
746 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
748 "Polygon":
"polygon",
+
+
+
+
752 data_extract_type = geom_name_map.get(geom_type,
"polygon")
+
+
754 return data_extract_type
+
+
+
757 async
def upload_custom_geojson_extract(
+
+
+
760 geojson_raw: Union[str, bytes],
+
+
762 """Upload a geojson data extract.
+
+
+
765 db (Session): SQLAlchemy database session.
+
766 project_id (int): The ID of the project.
+
767 geojson_raw (str): The custom data extracts contents.
+
+
+
770 str: URL to fgb file in S3.
+
+
772 project = await get_project(db, project_id)
+
773 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
775 featcol = parse_geojson_file_to_featcol(geojson_raw)
+
776 featcol_single_geom_type = featcol_keep_dominant_geom_type(featcol)
+
+
778 if not featcol_single_geom_type:
+
+
780 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
781 detail=
"Could not process geojson input",
+
+
+
784 await check_crs(featcol_single_geom_type)
+
+
786 data_extract_type = await get_data_extract_type(featcol_single_geom_type)
+
+
+
789 "Generating fgb object from geojson with "
+
790 f
"{len(featcol_single_geom_type.get('features', []))} features"
+
+
792 fgb_data = await featcol_to_flatgeobuf(db, featcol_single_geom_type)
+
+
+
795 msg = f
"Failed converting geojson to flatgeobuf for project ({project_id})"
+
+
797 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
799 return await upload_custom_extract_to_s3(
+
800 db, project_id, fgb_data, data_extract_type
+
+
+
+
804 def flatten_dict(d, parent_key="", sep="_"):
+
805 """Recursively flattens a nested dictionary into a single-level dictionary.
+
+
+
808 d (dict): The input dictionary.
+
809 parent_key (str): The parent key (used for recursion).
+
810 sep (str): The separator character to use in flattened keys.
+
+
+
813 dict: The flattened dictionary.
+
+
+
816 for k, v
in d.items():
+
817 new_key = f
"{parent_key}{sep}{k}" if parent_key
else k
+
818 if isinstance(v, dict):
+
819 items.update(flatten_dict(v, new_key, sep=sep))
+
+
+
+
+
+
825 async
def generate_odk_central_project_content(
+
826 project: db_models.DbProject,
+
827 odk_credentials: project_schemas.ODKCentralDecrypted,
+
+
+
+
+
+
+
834 """Populate the project in ODK Central with XForm, Appuser, Permissions."""
+
835 project_odk_id = project.odkid
+
+
+
+
839 with open(entities_registration,
"rb")
as f:
+
840 registration_xlsform = BytesIO(f.read())
+
841 registration_xform = await central_crud.read_and_test_xform(
+
842 registration_xlsform,
"xls", return_form_data=
True
+
+
+
845 log.info(
"Uploading Entity registration XForm to ODK Central")
+
846 central_crud.create_odk_xform(
+
+
+
+
+
+
+
853 xform = await central_crud.read_and_test_xform(
+
854 xlsform, form_file_ext, return_form_data=
True
+
+
+
857 updated_xform = await central_crud.modify_xform_xml(
+
+
+
+
+
+
863 log.info(
"Uploading survey XForm to ODK Central")
+
864 xform_id = central_crud.create_odk_xform(
+
+
+
+
+
+
+
+
+
873 project_id, odk_form_id, category
+
+
+
876 :project_id, :xform_id, :category
+
+
+
+
+
+
882 {
"project_id": project.id,
"xform_id": xform_id,
"category": form_category},
+
+
+
885 return await central_crud.get_appuser_token(
+
886 xform_id, project_odk_id, odk_credentials, db
+
+
+
+
890 async
def generate_project_files(
+
+
892 project: db_models.DbProject,
+
893 custom_form: Optional[BytesIO],
+
+
895 background_task_id: Optional[uuid.UUID] =
None,
+
+
897 """Generate the files for a project.
+
+
899 QR code (appuser), ODK XForm, ODK Entities from OSM data extract.
+
+
+
902 db (Session): the database session.
+
903 project (DbProject): FMTM database project.
+
904 custom_form (BytesIO): the xls file to upload if we have a custom form
+
905 form_file_ext (str): weather the form is xls, xlsx or xml
+
906 background_task_id (uuid): the task_id of the background task.
+
+
+
909 project_id = project.id
+
910 form_category = project.xform_category
+
911 log.info(f
"Starting generate_project_files for project {project_id}")
+
912 odk_credentials = await project_deps.get_odk_credentials(db, project_id)
+
+
+
915 log.debug(
"User provided custom XLSForm")
+
916 xlsform = custom_form
+
+
918 log.debug(f
"Using default XLSForm for category: '{form_category}'")
+
+
920 form_filename = XLSFormType(form_category).name
+
921 xlsform_path = f
"{xlsforms_path}/{form_filename}.xls"
+
922 with open(xlsform_path,
"rb")
as f:
+
923 xlsform = BytesIO(f.read())
+
+
+
926 log.debug(
"Getting data extract geojson from flatgeobuf")
+
927 feature_collection = await get_project_features_geojson(db, project)
+
+
+
930 log.debug(
"Splitting data extract per task area")
+
931 task_extract_dict = await split_geojson_by_task_areas(
+
932 db, feature_collection, project_id
+
+
+
935 if not task_extract_dict:
+
936 log.warning(f
"Project ({project_id}) failed splitting tasks")
+
+
938 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
939 detail=
"Failed splitting extract by tasks.",
+
+
+
+
943 project_odk_id = project.odkid
+
+
945 encrypted_odk_token = await generate_odk_central_project_content(
+
+
+
+
+
+
951 len(task_extract_dict.keys()),
+
+
+
+
955 f
"Setting odk token for FMTM project ({project_id}) "
+
956 f
"ODK project {project_odk_id}"
+
+
958 project.odk_token = encrypted_odk_token
+
+
960 for task
in project.tasks:
+
+
962 task_features = task_extract_dict.get(task.project_task_index, {})
+
963 feature_count = len(task_features.get(
"features", []))
+
964 task.feature_count = feature_count
+
+
966 f
"{feature_count} features added for task {task.project_task_index}"
+
+
+
+
+
+
+
973 entities_data_dict = await task_geojson_dict_to_entity_values(task_extract_dict)
+
+
+
+
977 async
with central_deps.get_odk_entity(odk_credentials)
as odk_central:
+
978 entities = await odk_central.createEntities(
+
+
+
+
+
+
984 log.debug(f
"Wrote {len(entities)} entities for project ({project_id})")
+
+
986 log.debug(f
"No entities uploaded for project ({project_id})")
+
+
988 if background_task_id:
+
+
990 await update_background_task_status_in_database(
+
991 db, background_task_id, 4
+
+
+
994 except Exception
as e:
+
+
+
997 if background_task_id:
+
+
999 await update_background_task_status_in_database(
+
1000 db, background_task_id, 2, str(e)
+
+
+
+
+
+
+
1007 async
def get_task_geometry(db: Session, project_id: int):
+
1008 """Retrieves the geometry of tasks associated with a project.
+
+
+
1011 db (Session): The database session.
+
1012 project_id (int): The ID of the project.
+
+
+
1015 str: A geojson of the task boundaries
+
+
1017 db_tasks = await tasks_crud.get_tasks(db, project_id,
None)
+
+
1019 for task
in db_tasks:
+
1020 geom = to_shape(task.outline)
+
+
1022 geometry = geom.__geo_interface__
+
+
+
+
1026 feature = {
"type":
"Feature",
"geometry": geometry,
"properties": properties}
+
1027 features.append(feature)
+
+
1029 feature_collection = {
"type":
"FeatureCollection",
"features": features}
+
1030 return json.dumps(feature_collection)
+
+
+
1033 async
def get_project_features_geojson(
+
+
1035 db_project: db_models.DbProject,
+
1036 task_id: Optional[int] =
None,
+
1037 ) -> FeatureCollection:
+
1038 """Get a geojson of all features for a task."""
+
1039 project_id = db_project.id
+
+
1041 data_extract_url = db_project.data_extract_url
+
+
1043 if not data_extract_url:
+
1044 raise HTTPException(
+
1045 status_code=HTTPStatus.NOT_FOUND,
+
1046 detail=f
"No data extract exists for project ({project_id})",
+
+
+
+
1050 data_extract_url = data_extract_url.replace(
+
1051 settings.S3_DOWNLOAD_ROOT,
+
1052 settings.S3_ENDPOINT,
+
+
+
1055 with requests.get(data_extract_url)
as response:
+
+
1057 msg = f
"Download failed for data extract, project ({project_id})"
+
+
1059 raise HTTPException(
+
1060 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
1063 log.debug(
"Converting download flatgeobuf to geojson")
+
1064 data_extract_geojson = await flatgeobuf_to_featcol(db, response.content)
+
+
1066 if not data_extract_geojson:
+
1067 msg =
"Failed to convert flatgeobuf --> geojson for " f
"project ({project_id})"
+
+
1069 raise HTTPException(
+
1070 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
1076 split_extract_dict = await split_geojson_by_task_areas(
+
1077 db, data_extract_geojson, project_id
+
+
1079 if not split_extract_dict:
+
1080 raise HTTPException(
+
1081 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
1082 detail=(f
"Failed to extract geojson for task ({task_id})"),
+
+
1084 return split_extract_dict[task_id]
+
+
1086 return data_extract_geojson
+
+
+
1089 async
def get_json_from_zip(zip, filename: str, error_detail: str):
+
1090 """Extract json file from zip."""
+
+
1092 with zip.open(filename)
as file:
+
+
1094 return json.loads(data)
+
1095 except Exception
as e:
+
+
1097 raise HTTPException(
+
1098 status_code=400, detail=f
"{error_detail} ----- Error: {e}"
+
+
+
+
1102 async
def get_shape_from_json_str(feature: str, error_detail: str):
+
1103 """Parse geojson outline within a zip to shapely geom."""
+
+
1105 geom = feature[
"geometry"]
+
+
1107 except Exception
as e:
+
+
1109 raise HTTPException(
+
+
1111 detail=f
"{error_detail} ----- Error: {e} ---- Json: {feature}",
+
+
+
+
+
+
+
+
+
+
+
1122 async
def convert_to_app_project(db_project: db_models.DbProject):
+
1123 """Legacy function to convert db models --> Pydantic.
+
+
1125 TODO refactor to use Pydantic model methods instead.
+
+
+
1128 log.debug(
"convert_to_app_project called, but no project provided")
+
+
+
1131 app_project = db_project
+
+
1133 if db_project.outline:
+
1134 app_project.outline_geojson = wkb_geom_to_feature(
+
1135 db_project.outline, {
"id": db_project.id}, db_project.id
+
+
+
1138 app_project.tasks = db_project.tasks
+
+
+
+
+
1143 async
def convert_to_app_project_info(db_project_info: db_models.DbProjectInfo):
+
1144 """Legacy function to convert db models --> Pydantic.
+
+
1146 TODO refactor to use Pydantic model methods instead.
+
+
+
1149 app_project_info: project_schemas.ProjectInfo = db_project_info
+
1150 return app_project_info
+
+
+
+
+
1155 async
def convert_to_app_projects(
+
1156 db_projects: List[db_models.DbProject],
+
1157 ) -> List[project_schemas.ProjectOut]:
+
1158 """Legacy function to convert db models --> Pydantic.
+
+
1160 TODO refactor to use Pydantic model methods instead.
+
+
1162 if db_projects
and len(db_projects) > 0:
+
+
1164 async
def convert_project(project):
+
1165 return await convert_to_app_project(project)
+
+
1167 app_projects = await gather(
+
1168 *[convert_project(project)
for project
in db_projects]
+
+
1170 return [project
for project
in app_projects
if project
is not None]
+
+
+
+
+
1175 async
def convert_to_project_summary(db: Session, db_project: db_models.DbProject):
+
1176 """Legacy function to convert db models --> Pydantic.
+
+
1178 TODO refactor to use Pydantic model methods instead.
+
+
+
1181 summary: project_schemas.ProjectSummary = db_project
+
+
1183 if db_project.project_info:
+
1184 summary.title = db_project.project_info.name
+
1185 summary.description = db_project.project_info.short_description
+
1186 summary.num_contributors = (
+
1187 db.query(db_models.DbTaskHistory.user_id)
+
+
1189 db_models.DbTaskHistory.project_id == db_project.id,
+
1190 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
1195 summary.organisation_logo = (
+
1196 db_project.organisation.logo
if db_project.organisation
else None
+
+
+
+
+
+
+
+
1204 async
def convert_to_project_summaries(
+
+
1206 db_projects: List[db_models.DbProject],
+
1207 ) -> List[project_schemas.ProjectSummary]:
+
1208 """Legacy function to convert db models --> Pydantic.
+
+
1210 TODO refactor to use Pydantic model methods instead.
+
+
1212 if db_projects
and len(db_projects) > 0:
+
+
1214 async
def convert_summary(db, project):
+
1215 return await convert_to_project_summary(db, project)
+
+
1217 project_summaries = await gather(
+
1218 *[convert_summary(db, project)
for project
in db_projects]
+
+
1220 return [summary
for summary
in project_summaries
if summary
is not None]
+
+
+
+
+
1225 async
def get_background_task_status(task_id: uuid.UUID, db: Session):
+
1226 """Get the status of a background task."""
+
+
1228 db.query(db_models.BackgroundTasks)
+
1229 .filter(db_models.BackgroundTasks.id == str(task_id))
+
+
+
+
1233 log.warning(f
"No background task with found with UUID: {task_id}")
+
1234 raise HTTPException(status_code=404, detail=
"Task not found")
+
1235 return task.status, task.message
+
+
+
1238 async
def insert_background_task_into_database(
+
1239 db: Session, name: Optional[str] =
None, project_id: Optional[int] =
None
+
+
1241 """Inserts a new task into the database.
+
+
+
1244 db (Session): database session
+
1245 name (int): name of the task.
+
1246 project_id (str): associated project id
+
+
+
1249 task_id (uuid.uuid4): The background task uuid for tracking.
+
+
1251 task_id = uuid.uuid4()
+
+
1253 task = db_models.BackgroundTasks(
+
1254 id=str(task_id), name=name, status=1, project_id=project_id
+
+
+
+
+
+
+
+
+
+
1264 async
def update_background_task_status_in_database(
+
1265 db: Session, task_id: uuid.UUID, status: int, message: str =
None
+
+
1267 """Updates the status of a task in the database.
+
+
+
1270 db (Session): database session.
+
1271 task_id (uuid.UUID): uuid of the task.
+
1272 status (int): status of the task.
+
1273 message (str): optional message to add to the db task.
+
+
+
+
+
1278 db.query(db_models.BackgroundTasks).filter(
+
1279 db_models.BackgroundTasks.id == str(task_id)
+
+
+
1282 db_models.BackgroundTasks.status: status,
+
1283 db_models.BackgroundTasks.message: message,
+
+
+
+
+
+
+
+
1291 async
def get_extracted_data_from_db(db: Session, project_id: int, outfile: str):
+
1292 """Get the geojson of those features for this project."""
+
+
1294 f
"""SELECT jsonb_build_object(
+
1295 'type', 'FeatureCollection',
+
1296 'features', jsonb_agg(feature)
+
+
+
1299 SELECT jsonb_build_object(
+
+
+
1302 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
1303 'properties', properties
+
+
+
1306 WHERE project_id={project_id}
+
+
+
+
1310 result = db.execute(query)
+
1311 features = result.fetchone()[0]
+
+
+
+
1315 with open(outfile,
"w")
as jsonfile:
+
1316 jsonfile.truncate(0)
+
1317 geojson.dump(features, jsonfile)
+
+
+
+
1321 def get_project_tiles(
+
+
+
1324 background_task_id: uuid.UUID,
+
+
1326 output_format: str =
"mbtiles",
+
1327 tms: Optional[str] =
None,
+
+
1329 """Get the tiles for a project.
+
+
+
1332 db (Session): SQLAlchemy db session.
+
1333 project_id (int): ID of project to create tiles for.
+
1334 background_task_id (uuid.UUID): UUID of background task to track.
+
1335 source (str): Tile source ("esri", "bing", "google", "custom" (tms)).
+
1336 output_format (str, optional): Default "mbtiles".
+
1337 Other options: "pmtiles", "sqlite3".
+
1338 tms (str, optional): Default None. Custom TMS provider URL.
+
+
+
+
+
+
+
1345 zooms =
"12-22" if tms
else "12-19"
+
1346 tiles_dir = f
"{TILESDIR}/{project_id}"
+
1347 outfile = f
"{tiles_dir}/{project_id}_{source}tiles.{output_format}"
+
+
1349 tile_path_instance = db_models.DbTilesPath(
+
1350 project_id=project_id,
+
1351 background_task_id=str(background_task_id),
+
+
+
+
+
+
+
1358 db.add(tile_path_instance)
+
+
+
+
1362 log.debug(f
"Getting bbox for project: {project_id}")
+
+
1364 f
"""SELECT ST_XMin(ST_Envelope(outline)) AS min_lon,
+
1365 ST_YMin(ST_Envelope(outline)) AS min_lat,
+
1366 ST_XMax(ST_Envelope(outline)) AS max_lon,
+
1367 ST_YMax(ST_Envelope(outline)) AS max_lat
+
+
1369 WHERE id = {project_id};"""
+
+
+
1372 result = db.execute(query)
+
1373 project_bbox = result.fetchone()
+
1374 log.debug(f
"Extracted project bbox: {project_bbox}")
+
+
+
1377 min_lon, min_lat, max_lon, max_lat = project_bbox
+
+
1379 msg = f
"Failed to get bbox from project: {project_id}"
+
+
1381 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
+
1384 "Creating basemap with params: "
+
1385 f
"boundary={min_lon},{min_lat},{max_lon},{max_lat} | "
+
1386 f
"outfile={outfile} | "
+
+
1388 f
"outdir={tiles_dir} | "
+
1389 f
"source={source} | "
+
1390 f
"xy={True if tms else False} | "
+
+
+
+
1394 create_basemap_file(
+
1395 boundary=f
"{min_lon},{min_lat},{max_lon},{max_lat}",
+
+
+
+
+
1400 xy=
True if tms
else False,
+
+
+
+
1404 log.info(f
"Basemap created for project ID {project_id}: {outfile}")
+
+
1406 tile_path_instance.status = 4
+
+
+
+
1410 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1411 update_bg_task_sync(db, background_task_id, 4)
+
+
1413 log.info(f
"Tiles generation process completed for project id {project_id}")
+
+
1415 except Exception
as e:
+
1416 log.exception(str(e))
+
1417 log.error(f
"Tiles generation process failed for project id {project_id}")
+
+
1419 tile_path_instance.status = 2
+
+
+
+
1423 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1424 update_bg_task_sync(db, background_task_id, 2, str(e))
+
+
+
1427 async
def get_mbtiles_list(db: Session, project_id: int):
+
1428 """List mbtiles in database for a project."""
+
+
+
+
1432 db_models.DbTilesPath.id,
+
1433 db_models.DbTilesPath.project_id,
+
1434 db_models.DbTilesPath.status,
+
1435 db_models.DbTilesPath.path,
+
1436 db_models.DbTilesPath.tile_source,
+
+
1438 .filter(db_models.DbTilesPath.project_id == str(project_id))
+
+
+
+
1442 processed_tiles_list = [
+
+
+
1445 "project_id": x.project_id,
+
1446 "status": x.status.name,
+
1447 "source": x.tile_source,
+
1448 "format": Path(x.path).suffix[1:],
+
+
+
+
+
1453 return processed_tiles_list
+
+
1455 except Exception
as e:
+
+
1457 raise HTTPException(status_code=400, detail=str(e))
from e
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
1479 async
def get_pagination(page: int, count: int, results_per_page: int, total: int):
+
1480 """Pagination result for splash page."""
+
1481 total_pages = (count + results_per_page - 1) // results_per_page
+
1482 has_next = (page * results_per_page) < count
+
+
+
1485 pagination = project_schemas.PaginationInfo(
+
+
+
1488 next_num=page + 1
if has_next
else None,
+
+
+
1491 prev_num=page - 1
if has_prev
else None,
+
1492 per_page=results_per_page,
+
+
+
+
+
+
+
1499 async
def get_dashboard_detail(
+
1500 project: db_models.DbProject, db_organisation: db_models.DbOrganisation, db: Session
+
+
1502 """Get project details for project dashboard."""
+
1503 odk_central = await project_deps.get_odk_credentials(db, project.id)
+
1504 xform = central_crud.get_odk_form(odk_central)
+
1505 db_xform = await project_deps.get_project_xform(db, project.id)
+
+
1507 submission_meta_data = xform.getFullDetails(project.odkid, db_xform.odk_form_id)
+
1508 project.total_submission = submission_meta_data.get(
"submissions", 0)
+
1509 project.last_active = submission_meta_data.get(
"lastSubmission")
+
+
+
1512 db.query(db_models.DbTaskHistory.user_id)
+
+
1514 db_models.DbTaskHistory.project_id == project.id,
+
1515 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
+
1521 project.total_tasks = await tasks_crud.get_task_count_in_project(db, project.id)
+
1522 project.organisation_name, project.organisation_logo = (
+
1523 db_organisation.name,
+
1524 db_organisation.logo,
+
+
1526 project.total_contributors = contributors
+
+
+
+
+
1531 async
def get_project_users(db: Session, project_id: int, db_user: db_models.DbUser):
+
1532 """Get the users and their contributions for a project.
+
+
+
1535 db (Session): The database session.
+
1536 project_id (int): The ID of the project.
+
1537 db_user (DbUser): User that called the endpoint.
+
+
+
1540 List[Dict[str, Union[str, int]]]: A list of dictionaries containing
+
1541 the username and the number of contributions made by each user
+
1542 for the specified project.
+
+
+
+
+
1547 db.query(db_models.DbTaskHistory)
+
1548 .filter(db_models.DbTaskHistory.project_id == project_id)
+
+
+
+
1552 user.user_id
for user
in contributors
if user.user_id
is not None
+
+
+
+
1556 for user_id
in unique_user_ids:
+
1557 contributions = count_user_contributions(db, user_id, project_id)
+
1558 response.append({
"user": db_user.username,
"contributions": contributions})
+
+
1560 response = sorted(response, key=
lambda x: x[
"contributions"], reverse=
True)
+
+
+
+
1564 def count_user_contributions(db: Session, user_id: int, project_id: int) -> int:
+
1565 """Count contributions for a specific user.
+
+
+
1568 db (Session): The database session.
+
1569 user_id (int): The ID of the user.
+
1570 project_id (int): The ID of the project.
+
+
+
1573 int: The number of contributions made by the user for the specified
+
+
+
1576 contributions_count = (
+
1577 db.query(func.count(db_models.DbTaskHistory.user_id))
+
+
1579 db_models.DbTaskHistory.user_id == user_id,
+
1580 db_models.DbTaskHistory.project_id == project_id,
+
+
+
+
+
1585 return contributions_count
+
+
+
1588 async
def add_project_manager(
+
1589 db: Session, user: db_models.DbUser, project: db_models.DbProject
+
+
1591 """Adds a user as an manager to the specified project.
+
+
+
1594 db (Session): The database session.
+
1595 user (int): The ID of the user to be added as an admin.
+
1596 project (DbOrganisation): The Project model instance.
+
+
+
1599 Response: The HTTP response with status code 200.
+
+
1601 new_user_role = db_models.DbUserRoles(
+
+
1603 project_id=project.id,
+
1604 role=ProjectRole.PROJECT_MANAGER,
+
+
+
+
1608 project.roles.append(new_user_role)
+
+
+
1611 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¶