+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
18 """Logic for FMTM project routes."""
+
+
+
+
+
23 from asyncio
import gather
+
24 from io
import BytesIO
+
25 from pathlib
import Path
+
26 from typing
import List, Optional, Union
+
+
+
+
+
31 import shapely.wkb
as wkblib
+
32 from asgiref.sync
import async_to_sync
+
33 from fastapi
import HTTPException, Response
+
34 from fastapi.concurrency
import run_in_threadpool
+
35 from fmtm_splitter.splitter
import split_by_sql, split_by_square
+
36 from geoalchemy2.shape
import to_shape
+
37 from geojson.feature
import Feature, FeatureCollection
+
38 from loguru
import logger
as log
+
39 from osm_fieldwork.basemapper
import create_basemap_file
+
40 from osm_fieldwork.json2osm
import json2osm
+
41 from osm_fieldwork.OdkCentral
import OdkAppUser
+
42 from osm_fieldwork.xlsforms
import entities_registration, xlsforms_path
+
43 from osm_rawdata.postgres
import PostgresClient
+
44 from shapely.geometry
import shape
+
45 from sqlalchemy
import and_, column, func, select, table, text
+
46 from sqlalchemy.orm
import Session
+
+
+
+
50 from app.db import db_models
+
+
+
53 flatgeobuf_to_geojson,
+
54 geojson_to_flatgeobuf,
+
+
56 get_featcol_main_geom_type,
+
+
58 parse_and_filter_geojson,
+
59 split_geojson_by_task_areas,
+
60 task_geojson_dict_to_entity_values,
+
+
+
+
64 from app.s3 import add_obj_to_bucket
+
+
+
+
68 TILESDIR =
"/opt/tiles"
+
+
+
71 async
def get_projects(
+
+
+
+
75 user_id: Optional[int] =
None,
+
76 hashtags: Optional[List[str]] =
None,
+
77 search: Optional[str] =
None,
+
+
79 """Get all projects, or a filtered subset."""
+
+
+
+
83 filters.append(db_models.DbProject.author_id == user_id)
+
+
+
86 filters.append(db_models.DbProject.hashtags.op(
"&&")(hashtags))
+
+
+
+
90 db_models.DbProject.project_info.has(
+
91 db_models.DbProjectInfo.name.ilike(f
"%{search}%")
+
+
+
+
+
+
97 db.query(db_models.DbProject)
+
98 .filter(and_(*filters))
+
99 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
104 project_count = db.query(db_models.DbProject).filter(and_(*filters)).count()
+
+
+
+
108 db.query(db_models.DbProject)
+
+
110 db_models.DbProject.visibility
+
111 == ProjectVisibility.PUBLIC
+
+
113 .order_by(db_models.DbProject.id.desc())
+
+
+
+
+
118 project_count = db.query(db_models.DbProject).count()
+
+
120 filtered_projects = await convert_to_app_projects(db_projects)
+
121 return project_count, filtered_projects
+
+
+
124 async
def get_projects_featcol(
+
+
126 bbox: Optional[str] =
None,
+
127 ) -> geojson.FeatureCollection:
+
128 """Get all projects, or a filtered subset."""
+
+
130 """AND ST_Intersects(
+
131 p.outline, ST_MakeEnvelope(:minx, :miny, :maxx, :maxy, 4326)
+
+
+
+
+
+
+
+
139 minx, miny, maxx, maxy = map(float, bbox.split(
","))
+
140 bbox_params = {
"minx": minx,
"miny": miny,
"maxx": maxx,
"maxy": maxy}
+
+
+
+
144 SELECT jsonb_build_object(
+
145 'type', 'FeatureCollection',
+
146 'features', COALESCE(jsonb_agg(feature), '[]'::jsonb)
+
+
+
149 SELECT jsonb_build_object(
+
+
+
152 'geometry', ST_AsGeoJSON(p.outline)::jsonb,
+
153 'properties', jsonb_build_object(
+
+
+
156 'percentValidated', 0,
+
157 'created', p.created,
+
158 'link', concat('https://', :domain, '/project/', p.id)
+
+
+
+
162 LEFT JOIN project_info pi ON p.id = pi.project_id
+
163 WHERE p.visibility = 'PUBLIC'
+
+
+
+
+
+
169 result = db.execute(query, {
"domain": settings.FMTM_DOMAIN, **bbox_params})
+
170 featcol = result.scalar_one()
+
+
+
+
+
175 async
def get_project_summaries(
+
+
+
+
179 user_id: Optional[int] =
None,
+
180 hashtags: Optional[List[str]] =
None,
+
181 search: Optional[str] =
None,
+
+
183 """Get project summary details for main page."""
+
184 project_count, db_projects = await get_projects(
+
185 db, skip, limit, user_id, hashtags, search
+
+
187 return project_count, await convert_to_project_summaries(db, db_projects)
+
+
+
190 async
def get_project(db: Session, project_id: int):
+
191 """Get a single project."""
+
+
193 db.query(db_models.DbProject)
+
194 .filter(db_models.DbProject.id == project_id)
+
+
+
+
+
199 status_code=HTTPStatus.NOT_FOUND,
+
200 detail=f
"Project with id {project_id} does not exist",
+
+
+
+
+
205 async
def get_project_by_id(db: Session, project_id: int):
+
206 """Get a single project by id."""
+
+
208 db.query(db_models.DbProject)
+
209 .filter(db_models.DbProject.id == project_id)
+
+
+
212 return await convert_to_app_project(db_project)
+
+
+
215 async
def get_project_info_by_id(db: Session, project_id: int):
+
216 """Get the project info only by id."""
+
+
218 db.query(db_models.DbProjectInfo)
+
219 .filter(db_models.DbProjectInfo.project_id == project_id)
+
220 .order_by(db_models.DbProjectInfo.project_id)
+
+
+
223 return await convert_to_app_project_info(db_project_info)
+
+
+
226 async
def delete_one_project(db: Session, db_project: db_models.DbProject) ->
None:
+
227 """Delete a project by id."""
+
+
229 project_id = db_project.id
+
230 db.delete(db_project)
+
+
232 log.info(f
"Deleted project with ID: {project_id}")
+
233 except Exception
as e:
+
+
235 raise HTTPException(e)
from e
+
+
+
238 async
def partial_update_project_info(
+
+
240 project_metadata: project_schemas.ProjectPartialUpdate,
+
241 db_project: db_models.DbProject,
+
+
243 """Partial project update for PATCH."""
+
+
245 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
248 if db_project
and db_project_info:
+
249 if project_metadata.name:
+
250 db_project.project_name_prefix = project_metadata.project_name_prefix
+
251 db_project_info.name = project_metadata.name
+
252 if project_metadata.description:
+
253 db_project_info.description = project_metadata.description
+
254 if project_metadata.short_description:
+
255 db_project_info.short_description = project_metadata.short_description
+
256 if project_metadata.per_task_instructions:
+
257 db_project_info.per_task_instructions = (
+
258 project_metadata.per_task_instructions
+
+
260 if project_metadata.hashtags:
+
261 db_project.hashtags = project_metadata.hashtags
+
+
+
264 db.refresh(db_project)
+
+
266 return await convert_to_app_project(db_project)
+
+
+
269 async
def update_project_info(
+
+
271 project_metadata: project_schemas.ProjectUpdate,
+
272 db_project: db_models.DbProject,
+
273 db_user: db_models.DbUser,
+
+
275 """Full project update for PUT."""
+
276 project_info = project_metadata.project_info
+
+
+
+
280 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
281 detail=
"No project info passed in",
+
+
+
+
285 project_info = project_metadata.project_info
+
+
+
288 db_project.author_id = db_user.id
+
289 db_project.project_name_prefix = project_metadata.project_name_prefix
+
+
+
292 db_project_info = await get_project_info_by_id(db, db_project.id)
+
+
+
295 if db_project
and db_project_info:
+
296 db_project_info.name = project_info.name
+
297 db_project_info.short_description = project_info.short_description
+
298 db_project_info.description = project_info.description
+
+
+
301 db.refresh(db_project)
+
+
303 return await convert_to_app_project(db_project)
+
+
+
306 async
def create_project_with_project_info(
+
+
308 project_metadata: project_schemas.ProjectUpload,
+
+
310 current_user: db_models.DbUser,
+
+
312 """Create a new project, including all associated info."""
+
313 if not odk_project_id:
+
+
315 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
316 detail=
"ODK Central project id is missing",
+
+
+
+
320 "Creating project in FMTM database with vars: "
+
321 f
"project_user: {current_user.username} | "
+
322 f
"project_name: {project_metadata.project_info.name} | "
+
323 f
"xform_category: {project_metadata.xform_category} | "
+
324 f
"hashtags: {project_metadata.hashtags} | "
+
325 f
"organisation_id: {project_metadata.organisation_id}"
+
+
+
+
329 db_project = db_models.DbProject(
+
330 author_id=current_user.id,
+
331 odkid=odk_project_id,
+
332 **project_metadata.model_dump(exclude=[
"project_info",
"outline_geojson"]),
+
+
+
+
+
337 db_project_info = db_models.DbProjectInfo(
+
+
339 name=project_metadata.project_info.name,
+
340 description=project_metadata.project_info.description,
+
341 short_description=project_metadata.project_info.short_description,
+
342 per_task_instructions=project_metadata.project_info.per_task_instructions,
+
+
344 db.add(db_project_info)
+
+
+
347 db.refresh(db_project)
+
+
+
350 generated_project_id = db_project.id
+
351 if db_project.hashtags:
+
352 db_project.hashtags = db_project.hashtags + [
+
353 f
"{settings.FMTM_DOMAIN}-{generated_project_id}"
+
+
+
+
357 return await convert_to_app_project(db_project)
+
+
+
360 async
def create_tasks_from_geojson(
+
+
+
+
+
365 """Create tasks for a project, from provided task boundaries."""
+
+
367 if isinstance(boundaries, str):
+
368 boundaries = json.loads(boundaries)
+
+
+
371 if boundaries[
"type"] ==
"Feature":
+
372 polygons = [boundaries]
+
+
374 polygons = boundaries[
"features"]
+
375 log.debug(f
"Processing {len(polygons)} task geometries")
+
376 for index, polygon
in enumerate(polygons):
+
+
378 if polygon[
"geometry"][
"type"] ==
"MultiPolygon":
+
379 log.debug(
"Converting MultiPolygon to Polygon")
+
380 polygon[
"geometry"][
"type"] =
"Polygon"
+
381 polygon[
"geometry"][
"coordinates"] = polygon[
"geometry"][
"coordinates"][
+
+
+
+
385 db_task = db_models.DbTask(
+
386 project_id=project_id,
+
387 outline=wkblib.dumps(shape(polygon[
"geometry"]), hex=
True),
+
388 project_task_index=index + 1,
+
+
+
+
392 "Created database task | "
+
393 f
"Project ID {project_id} | "
+
394 f
"Task index {index}"
+
+
+
+
+
+
400 log.debug(
"COMPLETE: creating project boundary, based on task boundaries")
+
+
+
403 except Exception
as e:
+
+
405 raise HTTPException(e)
from e
+
+
+
408 async
def preview_split_by_square(boundary: str, meters: int):
+
409 """Preview split by square for a project boundary.
+
+
411 Use a lambda function to remove the "z" dimension from each
+
412 coordinate in the feature's geometry.
+
+
414 boundary = merge_multipolygon(boundary)
+
+
416 return await run_in_threadpool(
+
417 lambda: split_by_square(
+
+
+
+
+
+
+
424 async
def generate_data_extract(
+
425 aoi: Union[FeatureCollection, Feature, dict],
+
426 extract_config: Optional[BytesIO] =
None,
+
+
428 """Request a new data extract in flatgeobuf format.
+
+
+
+
+
433 aoi (Union[FeatureCollection, Feature, dict]):
+
434 Area of interest for data extraction.
+
435 extract_config (Optional[BytesIO], optional):
+
436 Configuration for data extraction. Defaults to None.
+
+
+
+
440 When necessary parameters are missing or data extraction fails.
+
+
+
+
444 URL for the flatgeobuf data extract.
+
+
446 if not extract_config:
+
+
448 status_code=HTTPStatus.BAD_REQUEST,
+
449 detail=
"To generate a new data extract a form_category must be specified.",
+
+
+
+
+
+
455 auth_token=settings.RAW_DATA_API_AUTH_TOKEN
+
456 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
459 fgb_url = pg.execQuery(
+
+
+
+
463 f
"fmtm/{settings.FMTM_DOMAIN}/data_extract"
+
464 if settings.RAW_DATA_API_AUTH_TOKEN
+
+
+
+
+
469 "useStWithin":
False,
+
470 "fgb_wrap_geoms":
True,
+
+
+
+
+
475 msg =
"Could not get download URL for data extract. Did the API change?"
+
+
+
478 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
+
485 async
def split_geojson_into_tasks(
+
+
487 project_geojson: Union[dict, FeatureCollection],
+
488 no_of_buildings: int,
+
489 extract_geojson: Optional[FeatureCollection] =
None,
+
+
491 """Splits a project into tasks.
+
+
+
494 db (Session): A database session.
+
495 project_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
+
497 extract_geojson (Union[dict, FeatureCollection]): A GeoJSON of the project
+
498 boundary osm data extract (features).
+
499 extract_geojson (FeatureCollection): A GeoJSON of the project
+
500 boundary osm data extract (features).
+
501 If not included, an extract is generated automatically.
+
502 no_of_buildings (int): The number of buildings to include in each task.
+
+
+
505 Any: A GeoJSON object containing the tasks for the specified project.
+
+
507 log.debug(
"STARTED task splitting using provided boundary and data extract")
+
508 features = await run_in_threadpool(
+
509 lambda: split_by_sql(
+
+
+
512 num_buildings=no_of_buildings,
+
513 osm_extract=extract_geojson,
+
+
+
516 log.debug(
"COMPLETE task splitting")
+
+
+
+
+
+
+
+
+
525 async
def read_and_insert_xlsforms(db, directory):
+
526 """Read the list of XLSForms from the disk and insert to DB."""
+
527 existing_titles = set(
+
528 title
for (title,)
in db.query(db_models.DbXLSForm.title).all()
+
+
+
+
532 for file
in Path(directory).glob(
"*.xls")
+
533 if not file.stem.startswith(
"entities")
+
+
+
+
537 for xlsform_name
in xlsforms_on_disk:
+
538 file_path = Path(directory) / f
"{xlsform_name}.xls"
+
+
540 if file_path.stat().st_size == 0:
+
541 log.warning(f
"{file_path} is empty!")
+
+
+
544 with open(file_path,
"rb")
as xls:
+
+
+
+
+
+
550 INSERT INTO xlsforms (title, xls)
+
551 VALUES (:title, :xls)
+
552 ON CONFLICT (title) DO UPDATE SET
+
553 title = EXCLUDED.title, xls = EXCLUDED.xls
+
+
+
556 db.execute(insert_query, {
"title": xlsform_name,
"xls": data})
+
+
558 log.info(f
"Inserted or updated {xlsform_name} xlsform to database")
+
+
560 except Exception
as e:
+
+
562 f
"Failed to insert or update {xlsform_name} in the database. Error: {e}"
+
+
+
+
566 for title
in existing_titles - set(xlsforms_on_disk):
+
+
+
569 DELETE FROM xlsforms WHERE title = :title
+
+
+
572 db.execute(delete_query, {
"title": title})
+
+
574 log.info(f
"Deleted {title} from the database as it was not found on disk.")
+
+
576 return xlsforms_on_disk
+
+
+
579 async
def get_odk_id_for_project(db: Session, project_id: int):
+
580 """Get the odk project id for the fmtm project id."""
+
+
+
+
+
+
586 where = f
"id={project_id}"
+
587 sql = select(project).where(text(where))
+
+
589 result = db.execute(sql)
+
+
+
592 if result.rowcount != 1:
+
593 log.warning(str(sql))
+
+
595 project_info = result.first()
+
596 return project_info.odkid
+
+
+
599 async
def get_or_set_data_extract_url(
+
+
+
+
+
604 """Get or set the data extract URL for a project.
+
+
+
607 db (Session): SQLAlchemy database session.
+
608 project_id (int): The ID of the project.
+
609 url (str): URL to the streamable flatgeobuf data extract.
+
610 If not passed, a new extract is generated.
+
+
+
613 str: URL to fgb file in S3.
+
+
615 db_project = await get_project_by_id(db, project_id)
+
+
617 msg = f
"Project ({project_id}) not found"
+
+
619 raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=msg)
+
+
+
+
+
624 existing_url = db_project.data_extract_url
+
+
+
+
628 f
"No data extract exists for project ({project_id}). "
+
629 "To generate one, call 'projects/generate-data-extract/'"
+
+
+
632 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
+
+
+
637 extract_type =
"polygon"
+
+
639 await update_data_extract_url_in_db(db, db_project, url, extract_type)
+
+
+
+
+
644 async
def update_data_extract_url_in_db(
+
645 db: Session, project: db_models.DbProject, url: str, extract_type: str
+
+
647 """Update the data extract params in the database for a project."""
+
648 log.debug(f
"Setting data extract URL for project ({project.id}): {url}")
+
649 project.data_extract_url = url
+
650 project.data_extract_type = extract_type
+
+
+
+
654 async
def upload_custom_extract_to_s3(
+
+
+
+
658 data_extract_type: str,
+
+
660 """Uploads custom data extracts to S3.
+
+
+
663 db (Session): SQLAlchemy database session.
+
664 project_id (int): The ID of the project.
+
665 fgb_content (bytes): Content of read flatgeobuf file.
+
666 data_extract_type (str): centroid/polygon/line for database.
+
+
+
669 str: URL to fgb file in S3.
+
+
671 project = await get_project(db, project_id)
+
672 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
+
675 raise HTTPException(status_code=404, detail=
"Project not found")
+
+
677 fgb_obj = BytesIO(fgb_content)
+
678 s3_fgb_path = f
"{project.organisation_id}/{project_id}/custom_extract.fgb"
+
+
680 log.debug(f
"Uploading fgb to S3 path: /{s3_fgb_path}")
+
+
682 settings.S3_BUCKET_NAME,
+
+
+
685 content_type=
"application/octet-stream",
+
+
+
+
+
690 f
"{settings.S3_DOWNLOAD_ROOT}/{settings.S3_BUCKET_NAME}/{s3_fgb_path}"
+
+
+
693 await update_data_extract_url_in_db(db, project, s3_fgb_full_url, data_extract_type)
+
+
695 return s3_fgb_full_url
+
+
+
698 async
def upload_custom_fgb_extract(
+
+
+
+
+
703 """Upload a flatgeobuf data extract.
+
+
705 FIXME how can we validate this has the required fields for geojson conversion?
+
+
+
+
+
+
+
+
+
+
715 db (Session): SQLAlchemy database session.
+
716 project_id (int): The ID of the project.
+
717 fgb_content (bytes): Content of read flatgeobuf file.
+
+
+
720 str: URL to fgb file in S3.
+
+
722 featcol = await flatgeobuf_to_geojson(db, fgb_content)
+
+
+
725 msg = f
"Failed extracting geojson from flatgeobuf for project ({project_id})"
+
+
727 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
729 data_extract_type = await get_data_extract_type(featcol)
+
+
731 return await upload_custom_extract_to_s3(
+
+
+
+
+
+
+
+
739 async
def get_data_extract_type(featcol: FeatureCollection) -> str:
+
740 """Determine predominant geometry type for extract."""
+
741 geom_type = get_featcol_main_geom_type(featcol)
+
742 if geom_type
not in [
"Polygon",
"Polyline",
"Point"]:
+
+
744 "Extract does not contain valid geometry types, from 'Polygon' "
+
745 ", 'Polyline' and 'Point'."
+
+
+
748 raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg)
+
+
750 "Polygon":
"polygon",
+
+
+
+
754 data_extract_type = geom_name_map.get(geom_type,
"polygon")
+
+
756 return data_extract_type
+
+
+
759 async
def upload_custom_geojson_extract(
+
+
+
762 geojson_raw: Union[str, bytes],
+
+
764 """Upload a geojson data extract.
+
+
+
767 db (Session): SQLAlchemy database session.
+
768 project_id (int): The ID of the project.
+
769 geojson_raw (str): The custom data extracts contents.
+
+
+
772 str: URL to fgb file in S3.
+
+
774 project = await get_project(db, project_id)
+
775 log.debug(f
"Uploading custom data extract for project: {project}")
+
+
777 featcol_filtered = parse_and_filter_geojson(geojson_raw)
+
778 if not featcol_filtered:
+
+
780 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
781 detail=
"Could not process geojson input",
+
+
+
784 await check_crs(featcol_filtered)
+
+
786 data_extract_type = await get_data_extract_type(featcol_filtered)
+
+
+
789 "Generating fgb object from geojson with "
+
790 f
"{len(featcol_filtered.get('features', []))} features"
+
+
792 fgb_data = await geojson_to_flatgeobuf(db, featcol_filtered)
+
+
+
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
+
+
837 appuser_name = f
"fmtm_user_{form_category}"
+
+
839 f
"Creating ODK appuser ({appuser_name}) for ODK project ({project_odk_id})"
+
+
841 appuser = OdkAppUser(
+
842 odk_credentials.odk_central_url,
+
843 odk_credentials.odk_central_user,
+
844 odk_credentials.odk_central_password,
+
+
846 appuser_json = appuser.create(project_odk_id, appuser_name)
+
+
+
+
850 msg = f
"Couldn't create appuser {appuser_name} for project"
+
+
+
853 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
855 if not (appuser_token := appuser_json.get(
"token")):
+
856 msg = f
"Couldn't get token for appuser {appuser_name}"
+
+
+
859 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
861 appuser_id = appuser_json.get(
"id")
+
+
+
+
865 with open(entities_registration,
"rb")
as f:
+
866 registration_xlsform = BytesIO(f.read())
+
867 registration_xform = await central_crud.read_and_test_xform(
+
868 registration_xlsform,
"xls", return_form_data=
True
+
+
+
871 updated_reg_xform = await central_crud.update_entity_registration_xform(
+
872 registration_xform, form_category
+
+
+
875 log.info(
"Uploading Entity registration XForm to ODK Central")
+
876 central_crud.create_odk_xform(
+
+
+
+
+
+
+
883 xform = await central_crud.read_and_test_xform(
+
884 xlsform, form_file_ext, return_form_data=
True
+
+
+
887 updated_xform = await central_crud.update_survey_xform(
+
+
+
+
+
+
893 log.info(
"Uploading survey XForm to ODK Central")
+
894 xform_id = central_crud.create_odk_xform(
+
+
+
+
+
+
900 log.info(
"Updating XForm role for appuser in ODK Central")
+
+
902 response = appuser.updateRole(
+
903 projectId=project_odk_id,
+
+
+
+
+
+
909 json_data = response.json()
+
+
911 except json.decoder.JSONDecodeError:
+
+
913 "Could not parse response json during appuser update. "
+
914 f
"status_code={response.status_code}"
+
+
+
917 msg = f
"Failed to update appuser for formId: ({xform_id})"
+
+
+
920 status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail=msg
+
+
+
+
+
925 project_id, odk_form_id, category
+
+
+
928 :project_id, :xform_id, :category
+
+
+
+
+
+
934 {
"project_id": project.id,
"xform_id": xform_id,
"category": form_category},
+
+
+
937 odk_url = odk_credentials.odk_central_url
+
938 return encrypt_value(f
"{odk_url}/v1/key/{appuser_token}/projects/{project_odk_id}")
+
+
+
941 async
def generate_project_files(
+
+
943 project: db_models.DbProject,
+
944 custom_form: Optional[BytesIO],
+
+
946 background_task_id: Optional[uuid.UUID] =
None,
+
+
948 """Generate the files for a project.
+
+
950 QR code (appuser), ODK XForm, ODK Entities from OSM data extract.
+
+
+
953 db (Session): the database session.
+
954 project (DbProject): FMTM database project.
+
955 custom_form (BytesIO): the xls file to upload if we have a custom form
+
956 form_file_ext (str): weather the form is xls, xlsx or xml
+
957 background_task_id (uuid): the task_id of the background task.
+
+
+
960 project_id = project.id
+
961 form_category = project.xform_category
+
962 log.info(f
"Starting generate_project_files for project {project_id}")
+
963 odk_credentials = await project_deps.get_odk_credentials(db, project_id)
+
+
+
966 log.debug(
"User provided custom XLSForm")
+
967 xlsform = custom_form
+
+
969 log.debug(f
"Using default XLSForm for category: '{form_category}'")
+
+
971 xlsform_path = f
"{xlsforms_path}/{form_category}.xls"
+
972 with open(xlsform_path,
"rb")
as f:
+
973 xlsform = BytesIO(f.read())
+
+
+
976 log.debug(
"Getting data extract geojson from flatgeobuf")
+
977 feature_collection = await get_project_features_geojson(db, project)
+
+
+
980 log.debug(
"Splitting data extract per task area")
+
981 task_extract_dict = await split_geojson_by_task_areas(
+
982 db, feature_collection, project_id
+
+
+
985 if not task_extract_dict:
+
986 log.warning(f
"Project ({project_id}) failed splitting tasks")
+
+
988 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
989 detail=
"Failed splitting extract by tasks.",
+
+
+
+
993 project_odk_id = project.odkid
+
+
995 encrypted_odk_token = await generate_odk_central_project_content(
+
+
+
+
+
+
1001 list(task_extract_dict.keys()),
+
+
+
+
1005 f
"Setting odk token for FMTM project ({project_id}) "
+
1006 f
"ODK project {project_odk_id}"
+
+
1008 project.odk_token = encrypted_odk_token
+
+
1010 for task
in project.tasks:
+
+
1012 task_features = task_extract_dict.get(task.project_task_index, {})
+
1013 feature_count = len(task_features.get(
"features", []))
+
1014 task.feature_count = feature_count
+
+
1016 f
"{feature_count} features added for task {task.project_task_index}"
+
+
+
+
+
+
+
1023 entities_data_dict = await task_geojson_dict_to_entity_values(task_extract_dict)
+
+
+
+
1027 async
with central_deps.get_odk_entity(odk_credentials)
as odk_central:
+
1028 entities = await odk_central.createEntities(
+
+
+
+
+
+
1034 log.debug(f
"Wrote {len(entities)} entities for project ({project_id})")
+
+
1036 log.debug(f
"No entities uploaded for project ({project_id})")
+
+
1038 if background_task_id:
+
+
1040 await update_background_task_status_in_database(
+
1041 db, background_task_id, 4
+
+
+
1044 except Exception
as e:
+
+
+
1047 if background_task_id:
+
+
1049 await update_background_task_status_in_database(
+
1050 db, background_task_id, 2, str(e)
+
+
+
+
+
+
+
1057 async
def get_project_geometry(db: Session, project_id: int):
+
1058 """Retrieves the geometry of a project.
+
+
+
1061 db (Session): The database session.
+
1062 project_id (int): The ID of the project.
+
+
+
1065 str: A geojson of the project outline.
+
+
1067 projects = table(
"projects", column(
"outline"), column(
"id"))
+
1068 where = f
"projects.id={project_id}"
+
1069 sql = select(geoalchemy2.functions.ST_AsGeoJSON(projects.c.outline)).where(
+
+
+
1072 result = db.execute(sql)
+
+
1074 if result.rowcount != 1:
+
1075 log.warning(str(sql))
+
+
1077 row = eval(result.first()[0])
+
1078 return json.dumps(row)
+
+
+
1081 async
def get_task_geometry(db: Session, project_id: int):
+
1082 """Retrieves the geometry of tasks associated with a project.
+
+
+
1085 db (Session): The database session.
+
1086 project_id (int): The ID of the project.
+
+
+
1089 str: A geojson of the task boundaries
+
+
1091 db_tasks = await tasks_crud.get_tasks(db, project_id,
None)
+
+
1093 for task
in db_tasks:
+
1094 geom = to_shape(task.outline)
+
+
1096 geometry = geom.__geo_interface__
+
+
+
+
1100 feature = {
"type":
"Feature",
"geometry": geometry,
"properties": properties}
+
1101 features.append(feature)
+
+
1103 feature_collection = {
"type":
"FeatureCollection",
"features": features}
+
1104 return json.dumps(feature_collection)
+
+
+
1107 async
def get_project_features_geojson(
+
+
1109 project: Union[db_models.DbProject, int],
+
1110 task_id: Optional[int] =
None,
+
1111 ) -> FeatureCollection:
+
1112 """Get a geojson of all features for a task."""
+
1113 if isinstance(project, int):
+
1114 db_project = await get_project(db, project)
+
+
1116 db_project = project
+
1117 project_id = db_project.id
+
+
1119 data_extract_url = db_project.data_extract_url
+
+
1121 if not data_extract_url:
+
1122 raise HTTPException(
+
1123 status_code=HTTPStatus.NOT_FOUND,
+
1124 detail=f
"No data extract exists for project ({project_id})",
+
+
+
+
1128 data_extract_url = data_extract_url.replace(
+
1129 settings.S3_DOWNLOAD_ROOT,
+
1130 settings.S3_ENDPOINT,
+
+
+
1133 with requests.get(data_extract_url)
as response:
+
+
1135 msg = f
"Download failed for data extract, project ({project_id})"
+
+
1137 raise HTTPException(
+
1138 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
1141 log.debug(
"Converting download flatgeobuf to geojson")
+
1142 data_extract_geojson = await flatgeobuf_to_geojson(db, response.content)
+
+
1144 if not data_extract_geojson:
+
1145 msg =
"Failed to convert flatgeobuf --> geojson for " f
"project ({project_id})"
+
+
1147 raise HTTPException(
+
1148 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
+
+
+
+
+
1154 split_extract_dict = await split_geojson_by_task_areas(
+
1155 db, data_extract_geojson, project_id
+
+
1157 if not split_extract_dict:
+
1158 raise HTTPException(
+
1159 status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+
1160 detail=(f
"Failed to extract geojson for task ({task_id})"),
+
+
1162 return split_extract_dict[task_id]
+
+
1164 return data_extract_geojson
+
+
+
1167 async
def get_json_from_zip(zip, filename: str, error_detail: str):
+
1168 """Extract json file from zip."""
+
+
1170 with zip.open(filename)
as file:
+
+
1172 return json.loads(data)
+
1173 except Exception
as e:
+
+
1175 raise HTTPException(
+
1176 status_code=400, detail=f
"{error_detail} ----- Error: {e}"
+
+
+
+
1180 async
def get_outline_from_geojson_file_in_zip(
+
1181 zip, filename: str, error_detail: str, feature_index: int = 0
+
+
1183 """Parse geojson outline within a zip."""
+
+
1185 with zip.open(filename)
as file:
+
+
1187 json_dump = json.loads(data)
+
1188 await check_crs(json_dump)
+
1189 feature_collection = FeatureCollection(json_dump)
+
1190 feature = feature_collection[
"features"][feature_index]
+
1191 geom = feature[
"geometry"]
+
1192 shape_from_geom = shape(geom)
+
1193 return shape_from_geom
+
1194 except Exception
as e:
+
+
1196 raise HTTPException(
+
+
1198 detail=f
"{error_detail} ----- Error: {e} ----",
+
+
+
+
1202 async
def get_shape_from_json_str(feature: str, error_detail: str):
+
1203 """Parse geojson outline within a zip to shapely geom."""
+
+
1205 geom = feature[
"geometry"]
+
+
1207 except Exception
as e:
+
+
1209 raise HTTPException(
+
+
1211 detail=f
"{error_detail} ----- Error: {e} ---- Json: {feature}",
+
+
+
+
+
+
+
+
+
+
+
1222 async
def convert_to_app_project(db_project: db_models.DbProject):
+
1223 """Legacy function to convert db models --> Pydantic.
+
+
1225 TODO refactor to use Pydantic model methods instead.
+
+
+
1228 log.debug(
"convert_to_app_project called, but no project provided")
+
+
+
1231 app_project = db_project
+
+
1233 if db_project.outline:
+
1234 app_project.outline_geojson = geometry_to_geojson(
+
1235 db_project.outline, {
"id": db_project.id}, db_project.id
+
+
+
1238 app_project.tasks = db_project.tasks
+
+
+
+
+
1243 async
def convert_to_app_project_info(db_project_info: db_models.DbProjectInfo):
+
1244 """Legacy function to convert db models --> Pydantic.
+
+
1246 TODO refactor to use Pydantic model methods instead.
+
+
+
1249 app_project_info: project_schemas.ProjectInfo = db_project_info
+
1250 return app_project_info
+
+
+
+
+
1255 async
def convert_to_app_projects(
+
1256 db_projects: List[db_models.DbProject],
+
1257 ) -> List[project_schemas.ProjectOut]:
+
1258 """Legacy function to convert db models --> Pydantic.
+
+
1260 TODO refactor to use Pydantic model methods instead.
+
+
1262 if db_projects
and len(db_projects) > 0:
+
+
1264 async
def convert_project(project):
+
1265 return await convert_to_app_project(project)
+
+
1267 app_projects = await gather(
+
1268 *[convert_project(project)
for project
in db_projects]
+
+
1270 return [project
for project
in app_projects
if project
is not None]
+
+
+
+
+
1275 async
def convert_to_project_summary(db: Session, db_project: db_models.DbProject):
+
1276 """Legacy function to convert db models --> Pydantic.
+
+
1278 TODO refactor to use Pydantic model methods instead.
+
+
+
1281 summary: project_schemas.ProjectSummary = db_project
+
+
1283 if db_project.project_info:
+
1284 summary.title = db_project.project_info.name
+
1285 summary.description = db_project.project_info.short_description
+
1286 summary.num_contributors = (
+
1287 db.query(db_models.DbTaskHistory.user_id)
+
+
1289 db_models.DbTaskHistory.project_id == db_project.id,
+
1290 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
1295 summary.organisation_logo = (
+
1296 db_project.organisation.logo
if db_project.organisation
else None
+
+
+
+
+
+
+
+
1304 async
def convert_to_project_summaries(
+
+
1306 db_projects: List[db_models.DbProject],
+
1307 ) -> List[project_schemas.ProjectSummary]:
+
1308 """Legacy function to convert db models --> Pydantic.
+
+
1310 TODO refactor to use Pydantic model methods instead.
+
+
1312 if db_projects
and len(db_projects) > 0:
+
+
1314 async
def convert_summary(db, project):
+
1315 return await convert_to_project_summary(db, project)
+
+
1317 project_summaries = await gather(
+
1318 *[convert_summary(db, project)
for project
in db_projects]
+
+
1320 return [summary
for summary
in project_summaries
if summary
is not None]
+
+
+
+
+
1325 async
def get_background_task_status(task_id: uuid.UUID, db: Session):
+
1326 """Get the status of a background task."""
+
+
1328 db.query(db_models.BackgroundTasks)
+
1329 .filter(db_models.BackgroundTasks.id == str(task_id))
+
+
+
+
1333 log.warning(f
"No background task with found with UUID: {task_id}")
+
1334 raise HTTPException(status_code=404, detail=
"Task not found")
+
1335 return task.status, task.message
+
+
+
1338 async
def insert_background_task_into_database(
+
1339 db: Session, name: str =
None, project_id: str =
None
+
+
1341 """Inserts a new task into the database.
+
+
+
1344 db (Session): database session
+
1345 name (str): name of the task.
+
1346 project_id (str): associated project id
+
+
+
1349 task_id (uuid.uuid4): The background task uuid for tracking.
+
+
1351 task_id = uuid.uuid4()
+
+
1353 task = db_models.BackgroundTasks(
+
1354 id=str(task_id), name=name, status=1, project_id=project_id
+
+
+
+
+
+
+
+
+
+
1364 async
def update_background_task_status_in_database(
+
1365 db: Session, task_id: uuid.UUID, status: int, message: str =
None
+
+
1367 """Updates the status of a task in the database.
+
+
+
1370 db (Session): database session.
+
1371 task_id (uuid.UUID): uuid of the task.
+
1372 status (int): status of the task.
+
1373 message (str): optional message to add to the db task.
+
+
+
+
+
1378 db.query(db_models.BackgroundTasks).filter(
+
1379 db_models.BackgroundTasks.id == str(task_id)
+
+
+
1382 db_models.BackgroundTasks.status: status,
+
1383 db_models.BackgroundTasks.message: message,
+
+
+
+
+
+
+
+
1391 async
def get_extracted_data_from_db(db: Session, project_id: int, outfile: str):
+
1392 """Get the geojson of those features for this project."""
+
+
1394 f
"""SELECT jsonb_build_object(
+
1395 'type', 'FeatureCollection',
+
1396 'features', jsonb_agg(feature)
+
+
+
1399 SELECT jsonb_build_object(
+
+
+
1402 'geometry', ST_AsGeoJSON(geometry)::jsonb,
+
1403 'properties', properties
+
+
+
1406 WHERE project_id={project_id}
+
+
+
+
1410 result = db.execute(query)
+
1411 features = result.fetchone()[0]
+
+
+
+
1415 with open(outfile,
"w")
as jsonfile:
+
1416 jsonfile.truncate(0)
+
1417 geojson.dump(features, jsonfile)
+
+
+
+
1421 def get_project_tiles(
+
+
+
1424 background_task_id: uuid.UUID,
+
+
1426 output_format: str =
"mbtiles",
+
+
+
1429 """Get the tiles for a project.
+
+
+
1432 db (Session): SQLAlchemy db session.
+
1433 project_id (int): ID of project to create tiles for.
+
1434 background_task_id (uuid.UUID): UUID of background task to track.
+
1435 source (str): Tile source ("esri", "bing", "topo", "google", "oam").
+
1436 output_format (str, optional): Default "mbtiles".
+
1437 Other options: "pmtiles", "sqlite3".
+
1438 tms (str, optional): Default None. Custom TMS provider URL.
+
+
+
1441 tiles_dir = f
"{TILESDIR}/{project_id}"
+
1442 outfile = f
"{tiles_dir}/{project_id}_{source}tiles.{output_format}"
+
+
1444 tile_path_instance = db_models.DbTilesPath(
+
1445 project_id=project_id,
+
1446 background_task_id=str(background_task_id),
+
+
+
+
+
+
+
1453 db.add(tile_path_instance)
+
+
+
+
1457 log.debug(f
"Getting bbox for project: {project_id}")
+
+
1459 f
"""SELECT ST_XMin(ST_Envelope(outline)) AS min_lon,
+
1460 ST_YMin(ST_Envelope(outline)) AS min_lat,
+
1461 ST_XMax(ST_Envelope(outline)) AS max_lon,
+
1462 ST_YMax(ST_Envelope(outline)) AS max_lat
+
+
1464 WHERE id = {project_id};"""
+
+
+
1467 result = db.execute(query)
+
1468 project_bbox = result.fetchone()
+
1469 log.debug(f
"Extracted project bbox: {project_bbox}")
+
+
+
1472 min_lon, min_lat, max_lon, max_lat = project_bbox
+
+
1474 log.error(f
"Failed to get bbox from project: {project_id}")
+
+
+
1477 "Creating basemap with params: "
+
1478 f
"boundary={min_lon},{min_lat},{max_lon},{max_lat} | "
+
1479 f
"outfile={outfile} | "
+
+
1481 f
"outdir={tiles_dir} | "
+
1482 f
"source={source} | "
+
+
+
+
+
+
+
1489 if output_format ==
"pmtiles":
+
1490 create_basemap_file(
+
1491 boundary=f
"{min_lon},{min_lat},{max_lon},{max_lat}",
+
1492 outfile=outfile.replace(
"pmtiles",
"mbtiles"),
+
+
+
+
+
+
+
+
1500 "pmtiles convert " f
"{outfile.replace('pmtiles', 'mbtiles')} {outfile}",
+
+
+
+
1504 create_basemap_file(
+
1505 boundary=f
"{min_lon},{min_lat},{max_lon},{max_lat}",
+
+
+
+
+
+
+
+
+
1514 log.info(f
"Basemap created for project ID {project_id}: {outfile}")
+
+
1516 tile_path_instance.status = 4
+
+
+
+
1520 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1521 update_bg_task_sync(db, background_task_id, 4)
+
+
1523 log.info(f
"Tiles generation process completed for project id {project_id}")
+
+
1525 except Exception
as e:
+
1526 log.exception(str(e))
+
1527 log.error(f
"Tiles generation process failed for project id {project_id}")
+
+
1529 tile_path_instance.status = 2
+
+
+
+
1533 update_bg_task_sync = async_to_sync(update_background_task_status_in_database)
+
1534 update_bg_task_sync(db, background_task_id, 2, str(e))
+
+
+
1537 async
def get_mbtiles_list(db: Session, project_id: int):
+
1538 """List mbtiles in database for a project."""
+
+
+
+
1542 db_models.DbTilesPath.id,
+
1543 db_models.DbTilesPath.project_id,
+
1544 db_models.DbTilesPath.status,
+
1545 db_models.DbTilesPath.tile_source,
+
+
1547 .filter(db_models.DbTilesPath.project_id == str(project_id))
+
+
+
+
1551 processed_tiles_list = [
+
+
+
1554 "project_id": x.project_id,
+
1555 "status": x.status.name,
+
1556 "tile_source": x.tile_source,
+
+
+
+
+
1561 return processed_tiles_list
+
+
1563 except Exception
as e:
+
+
1565 raise HTTPException(status_code=400, detail=str(e))
from e
+
+
+
1568 async
def convert_geojson_to_osm(geojson_file: str):
+
1569 """Convert a GeoJSON file to OSM format."""
+
1570 return json2osm(geojson_file)
+
+
+
1573 async
def get_pagination(page: int, count: int, results_per_page: int, total: int):
+
1574 """Pagination result for splash page."""
+
1575 total_pages = (count + results_per_page - 1) // results_per_page
+
1576 has_next = (page * results_per_page) < count
+
+
+
1579 pagination = project_schemas.PaginationInfo(
+
+
+
1582 next_num=page + 1
if has_next
else None,
+
+
+
1585 prev_num=page - 1
if has_prev
else None,
+
1586 per_page=results_per_page,
+
+
+
+
+
+
+
1593 async
def get_dashboard_detail(
+
1594 project: db_models.DbProject, db_organisation: db_models.DbOrganisation, db: Session
+
+
1596 """Get project details for project dashboard."""
+
1597 odk_central = await project_deps.get_odk_credentials(db, project.id)
+
1598 xform = central_crud.get_odk_form(odk_central)
+
1599 db_xform = await project_deps.get_project_xform(db, project.id)
+
+
1601 submission_meta_data = xform.getFullDetails(project.odkid, db_xform.odk_form_id)
+
1602 project.total_submission = submission_meta_data.get(
"submissions", 0)
+
1603 project.last_active = submission_meta_data.get(
"lastSubmission")
+
+
+
1606 db.query(db_models.DbTaskHistory.user_id)
+
+
1608 db_models.DbTaskHistory.project_id == project.id,
+
1609 db_models.DbTaskHistory.user_id
is not None,
+
+
+
+
+
+
1615 project.total_tasks = await tasks_crud.get_task_count_in_project(db, project.id)
+
1616 project.organisation_name, project.organisation_logo = (
+
1617 db_organisation.name,
+
1618 db_organisation.logo,
+
+
1620 project.total_contributors = contributors
+
+
+
+
+
1625 async
def get_project_users(db: Session, project_id: int):
+
1626 """Get the users and their contributions for a project.
+
+
+
1629 db (Session): The database session.
+
1630 project_id (int): The ID of the project.
+
+
+
1633 List[Dict[str, Union[str, int]]]: A list of dictionaries containing
+
1634 the username and the number of contributions made by each user
+
1635 for the specified project.
+
+
+
1638 db.query(db_models.DbTaskHistory)
+
1639 .filter(db_models.DbTaskHistory.project_id == project_id)
+
+
+
+
1643 user.user_id
for user
in contributors
if user.user_id
is not None
+
+
+
+
1647 for user_id
in unique_user_ids:
+
1648 contributions = count_user_contributions(db, user_id, project_id)
+
1649 db_user = await user_crud.get_user(db, user_id)
+
1650 response.append({
"user": db_user.username,
"contributions": contributions})
+
+
1652 response = sorted(response, key=
lambda x: x[
"contributions"], reverse=
True)
+
+
+
+
1656 def count_user_contributions(db: Session, user_id: int, project_id: int) -> int:
+
1657 """Count contributions for a specific user.
+
+
+
1660 db (Session): The database session.
+
1661 user_id (int): The ID of the user.
+
1662 project_id (int): The ID of the project.
+
+
+
1665 int: The number of contributions made by the user for the specified
+
+
+
1668 contributions_count = (
+
1669 db.query(func.count(db_models.DbTaskHistory.user_id))
+
+
1671 db_models.DbTaskHistory.user_id == user_id,
+
1672 db_models.DbTaskHistory.project_id == project_id,
+
+
+
+
+
1677 return contributions_count
+
+
+
1680 async
def add_project_admin(
+
1681 db: Session, user: db_models.DbUser, project: db_models.DbProject
+
+
1683 """Adds a user as an admin to the specified organisation.
+
+
+
1686 db (Session): The database session.
+
1687 user (int): The ID of the user to be added as an admin.
+
1688 project (DbOrganisation): The Project model instance.
+
+
+
1691 Response: The HTTP response with status code 200.
+
+
1693 new_user_role = db_models.DbUserRoles(
+
+
1695 project_id=project.id,
+
1696 role=ProjectRole.PROJECT_MANAGER,
+
+
+
+
1700 project.roles.append(new_user_role)
+
+
+
1703 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¶