akvo / iwsims / #59

Committed 18 Jun 2026 07:20AM UTC coverage: 88.033% (-0.1%) from 88.13%

Build # #59

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 5dfcb298b into a6f6761c9

Coverage Stats

5183 of 6053 branches covered (85.63%)

Branch coverage included in aggregate %.

9979 of 11170 relevant lines covered (89.34%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.47

backend/api/v1/v1_visualization/views.py

import json
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework import status
from datetime import datetime
from django.db.models import Q
from api.v1.v1_data.models import FormData, Answers
from api.v1.v1_forms.models import Forms, QuestionTypes
from api.v1.v1_visualization.serializers import (
    MonitoringStatSerializer,
    GeoLocationListSerializer,
    GeoLocationFilterSerializer,
    FormDataStatSerializer,
    FormDataStatsFilterSerializer,
    FormulaValuesSerializer,
    DatapointDetailSerializer,
)
from api.v1.v1_visualization.models import (
    ViewDataOptions,
    MVAnswerDenormalized,
)
from api.v1.v1_visualization.functions import (
    apply_criteria_to_monitoring_qs,
)
from api.v1.v1_visualization.formula import (
    evaluate as formula_evaluate,
    pick_latest_repeat,
)
from drf_spectacular.utils import extend_schema, OpenApiParameter
from drf_spectacular.types import OpenApiTypes
from rest_framework.generics import get_object_or_404
from rest_framework.views import APIView
# from rest_framework.permissions import IsAuthenticated
from utils.custom_serializer_fields import validate_serializers_message


@extend_schema(
    description=(
        "Get the statistics of form data based on"
        "a specific monitoring form ID and question ID."
    ),
    tags=["Visualization"],
    responses=FormDataStatSerializer(many=True),
    parameters=[
        OpenApiParameter(
            name="question_id",
            required=True,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
            description="The question ID to extract the value from",
        ),
    ],
)
@api_view(["GET"])
def formdata_stats(request, form_id, version):
    form = get_object_or_404(Forms, pk=form_id)
    serializer = FormDataStatsFilterSerializer(
        data=request.GET,
        context={"form": form}
    )
    if not serializer.is_valid():
        return Response(
            {"message": validate_serializers_message(serializer.errors)},
            status=status.HTTP_400_BAD_REQUEST,
        )
    question = serializer.validated_data.get("question_id")
    options = []
    if not form.parent:
        if question.type in [
            QuestionTypes.option,
            QuestionTypes.multiple_option,
        ]:
            options = question.options.all()
        data = []
        for d in form.form_form_data.filter(
            is_pending=False,
            is_draft=False,
        ).all():
            if question.type == QuestionTypes.number:
                data.extend([
                    {
                        "id": d.id,
                        "value": a.value,
                    }
                    for a in d.data_answer.filter(
                        question_id=question.id
                    ).all()
                ])
            if question.type in [
                QuestionTypes.option,
                QuestionTypes.multiple_option,
            ]:
                for a in d.data_answer.filter(
                    question_id=question.id
                ).all():
                    for v in a.options:
                        v_data = question.options.filter(value=v).first()
                        if v_data:
                            data.append({
                                "id": d.id,
                                "value": v_data.id,
                            })
        return Response(
            FormDataStatSerializer(
                instance={
                    "options": options,
                    "data": data,
                }
            ).data,
            status=status.HTTP_200_OK,
        )
    if question.type == QuestionTypes.number:
        parent_form = form.parent
        form_data = parent_form.form_form_data.filter(
            is_pending=False,
            is_draft=False,
        ).all()
        data = [
            {
                "id": fd.id,
                "value": a.value,
            }
            for fd in form_data
            for ld in [fd.children.filter(
                form_id=form_id,
                is_pending=False,
                is_draft=False,
            ).last()] if ld
            for a in ld.data_answer.filter(
                question_id=question.id
            ).all()
        ]
        return Response(
            FormDataStatSerializer(
                instance={
                    "options": options,
                    "data": data,
                }
            ).data,
            status=status.HTTP_200_OK,
        )
    if question.type in [
        QuestionTypes.option,
        QuestionTypes.multiple_option,
    ]:
        options = question.options.all()
    data_options = ViewDataOptions.objects.filter(
        form=form,
    ).all()
    data = [
        {
            "id": do.parent_data_id,
            "value": v,
            "question_id": int(o.split("||")[0]),
        }
        for do in data_options
        for o in do.options
        for v in json.loads(o.split("||")[1])
    ]
    # filter data based on the question_id
    data = list(filter(
        lambda x: x["question_id"] == question.id, data
    ))
    return Response(
        FormDataStatSerializer(
            instance={
                "options": options,
                "data": data,
            }
        ).data,
        status=status.HTTP_200_OK,
    )


@extend_schema(
    description="Get the statistic of on monitoring data",
    tags=["Visualization"],
    responses=MonitoringStatSerializer(many=True),
    parameters=[
        OpenApiParameter(
            name="parent_id",
            required=True,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
            description="The parent ID to filter FormData",
        ),
        OpenApiParameter(
            name="question_id",
            required=True,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
            description="The question ID to extract the value from",
        ),
        OpenApiParameter(
            name="question_date",
            required=False,
            type=OpenApiTypes.NUMBER,
            location=OpenApiParameter.QUERY,
            description="the question to extract the date from (optional)",
        ),
    ],
)
@api_view(["GET"])
def monitoring_stats(request, version):
    parent_id = request.query_params.get("parent_id")
    question_id = request.query_params.get("question_id")
    question_date_key = request.query_params.get("question_date")

    if not parent_id or not question_id:
        return Response(
            {"detail": "Missing required parameters."},
            status=status.HTTP_400_BAD_REQUEST,
        )

    try:
        formdata_qs = FormData.objects.filter(parent_id=parent_id)
        stats = []

        for formdata in formdata_qs:
            answer = Answers.objects.filter(
                data=formdata, question_id=question_id
            ).first()
            if not answer:
                continue

            # Default date
            date = formdata.created

            # Optional override from another question
            if question_date_key:
                date_answer = Answers.objects.filter(
                    data=formdata, question_id=question_date_key
                ).first()
                if date_answer and date_answer.name:
                    parsed_date = datetime.strptime(
                        date_answer.name, "%Y-%m-%dT%H:%M:%S.%fZ"
                    )
                    if parsed_date:
                        date = parsed_date

            stats.append(
                {
                    "date": date.date(),
                    "value": answer.name or answer.value or answer.options,
                }
            )

        serializer = MonitoringStatSerializer(stats, many=True)
        return Response(serializer.data, status=status.HTTP_200_OK)

    except Exception as e:
        return Response(
            {"detail": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )


class GeolocationListView(APIView):
    # permission_classes = [IsAuthenticated]

    @extend_schema(
        responses=GeoLocationListSerializer,
        parameters=[
            OpenApiParameter(
                name="administration",
                required=False,
                type=OpenApiTypes.NUMBER,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="criteria",
                required=False,
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                description=(
                    "AND-joined multi-criteria filter "
                    "(same grammar as /values)."
                ),
            ),
            OpenApiParameter(
                name="from_date",
                required=False,
                type=OpenApiTypes.DATE,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="to_date",
                required=False,
                type=OpenApiTypes.DATE,
                location=OpenApiParameter.QUERY,
            ),
            OpenApiParameter(
                name="include_monitoring",
                required=False,
                type=OpenApiTypes.BOOL,
                location=OpenApiParameter.QUERY,
                description=(
                    "When true, from_date / to_date filter by the "
                    "datapoint's monitoring children's created date "
                    "instead of the datapoint's own created date."
                ),
            ),
            OpenApiParameter(
                name="monitoring_form_id",
                required=False,
                type=OpenApiTypes.NUMBER,
                location=OpenApiParameter.QUERY,
                description=(
                    "When include_monitoring=true, restrict the "
                    "children join to this form ID so that unrelated "
                    "child forms do not satisfy the date window."
                ),
            ),
        ],
        tags=["Maps"],
        summary="To get list of geolocations for a form",
    )
    def get(self, request, form_id, version):
        serializer = GeoLocationFilterSerializer(
            data=request.GET, context={"form_id": form_id}
        )
        if not serializer.is_valid():
            # Return empty list if serializer is not valid
            return Response(
                data=[],
                status=status.HTTP_200_OK,
            )
        form = get_object_or_404(Forms, pk=form_id)
        queryset = form.form_form_data.filter(
            is_pending=False,
            is_draft=False,
            geo__isnull=False
        )
        criteria = serializer.validated_data.get("criteria")
        if criteria:
            queryset = apply_criteria_to_monitoring_qs(
                queryset, False, criteria,
            )

        from_date = serializer.validated_data.get("from_date")
        to_date = serializer.validated_data.get("to_date")
        include_monitoring = serializer.validated_data.get(
            "include_monitoring", False
        )

        monitoring_form_id = serializer.validated_data.get(
            "monitoring_form_id"
        )
        if include_monitoring and (from_date or to_date):
            child_q = Q()
            if from_date:
                child_q &= Q(children__created__date__gte=from_date)
            if to_date:
                child_q &= Q(children__created__date__lte=to_date)
            child_filter = {
                "children__is_pending": False,
                "children__is_draft": False,
            }
            if monitoring_form_id:
                child_filter["children__form_id"] = monitoring_form_id
            queryset = queryset.filter(
                child_q, **child_filter
            ).distinct()
        else:
            if from_date:
                queryset = queryset.filter(created__date__gte=from_date)
            if to_date:
                queryset = queryset.filter(created__date__lte=to_date)

        if serializer.validated_data.get("administration"):
            adm = serializer.validated_data.get("administration")
            adm_path = f"{adm.id}."
            if adm.path:
                adm_path = f"{adm.path}{adm.id}."
            queryset = queryset.filter(
                Q(administration=adm) |
                Q(administration__path__startswith=adm_path)
            )
        if (
            request.user.is_authenticated and
            not request.user.is_superuser and
            not serializer.validated_data.get("administration")
        ):
            user_role = request.user.user_user_role.order_by(
                "administration__level__level"
            ).first()
            adm = user_role.administration if user_role else None
            if not adm:
                return Response(
                    data=[],
                    status=status.HTTP_200_OK,
                )
            adm_path = f"{adm.id}."
            if adm.path:
                adm_path = f"{adm.path}{adm.id}."
            queryset = queryset.filter(
                Q(administration=adm) |
                Q(administration__path__startswith=adm_path)
            )
        rows = list(
            queryset.values("id", "name", "geo", "administration_id")
        )
        serializer = GeoLocationListSerializer(rows, many=True)
        return Response(serializer.data, status=status.HTTP_200_OK)


class DatapointDetailView(APIView):
    # permission_classes = [IsAuthenticated]
    # public, same as GeolocationListView

    @extend_schema(
        responses=DatapointDetailSerializer,
        tags=["Maps"],
        summary="Lightweight datapoint metadata for the map popup",
        description=(
            "Returns name, administration_full_name, and updated for "
            "a single registration datapoint. Fetched on demand when "
            "the user clicks a map marker; results are cached client-side."
        ),
    )
    def get(self, request, data_id, version):
        point = get_object_or_404(
            FormData, pk=data_id,
            is_pending=False, is_draft=False, parent__isnull=True
        )
        return Response(
            DatapointDetailSerializer(instance=point).data,
            status=status.HTTP_200_OK,
        )


@extend_schema(
    description=(
        "Evaluate a formula against the latest monitoring child of "
        "each datapoint and group the resulting bucket value by "
        "parent_id. Same response shape as /visualization/values."
    ),
    tags=["Visualization"],
    parameters=[
        OpenApiParameter(
            name="form_id", required=True,
            type=OpenApiTypes.INT,
            location=OpenApiParameter.QUERY,
            description="The monitoring form id.",
        ),
        OpenApiParameter(
            name="group_by", required=True,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
            enum=["parent_id"],
        ),
        OpenApiParameter(
            name="monitoring", required=False,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
            enum=["latest"],
        ),
        OpenApiParameter(
            name="formula", required=True,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
            description=(
                "URL-encoded JSON formula. See "
                "doc/claude/filters-dashboard-mapview/design.md §1.2."
            ),
        ),
        OpenApiParameter(
            name="criteria", required=False,
            type=OpenApiTypes.STR,
            location=OpenApiParameter.QUERY,
            description=(
                "AND-joined multi-criteria filter "
                "(same grammar as /values)."
            ),
        ),
        OpenApiParameter(
            name="from_date", required=False,
            type=OpenApiTypes.DATE,
            location=OpenApiParameter.QUERY,
        ),
        OpenApiParameter(
            name="to_date", required=False,
            type=OpenApiTypes.DATE,
            location=OpenApiParameter.QUERY,
        ),
    ],
)
@api_view(["GET"])
def visualization_values_formula(request, version):
    """Evaluate a formula per datapoint.

    For monitoring forms (form.parent is set): groups by parent_id,
    using the latest monitoring child per registration datapoint.
    ``group`` in the response is the registration datapoint id.

    For registration forms (form.parent is None): evaluates the formula
    directly against each registration datapoint's answers.
    ``group`` in the response is the datapoint's own id.

    Both cases produce {group: datapoint_id, label: bucket_value} so
    the frontend's byParent[point.id] lookup works identically.
    """
    serializer = FormulaValuesSerializer(data=request.query_params)
    if not serializer.is_valid():
        return Response(
            {"message": validate_serializers_message(serializer.errors)},
            status=status.HTTP_400_BAD_REQUEST,
        )

    validated = serializer.validated_data
    form = get_object_or_404(Forms, pk=validated["form_id"])
    formula = validated["formula"]
    criteria = validated.get("criteria")
    from_date = validated.get("from_date")
    to_date = validated.get("to_date")

    is_registration = form.parent_id is None

    qs = form.form_form_data.filter(
        is_pending=False,
        is_draft=False,
        parent__isnull=is_registration,
    )
    if criteria:
        qs = apply_criteria_to_monitoring_qs(qs, False, criteria)
    if from_date:
        qs = qs.filter(created__date__gte=from_date)
    if to_date:
        qs = qs.filter(created__date__lte=to_date)

    if is_registration:
        # Each registration datapoint is its own "group".
        data_ids = list(qs.values_list("id", flat=True))
        if not data_ids:
            return Response({"data": []}, status=status.HTTP_200_OK)
        id_to_group = {d: d for d in data_ids}
    else:
        # Latest monitoring child per parent_id.
        monitorings = qs.order_by("parent_id", "-created").values(
            "id", "parent_id", "created"
        )
        latest_by_parent = {}
        for row in monitorings:
            parent_id = row["parent_id"]
            if parent_id not in latest_by_parent:
                latest_by_parent[parent_id] = row["id"]
        if not latest_by_parent:
            return Response({"data": []}, status=status.HTTP_200_OK)
        # Map data_id → group (parent_id) for response assembly.
        id_to_group = {v: k for k, v in latest_by_parent.items()}
        data_ids = list(id_to_group.keys())

    # Source answers from the denormalized MV so each row carries
    # question_name (the base Answers table has no name; filtering its
    # unindexed question__name would also be slower). Map MV columns to
    # the dict shape the formula evaluator expects (value/options/index).
    answers = MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids
    ).values(
        "data_id", "question_name",
        "answer_value", "answer_options", "answer_index",
    )

    answers_by_data = {}
    for ans in answers:
        answers_by_data.setdefault(ans["data_id"], []).append({
            "question_name": ans["question_name"],
            "value": ans["answer_value"],
            "options": ans["answer_options"],
            "index": ans["answer_index"],
        })

    data = []
    for data_id, group in id_to_group.items():
        per_question = pick_latest_repeat(answers_by_data.get(data_id, []))
        bucket = formula_evaluate(formula, per_question)
        data.append({"group": group, "label": bucket})

    return Response({"data": data}, status=status.HTTP_200_OK)

1	import json	1✔
2	from rest_framework.decorators import api_view	1✔
3	from rest_framework.response import Response	1✔
4	from rest_framework import status	1✔
5	from datetime import datetime	1✔
6	from django.db.models import Q	1✔
7	from api.v1.v1_data.models import FormData, Answers	1✔
8	from api.v1.v1_forms.models import Forms, QuestionTypes	1✔
9	from api.v1.v1_visualization.serializers import (	1✔
10	MonitoringStatSerializer,
11	GeoLocationListSerializer,
12	GeoLocationFilterSerializer,
13	FormDataStatSerializer,
14	FormDataStatsFilterSerializer,
15	FormulaValuesSerializer,
16	DatapointDetailSerializer,
17	)
18	from api.v1.v1_visualization.models import (	1✔
19	ViewDataOptions,
20	MVAnswerDenormalized,
21	)
22	from api.v1.v1_visualization.functions import (	1✔
23	apply_criteria_to_monitoring_qs,
24	)
25	from api.v1.v1_visualization.formula import (	1✔
26	evaluate as formula_evaluate,
27	pick_latest_repeat,
28	)
29	from drf_spectacular.utils import extend_schema, OpenApiParameter	1✔
30	from drf_spectacular.types import OpenApiTypes	1✔
31	from rest_framework.generics import get_object_or_404	1✔
32	from rest_framework.views import APIView	1✔
33	# from rest_framework.permissions import IsAuthenticated
34	from utils.custom_serializer_fields import validate_serializers_message	1✔
35
36
37	@extend_schema(	1✔
38	description=(
39	"Get the statistics of form data based on"
40	"a specific monitoring form ID and question ID."
41	),
42	tags=["Visualization"],
43	responses=FormDataStatSerializer(many=True),
44	parameters=[
45	OpenApiParameter(
46	name="question_id",
47	required=True,
48	type=OpenApiTypes.NUMBER,
49	location=OpenApiParameter.QUERY,
50	description="The question ID to extract the value from",
51	),
52	],
53	)
54	@api_view(["GET"])	1✔
55	def formdata_stats(request, form_id, version):	1✔
56	form = get_object_or_404(Forms, pk=form_id)	1✔
57	serializer = FormDataStatsFilterSerializer(	1✔
58	data=request.GET,
59	context={"form": form}
60	)
61	if not serializer.is_valid():	1✔
62	return Response(	1✔
63	{"message": validate_serializers_message(serializer.errors)},
64	status=status.HTTP_400_BAD_REQUEST,
65	)
66	question = serializer.validated_data.get("question_id")	1✔
67	options = []	1✔
68	if not form.parent:	1✔
69	if question.type in [	1✔
70	QuestionTypes.option,
71	QuestionTypes.multiple_option,
72	]:
73	options = question.options.all()	1✔
74	data = []	1✔
75	for d in form.form_form_data.filter(	1✔
76	is_pending=False,
77	is_draft=False,
78	).all():
79	if question.type == QuestionTypes.number:	1✔
80	data.extend([	1✔
81	{
82	"id": d.id,
83	"value": a.value,
84	}
85	for a in d.data_answer.filter(
86	question_id=question.id
87	).all()
88	])
89	if question.type in [	1✔
90	QuestionTypes.option,
91	QuestionTypes.multiple_option,
92	]:
93	for a in d.data_answer.filter(	1✔
94	question_id=question.id
95	).all():
96	for v in a.options:	1✔
97	v_data = question.options.filter(value=v).first()	1✔
98	if v_data:	1!
99	data.append({	1✔
100	"id": d.id,
101	"value": v_data.id,
102	})
103	return Response(	1✔
104	FormDataStatSerializer(
105	instance={
106	"options": options,
107	"data": data,
108	}
109	).data,
110	status=status.HTTP_200_OK,
111	)
112	if question.type == QuestionTypes.number:	1✔
113	parent_form = form.parent	1✔
114	form_data = parent_form.form_form_data.filter(	1✔
115	is_pending=False,
116	is_draft=False,
117	).all()
118	data = [	1✔
119	{
120	"id": fd.id,
121	"value": a.value,
122	}
123	for fd in form_data
124	for ld in [fd.children.filter(
125	form_id=form_id,
126	is_pending=False,
127	is_draft=False,
128	).last()] if ld
129	for a in ld.data_answer.filter(
130	question_id=question.id
131	).all()
132	]
133	return Response(	1✔
134	FormDataStatSerializer(
135	instance={
136	"options": options,
137	"data": data,
138	}
139	).data,
140	status=status.HTTP_200_OK,
141	)
142	if question.type in [	1!
143	QuestionTypes.option,
144	QuestionTypes.multiple_option,
145	]:
146	options = question.options.all()	1✔
147	data_options = ViewDataOptions.objects.filter(	1✔
148	form=form,
149	).all()
150	data = [	1✔
151	{
152	"id": do.parent_data_id,
153	"value": v,
154	"question_id": int(o.split("\|\|")[0]),
155	}
156	for do in data_options
157	for o in do.options
158	for v in json.loads(o.split("\|\|")[1])
159	]
160	# filter data based on the question_id
161	data = list(filter(	1✔
162	lambda x: x["question_id"] == question.id, data
163	))
164	return Response(	1✔
165	FormDataStatSerializer(
166	instance={
167	"options": options,
168	"data": data,
169	}
170	).data,
171	status=status.HTTP_200_OK,
172	)
173
174
175	@extend_schema(	1✔
176	description="Get the statistic of on monitoring data",
177	tags=["Visualization"],
178	responses=MonitoringStatSerializer(many=True),
179	parameters=[
180	OpenApiParameter(
181	name="parent_id",
182	required=True,
183	type=OpenApiTypes.NUMBER,
184	location=OpenApiParameter.QUERY,
185	description="The parent ID to filter FormData",
186	),
187	OpenApiParameter(
188	name="question_id",
189	required=True,
190	type=OpenApiTypes.NUMBER,
191	location=OpenApiParameter.QUERY,
192	description="The question ID to extract the value from",
193	),
194	OpenApiParameter(
195	name="question_date",
196	required=False,
197	type=OpenApiTypes.NUMBER,
198	location=OpenApiParameter.QUERY,
199	description="the question to extract the date from (optional)",
200	),
201	],
202	)
203	@api_view(["GET"])	1✔
204	def monitoring_stats(request, version):	1✔
205	parent_id = request.query_params.get("parent_id")	1✔
206	question_id = request.query_params.get("question_id")	1✔
207	question_date_key = request.query_params.get("question_date")	1✔
208
209	if not parent_id or not question_id:	1✔
210	return Response(	1✔
211	{"detail": "Missing required parameters."},
212	status=status.HTTP_400_BAD_REQUEST,
213	)
214
215	try:	1✔
216	formdata_qs = FormData.objects.filter(parent_id=parent_id)	1✔
217	stats = []	1✔
218
219	for formdata in formdata_qs:	1✔
220	answer = Answers.objects.filter(	1✔
221	data=formdata, question_id=question_id
222	).first()
223	if not answer:	1!
224	continue	×
225
226	# Default date
227	date = formdata.created	1✔
228
229	# Optional override from another question
230	if question_date_key:	1✔
231	date_answer = Answers.objects.filter(	1✔
232	data=formdata, question_id=question_date_key
233	).first()
234	if date_answer and date_answer.name:	1!
235	parsed_date = datetime.strptime(	1✔
236	date_answer.name, "%Y-%m-%dT%H:%M:%S.%fZ"
237	)
238	if parsed_date:	1!
239	date = parsed_date	1✔
240
241	stats.append(	1✔
242	{
243	"date": date.date(),
244	"value": answer.name or answer.value or answer.options,
245	}
246	)
247
248	serializer = MonitoringStatSerializer(stats, many=True)	1✔
249	return Response(serializer.data, status=status.HTTP_200_OK)	1✔
250
251	except Exception as e:	×
252	return Response(	×
253	{"detail": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
254	)
255
256
257	class GeolocationListView(APIView):	1✔
258	# permission_classes = [IsAuthenticated]
259
260	@extend_schema(	1✔
261	responses=GeoLocationListSerializer,
262	parameters=[
263	OpenApiParameter(
264	name="administration",
265	required=False,
266	type=OpenApiTypes.NUMBER,
267	location=OpenApiParameter.QUERY,
268	),
269	OpenApiParameter(
270	name="criteria",
271	required=False,
272	type=OpenApiTypes.STR,
273	location=OpenApiParameter.QUERY,
274	description=(
275	"AND-joined multi-criteria filter "
276	"(same grammar as /values)."
277	),
278	),
279	OpenApiParameter(
280	name="from_date",
281	required=False,
282	type=OpenApiTypes.DATE,
283	location=OpenApiParameter.QUERY,
284	),
285	OpenApiParameter(
286	name="to_date",
287	required=False,
288	type=OpenApiTypes.DATE,
289	location=OpenApiParameter.QUERY,
290	),
291	OpenApiParameter(
292	name="include_monitoring",
293	required=False,
294	type=OpenApiTypes.BOOL,
295	location=OpenApiParameter.QUERY,
296	description=(
297	"When true, from_date / to_date filter by the "
298	"datapoint's monitoring children's created date "
299	"instead of the datapoint's own created date."
300	),
301	),
302	OpenApiParameter(
303	name="monitoring_form_id",
304	required=False,
305	type=OpenApiTypes.NUMBER,
306	location=OpenApiParameter.QUERY,
307	description=(
308	"When include_monitoring=true, restrict the "
309	"children join to this form ID so that unrelated "
310	"child forms do not satisfy the date window."
311	),
312	),
313	],
314	tags=["Maps"],
315	summary="To get list of geolocations for a form",
316	)
317	def get(self, request, form_id, version):	1✔
318	serializer = GeoLocationFilterSerializer(	1✔
319	data=request.GET, context={"form_id": form_id}
320	)
321	if not serializer.is_valid():	1✔
322	# Return empty list if serializer is not valid
323	return Response(	1✔
324	data=[],
325	status=status.HTTP_200_OK,
326	)
327	form = get_object_or_404(Forms, pk=form_id)	1✔
328	queryset = form.form_form_data.filter(	1✔
329	is_pending=False,
330	is_draft=False,
331	geo__isnull=False
332	)
333	criteria = serializer.validated_data.get("criteria")	1✔
334	if criteria:	1✔
335	queryset = apply_criteria_to_monitoring_qs(	1✔
336	queryset, False, criteria,
337	)
338
339	from_date = serializer.validated_data.get("from_date")	1✔
340	to_date = serializer.validated_data.get("to_date")	1✔
341	include_monitoring = serializer.validated_data.get(	1✔
342	"include_monitoring", False
343	)
344
345	monitoring_form_id = serializer.validated_data.get(	1✔
346	"monitoring_form_id"
347	)
348	if include_monitoring and (from_date or to_date):	1✔
349	child_q = Q()	1✔
350	if from_date:	1!
351	child_q &= Q(children__created__date__gte=from_date)	1✔
352	if to_date:	1✔
353	child_q &= Q(children__created__date__lte=to_date)	1✔
354	child_filter = {	1✔
355	"children__is_pending": False,
356	"children__is_draft": False,
357	}
358	if monitoring_form_id:	1!
359	child_filter["children__form_id"] = monitoring_form_id	×
360	queryset = queryset.filter(	1✔
361	child_q, **child_filter
362	).distinct()
363	else:
364	if from_date:	1✔
365	queryset = queryset.filter(created__date__gte=from_date)	1✔
366	if to_date:	1✔
367	queryset = queryset.filter(created__date__lte=to_date)	1✔
368
369	if serializer.validated_data.get("administration"):	1✔
370	adm = serializer.validated_data.get("administration")	1✔
371	adm_path = f"{adm.id}."	1✔
372	if adm.path:	1!
373	adm_path = f"{adm.path}{adm.id}."	1✔
374	queryset = queryset.filter(	1✔
375	Q(administration=adm) \|
376	Q(administration__path__startswith=adm_path)
377	)
378	if (	1✔
379	request.user.is_authenticated and
380	not request.user.is_superuser and
381	not serializer.validated_data.get("administration")
382	):
383	user_role = request.user.user_user_role.order_by(	1✔
384	"administration__level__level"
385	).first()
386	adm = user_role.administration if user_role else None	1✔
387	if not adm:	1✔
388	return Response(	1✔
389	data=[],
390	status=status.HTTP_200_OK,
391	)
392	adm_path = f"{adm.id}."	1✔
393	if adm.path:	1!
394	adm_path = f"{adm.path}{adm.id}."	1✔
395	queryset = queryset.filter(	1✔
396	Q(administration=adm) \|
397	Q(administration__path__startswith=adm_path)
398	)
399	rows = list(	1✔
400	queryset.values("id", "name", "geo", "administration_id")
401	)
402	serializer = GeoLocationListSerializer(rows, many=True)	1✔
403	return Response(serializer.data, status=status.HTTP_200_OK)	1✔
404
405
406	class DatapointDetailView(APIView):	1✔
407	# permission_classes = [IsAuthenticated]
408	# public, same as GeolocationListView
409
410	@extend_schema(	1✔
411	responses=DatapointDetailSerializer,
412	tags=["Maps"],
413	summary="Lightweight datapoint metadata for the map popup",
414	description=(
415	"Returns name, administration_full_name, and updated for "
416	"a single registration datapoint. Fetched on demand when "
417	"the user clicks a map marker; results are cached client-side."
418	),
419	)
420	def get(self, request, data_id, version):	1✔
421	point = get_object_or_404(	1✔
422	FormData, pk=data_id,
423	is_pending=False, is_draft=False, parent__isnull=True
424	)
425	return Response(	1✔
426	DatapointDetailSerializer(instance=point).data,
427	status=status.HTTP_200_OK,
428	)
429
430
431	@extend_schema(	1✔
432	description=(
433	"Evaluate a formula against the latest monitoring child of "
434	"each datapoint and group the resulting bucket value by "
435	"parent_id. Same response shape as /visualization/values."
436	),
437	tags=["Visualization"],
438	parameters=[
439	OpenApiParameter(
440	name="form_id", required=True,
441	type=OpenApiTypes.INT,
442	location=OpenApiParameter.QUERY,
443	description="The monitoring form id.",
444	),
445	OpenApiParameter(
446	name="group_by", required=True,
447	type=OpenApiTypes.STR,
448	location=OpenApiParameter.QUERY,
449	enum=["parent_id"],
450	),
451	OpenApiParameter(
452	name="monitoring", required=False,
453	type=OpenApiTypes.STR,
454	location=OpenApiParameter.QUERY,
455	enum=["latest"],
456	),
457	OpenApiParameter(
458	name="formula", required=True,
459	type=OpenApiTypes.STR,
460	location=OpenApiParameter.QUERY,
461	description=(
462	"URL-encoded JSON formula. See "
463	"doc/claude/filters-dashboard-mapview/design.md §1.2."
464	),
465	),
466	OpenApiParameter(
467	name="criteria", required=False,
468	type=OpenApiTypes.STR,
469	location=OpenApiParameter.QUERY,
470	description=(
471	"AND-joined multi-criteria filter "
472	"(same grammar as /values)."
473	),
474	),
475	OpenApiParameter(
476	name="from_date", required=False,
477	type=OpenApiTypes.DATE,
478	location=OpenApiParameter.QUERY,
479	),
480	OpenApiParameter(
481	name="to_date", required=False,
482	type=OpenApiTypes.DATE,
483	location=OpenApiParameter.QUERY,
484	),
485	],
486	)
487	@api_view(["GET"])	1✔
488	def visualization_values_formula(request, version):	1✔
489	"""Evaluate a formula per datapoint.
490
491	For monitoring forms (form.parent is set): groups by parent_id,
492	using the latest monitoring child per registration datapoint.
493	``group`` in the response is the registration datapoint id.
494
495	For registration forms (form.parent is None): evaluates the formula
496	directly against each registration datapoint's answers.
497	``group`` in the response is the datapoint's own id.
498
499	Both cases produce {group: datapoint_id, label: bucket_value} so
500	the frontend's byParent[point.id] lookup works identically.
501	"""
502	serializer = FormulaValuesSerializer(data=request.query_params)	1✔
503	if not serializer.is_valid():	1✔
504	return Response(	1✔
505	{"message": validate_serializers_message(serializer.errors)},
506	status=status.HTTP_400_BAD_REQUEST,
507	)
508
509	validated = serializer.validated_data	1✔
510	form = get_object_or_404(Forms, pk=validated["form_id"])	1✔
511	formula = validated["formula"]	1✔
512	criteria = validated.get("criteria")	1✔
513	from_date = validated.get("from_date")	1✔
514	to_date = validated.get("to_date")	1✔
515
516	is_registration = form.parent_id is None	1✔
517
518	qs = form.form_form_data.filter(	1✔
519	is_pending=False,
520	is_draft=False,
521	parent__isnull=is_registration,
522	)
523	if criteria:	1✔
524	qs = apply_criteria_to_monitoring_qs(qs, False, criteria)	1✔
525	if from_date:	1!
526	qs = qs.filter(created__date__gte=from_date)	×
527	if to_date:	1!
528	qs = qs.filter(created__date__lte=to_date)	×
529
530	if is_registration:	1✔
531	# Each registration datapoint is its own "group".
532	data_ids = list(qs.values_list("id", flat=True))	1✔
533	if not data_ids:	1!
534	return Response({"data": []}, status=status.HTTP_200_OK)	×
535	id_to_group = {d: d for d in data_ids}	1✔
536	else:
537	# Latest monitoring child per parent_id.
538	monitorings = qs.order_by("parent_id", "-created").values(	1✔
539	"id", "parent_id", "created"
540	)
541	latest_by_parent = {}	1✔
542	for row in monitorings:	1✔
543	parent_id = row["parent_id"]	1✔
544	if parent_id not in latest_by_parent:	1✔
545	latest_by_parent[parent_id] = row["id"]	1✔
546	if not latest_by_parent:	1!
547	return Response({"data": []}, status=status.HTTP_200_OK)	×
548	# Map data_id → group (parent_id) for response assembly.
549	id_to_group = {v: k for k, v in latest_by_parent.items()}	1✔
550	data_ids = list(id_to_group.keys())	1✔
551
552	# Source answers from the denormalized MV so each row carries
553	# question_name (the base Answers table has no name; filtering its
554	# unindexed question__name would also be slower). Map MV columns to
555	# the dict shape the formula evaluator expects (value/options/index).
556	answers = MVAnswerDenormalized.objects.filter(	1✔
557	data_id__in=data_ids
558	).values(
559	"data_id", "question_name",
560	"answer_value", "answer_options", "answer_index",
561	)
562
563	answers_by_data = {}	1✔
564	for ans in answers:	1✔
565	answers_by_data.setdefault(ans["data_id"], []).append({	1✔
566	"question_name": ans["question_name"],
567	"value": ans["answer_value"],
568	"options": ans["answer_options"],
569	"index": ans["answer_index"],
570	})
571
572	data = []	1✔
573	for data_id, group in id_to_group.items():	1✔
574	per_question = pick_latest_repeat(answers_by_data.get(data_id, []))	1✔
575	bucket = formula_evaluate(formula, per_question)	1✔
576	data.append({"group": group, "label": bucket})	1✔
577
578	return Response({"data": data}, status=status.HTTP_200_OK)	1✔

akvo / iwsims / #59

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous