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

82.55

backend/api/v1/v1_visualization/values_functions.py

from collections import defaultdict

from django.db.models import Count, Avg, F, OuterRef, Subquery
from django.db.models.functions import TruncDate, TruncMonth, Substr

from api.v1.v1_data.models import FormData
from api.v1.v1_forms.models import QuestionOptions
from api.v1.v1_visualization.constants import AGG_FUNCS
from api.v1.v1_visualization.models import (
    MVAnswerDenormalized,
    MVLatestMonitoring,
    MVParentAggregates,
)
from api.v1.v1_visualization.functions import (
    get_base_monitoring_qs,
    get_monitoring_data_ids,
    format_month_label,
    format_month_group,
    format_date_group,
    fill_month_gaps,
    fill_date_gaps,
    apply_administration_filter,
    apply_parent_criteria_to_qs,
)


def _should_fill_gaps(params):
    """Only gap-fill when both from_date and to_date are provided."""
    return bool(
        params.get("from_date") and params.get("to_date")
    )


def _date_answer_sq(date_qname):
    """Subquery: the date-question answer_name for the same data_id.

    Used to bucket value/option rows by the month of a separate date
    question (rather than the submission's own created timestamp).
    """
    return MVAnswerDenormalized.objects.filter(
        data_id=OuterRef("data_id"),
        question_name=date_qname,
        answer_name__isnull=False,
    ).values("answer_name")[:1]


def _finalize_month(data, params):
    """Apply month gap-fill (when bounded) and return (data, labels)."""
    if _should_fill_gaps(params):
        data = fill_month_gaps(
            data, params["from_date"], params["to_date"]
        )
    return data, [d["label"] for d in data]


def _finalize_date(data, params):
    """Apply day gap-fill (when bounded) and return (data, labels)."""
    if _should_fill_gaps(params):
        data = fill_date_gaps(
            data, params["from_date"], params["to_date"]
        )
    return data, [d["label"] for d in data]


def _total_parents_in_scope(form, params):
    """Count all parent registrations in scope, respecting filters."""
    scope_form = form.parent if form.parent else form
    qs = FormData.objects.filter(
        form=scope_form,
        parent__isnull=True,
        is_pending=False,
        is_draft=False,
    )
    administration_id = params.get("administration_id")
    if administration_id:
        qs = apply_administration_filter(qs, administration_id)
    qs = apply_parent_criteria_to_qs(
        qs, True, params.get("parent_criteria"),
    )
    return qs.count()


def _count_no_info_parents(form, params, qualifying_ids):
    """Count datapoints in scope with no qualifying answer.

    For monitoring forms: counts parent registrations without any
    qualifying monitoring submission (gap in monitoring coverage).
    For registration forms: counts registrations that exist but have no
    answer for the question (field left blank / skipped).

    Respects administration_id and parent_criteria so the count
    reconciles with option counts under filtering (FR-3).
    """
    total = _total_parents_in_scope(form, params)
    return max(0, total - len(qualifying_ids))


# -- Count mode handler --

def handle_count_mode(form, params):
    """Handle count mode (no question_id)."""
    form_id = form.id
    monitoring = params.get("monitoring", "latest")
    group_by = params.get("group_by")
    value_type = params.get("value_type", "number")
    sum_by = params.get("sum_by")
    is_monitoring = form.parent is not None

    if is_monitoring and monitoring == "latest" \
            and sum_by == "parent_id":
        qs, is_latest, _ = get_base_monitoring_qs(
            form, form_id, params
        )
        count = qs.count()
        if value_type == "percentage":
            total = FormData.objects.filter(
                form=form.parent,
                parent__isnull=True,
                is_pending=False,
                is_draft=False,
            ).count()
            value = round(
                (count / total * 100), 2
            ) if total > 0 else 0
        else:
            value = count
        return (
            [{"value": value, "label": "Total"}],
            ["Total"],
        )

    qs, is_latest, _ = get_base_monitoring_qs(
        form, form_id, params
    )

    if not group_by:
        count = qs.count()
        if value_type == "percentage" and is_monitoring:
            total = FormData.objects.filter(
                form=form.parent,
                parent__isnull=True,
                is_pending=False,
                is_draft=False,
            ).count()
            value = round(
                (count / total * 100), 2
            ) if total > 0 else 0
        else:
            value = count
        return (
            [{"value": value, "label": "Total"}],
            ["Total"],
        )

    if group_by == "month":
        return _count_group_by_month(qs, is_latest, params)

    if group_by == "parent_id":
        return _count_group_by_parent(qs, is_latest)

    if group_by == "id":
        return _count_group_by_id(qs, is_latest)

    if group_by == "date":
        return _count_group_by_date(qs, is_latest, params)

    return [{"value": 0, "label": "Total"}], ["Total"]


def _count_group_by_month(qs, is_latest, params):
    """Count grouped by month."""
    date_qname = params.get("date_question_name")

    if is_latest:
        data_ids = get_monitoring_data_ids(qs, is_latest)
        if date_qname:
            answer_qs = MVAnswerDenormalized.objects.filter(
                data_id__in=data_ids,
                question_name=date_qname,
                answer_name__isnull=False,
            )
            results = answer_qs.annotate(
                year_month=Substr("answer_name", 1, 7),
            ).values("year_month").annotate(
                count=Count("data_id", distinct=True),
            ).order_by("year_month")
            data = [
                {
                    "value": r["count"],
                    "label": format_month_label(
                        r["year_month"]
                    ),
                    "group": r["year_month"],
                }
                for r in results
            ]
        else:
            results = MVAnswerDenormalized.objects.filter(
                data_id__in=data_ids,
            ).annotate(
                month=TruncMonth("data_created"),
            ).values("month").annotate(
                count=Count("data_id", distinct=True),
            ).order_by("month")
            data = [
                {
                    "value": r["count"],
                    "label": format_month_label(r["month"]),
                    "group": format_month_group(r["month"]),
                }
                for r in results
            ]
    else:
        if date_qname:
            answer_qs = MVAnswerDenormalized.objects.filter(
                data_id__in=qs.values("id"),
                question_name=date_qname,
                answer_name__isnull=False,
            )
            results = answer_qs.annotate(
                year_month=Substr("answer_name", 1, 7),
            ).values("year_month").annotate(
                count=Count("data_id", distinct=True),
            ).order_by("year_month")
            data = [
                {
                    "value": r["count"],
                    "label": format_month_label(
                        r["year_month"]
                    ),
                    "group": r["year_month"],
                }
                for r in results
            ]
        else:
            results = qs.annotate(
                month=TruncMonth("created"),
            ).values("month").annotate(
                count=Count("id"),
            ).order_by("month")
            data = [
                {
                    "value": r["count"],
                    "label": format_month_label(r["month"]),
                    "group": format_month_group(r["month"]),
                }
                for r in results
            ]

    return _finalize_month(data, params)


def _count_group_by_parent(qs, is_latest):
    """Count grouped by parent_id."""
    if is_latest:
        data = [
            {
                "value": 1,
                "label": p.name,
                "group": str(p.id),
            }
            for p in qs.only("id", "name")
        ]
    else:
        results = qs.filter(
            parent__isnull=False,
        ).values(
            "parent_id",
            parent_name=F("parent__name"),
        ).annotate(
            count=Count("id"),
        ).order_by("parent_name")
        data = [
            {
                "value": r["count"],
                "label": r["parent_name"],
                "group": str(r["parent_id"]),
            }
            for r in results
        ]
    labels = [d["label"] for d in data]
    return data, labels


def _count_group_by_id(qs, is_latest):
    """Count grouped by individual record id (value=1 per row)."""
    if is_latest:
        data = [
            {
                "value": 1,
                "label": p.name,
                "group": str(p.latest_id),
            }
            for p in qs.only("id", "name")
        ]
    else:
        data = [
            {
                "value": 1,
                "label": r.name,
                "group": str(r.id),
            }
            for r in qs.only("id", "name").order_by("id")
        ]
    labels = [d["label"] for d in data]
    return data, labels


def _count_group_by_date(qs, is_latest, params):
    """Count grouped by individual date (not month bucket)."""
    date_qname = params.get("date_question_name")
    data_ids = get_monitoring_data_ids(qs, is_latest)

    if date_qname:
        results = MVAnswerDenormalized.objects.filter(
            data_id__in=data_ids,
            question_name=date_qname,
            answer_name__isnull=False,
        ).annotate(
            day=Substr("answer_name", 1, 10),
        ).values("day").annotate(
            count=Count("data_id", distinct=True),
        ).order_by("day")
        data = [
            {
                "value": r["count"],
                "label": r["day"],
                "group": r["day"],
            }
            for r in results
        ]
    else:
        results = MVAnswerDenormalized.objects.filter(
            data_id__in=data_ids,
        ).annotate(
            day=TruncDate("data_created"),
        ).values("day").annotate(
            count=Count("data_id", distinct=True),
        ).order_by("day")
        data = [
            {
                "value": r["count"],
                "label": format_date_group(r["day"]),
                "group": format_date_group(r["day"]),
            }
            for r in results
        ]
    return _finalize_date(data, params)


# -- Option question handler --

def handle_option_question(form, question, params):
    """Handle option/multiple_option questions."""
    form_id = form.id
    group_by = params.get("group_by")
    option_value = params.get("option_value")
    sum_by = params.get("sum_by")
    value_type = params.get("value_type", "number")
    stack_by = params.get("stack_by")

    qs, is_latest, _ = get_base_monitoring_qs(
        form, form_id, params
    )
    data_ids = get_monitoring_data_ids(qs, is_latest)

    options = QuestionOptions.objects.filter(
        question=question,
    ).order_by("order")

    if option_value and group_by == "month":
        return _option_value_group_by_month(
            question, data_ids, option_value, sum_by, params
        )

    if option_value:
        return _option_value_filter(
            question, data_ids, qs, is_latest,
            option_value, sum_by, value_type,
            include_unanswered=params.get(
                "include_unanswered", False
            ),
            form=form,
            params=params,
            include_empty=params.get("include_empty", False),
        )

    if stack_by == "option" and group_by:
        return handle_stack_by_option(
            question, options, data_ids,
            qs, is_latest, params
        )

    if group_by == "option":
        restricted = _extract_criteria_option_values(
            params, question.name
        )
        return _option_group_by_option(
            question, options, data_ids, qs,
            is_latest, value_type, restricted,
            include_unanswered=params.get(
                "include_unanswered", False
            ),
            form=form,
            params=params,
        )

    return [], []


def _option_value_filter(
    question, data_ids, qs, is_latest,
    option_value, sum_by, value_type,
    include_unanswered=False, form=None, params=None,
    include_empty=False,
):
    """Filter by specific option value and count.

    include_unanswered=True: parents with no answer for the question
    (monitored but null options) are added to the count.

    include_empty=True: parents with zero monitoring submissions
    (never visited) are added to the count. Takes precedence over
    include_unanswered when both are set, as the coverage-gap count
    already subsumes the answer-gap count.
    """
    count = MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
        answer_options__contains=[option_value],
    )
    if sum_by == "parent_id":
        count = count.values(
            "parent_id"
        ).distinct().count()
    else:
        count = count.count()

    is_monitoring = form is not None and form.parent is not None
    extra = 0

    if include_empty and is_monitoring:
        monitored_parent_ids = set(
            FormData.objects.filter(id__in=data_ids)
            .values_list("parent_id", flat=True)
            .distinct()
        )
        extra = _count_no_info_parents(
            form, params or {}, monitored_parent_ids
        )
    elif include_unanswered and is_monitoring:
        all_answered_ids = set(
            MVAnswerDenormalized.objects.filter(
                data_id__in=data_ids,
                question_name=question.name,
                answer_options__isnull=False,
            ).values_list("parent_id", flat=True).distinct()
        )
        extra = _count_no_info_parents(
            form, params or {}, all_answered_ids
        )

    if value_type == "percentage":
        if (include_empty or include_unanswered) and is_monitoring:
            total = _total_parents_in_scope(form, params or {})
            numerator = count + extra
        else:
            total = qs.count() if is_latest else len(data_ids)
            numerator = count
        value = round(
            (numerator / total * 100), 2
        ) if total > 0 else 0
    else:
        value = count + extra

    return (
        [{"value": value, "label": option_value}],
        [option_value],
    )


def _option_value_group_by_month(
    question, data_ids, option_value, sum_by, params
):
    """Filter by option_value, then bucket by month.

    Used by charts like "Proposed completion date": filter to
    incomplete projects (option_value='no') and bucket the count
    by a date question (e.g. project deadline). When `sum_by` is
    `parent_id`, counts distinct parents per month.
    """
    date_qname = params.get("date_question_name")

    matching_ids = list(MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
        answer_options__contains=[option_value],
    ).values_list("data_id", flat=True))

    if not matching_ids:
        data = []
    elif date_qname:
        answer_qs = MVAnswerDenormalized.objects.filter(
            data_id__in=matching_ids,
            question_name=date_qname,
            answer_name__isnull=False,
        )
        if sum_by == "parent_id":
            answer_qs = answer_qs.annotate(
                year_month=Substr("answer_name", 1, 7),
            ).values("year_month").annotate(
                count=Count("parent_id", distinct=True),
            ).order_by("year_month")
        else:
            answer_qs = answer_qs.annotate(
                year_month=Substr("answer_name", 1, 7),
            ).values("year_month").annotate(
                count=Count("data_id", distinct=True),
            ).order_by("year_month")
        data = [
            {
                "value": r["count"],
                "label": format_month_label(
                    r["year_month"]
                ),
                "group": r["year_month"],
            }
            for r in answer_qs
        ]
    else:
        mv_qs = MVAnswerDenormalized.objects.filter(
            data_id__in=matching_ids,
        ).annotate(
            month=TruncMonth("data_created"),
        ).values("month")
        if sum_by == "parent_id":
            mv_qs = mv_qs.annotate(
                count=Count("parent_id", distinct=True),
            ).order_by("month")
        else:
            mv_qs = mv_qs.annotate(
                count=Count("data_id", distinct=True),
            ).order_by("month")
        data = [
            {
                "value": r["count"],
                "label": format_month_label(r["month"]),
                "group": format_month_group(r["month"]),
            }
            for r in mv_qs
        ]

    return _finalize_month(data, params)


def _extract_criteria_option_values(params, question_name):
    """Extract option values that criteria restricts for a given qid.

    When criteria includes option_equals/option_contains/option_in
    targeting the same question_id as the donut chart, the tally
    should only count those specific values — not every value in
    a multiple_option answer array. Returns None if no restriction.
    """
    all_criteria = list(params.get("criteria") or [])
    all_criteria.extend(params.get("parent_criteria") or [])
    values = set()
    for c in all_criteria:
        ctype = c["type"]
        parts = c["parts"]
        if parts[0] != question_name:
            continue
        if ctype in ("option_equals", "option_contains"):
            values.add(parts[1])
        elif ctype == "option_in":
            values.update(parts[1])
    return values or None


def _option_group_by_option(
    question, options, data_ids, qs,
    is_latest, value_type, restricted_values=None,
    include_unanswered=False, form=None, params=None,
):
    """Group by option values (donut chart).

    Returns a row for every defined option — including zero-count
    options — so pie/doughnut charts have stable legends and colors
    across refreshes and filter changes.

    When `restricted_values` is set (from a criteria filter on the
    same question), only those values are tallied — so a
    multiple_option record ["a", "b"] filtered by "a" counts only
    for "a", not "b".

    When `include_unanswered=True`, appends one synthetic row
    (group="_no_info") for parents with no qualifying answer,
    and adjusts the percentage denominator to include the bucket
    so single-choice rows sum to 100%.
    """
    option_values = {o.value for o in options}
    tally_values = (
        option_values & restricted_values
        if restricted_values else option_values
    )
    tallies = defaultdict(int)
    qualifying_parents = set()
    # Registration forms have no parent; track data_id directly.
    # Monitoring forms track data__parent_id (the registration ID).
    is_registration = form is not None and form.parent is None
    tracking_field = (
        "data_id" if is_registration else "parent_id"
    )
    for tracking_id, opts in MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
        answer_options__isnull=False,
    ).values_list(tracking_field, "answer_options"):
        matched = False
        for v in (opts or []):
            if v in tally_values:
                tallies[v] += 1
                matched = True
        if matched:
            qualifying_parents.add(tracking_id)

    counts = [tallies.get(opt.value, 0) for opt in options]

    bucket_count = (
        _count_no_info_parents(form, params, qualifying_parents)
        if include_unanswered else 0
    )

    if value_type == "percentage":
        if include_unanswered:
            denom = len(qualifying_parents) + bucket_count
        else:
            denom = sum(counts)
    else:
        denom = None

    data = []
    for opt, count in zip(options, counts):
        val = (
            round((count / denom * 100), 2)
            if value_type == "percentage" and denom else count
        )
        data.append({
            "value": val,
            "label": opt.label,
            "group": opt.value,
            "color": opt.color,
        })

    if include_unanswered and bucket_count > 0:
        bucket_val = (
            round((bucket_count / denom * 100), 2)
            if value_type == "percentage" and denom else bucket_count
        )
        data.append({
            "value": bucket_val,
            "label": "No information available",
            "group": "_no_info",
            "color": "#bfbfbf",
        })

    labels = [d["label"] for d in data]
    return data, labels


# -- Number question handler --

def handle_number_question(form, question, params):
    """Handle number questions."""
    form_id = form.id
    group_by = params.get("group_by")
    repeat_agg = params.get("repeat_agg", "average")
    value_type = params.get("value_type", "number")
    stack_by = params.get("stack_by")

    qs, is_latest, _ = get_base_monitoring_qs(
        form, form_id, params
    )
    data_ids = get_monitoring_data_ids(qs, is_latest)
    agg_func = AGG_FUNCS.get(repeat_agg, Avg)

    if stack_by == "parent_id":
        return handle_stack_by_parent(
            question, qs, is_latest,
            data_ids, params
        )

    if group_by == "parent_id":
        return _number_group_by_parent(
            question, data_ids, agg_func, value_type
        )

    if group_by == "date":
        return _number_group_by_date(
            question, data_ids, params
        )

    if group_by == "month":
        return _number_group_by_month(
            question, data_ids, agg_func, value_type, params
        )

    result = MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
        answer_value__isnull=False,
    ).aggregate(agg_value=agg_func("answer_value"))

    value = (
        round(result["agg_value"], 2)
        if result["agg_value"] else 0
    )
    return [{"value": value, "label": "Total"}], ["Total"]


def _number_group_by_parent(
    question, data_ids, agg_func, value_type
):
    """Number question grouped by parent_id."""
    agg_rows = list(
        MVAnswerDenormalized.objects.filter(
            data_id__in=data_ids,
            question_name=question.name,
            answer_value__isnull=False,
        ).values("parent_id").annotate(
            agg_value=agg_func("answer_value"),
        )
    )

    # Fetch parent names from mv_latest_monitoring (parent_name is
    # pre-joined from the registration FormData)
    parent_ids = [r["parent_id"] for r in agg_rows if r["parent_id"]]
    name_map = dict(
        MVLatestMonitoring.objects.filter(parent_id__in=parent_ids)
        .values_list("parent_id", "parent_name")
        .distinct()
    )

    data = sorted(
        [
            {
                "value": round(r["agg_value"], 2),
                "label": name_map.get(r["parent_id"], ""),
                "group": str(r["parent_id"]),
            }
            for r in agg_rows
            if r["parent_id"]
        ],
        key=lambda x: x["label"],
    )

    if value_type == "percentage":
        total = sum(d["value"] for d in data)
        if total > 0:
            for d in data:
                d["value"] = round(
                    d["value"] / total * 100, 2
                )

    labels = [d["label"] for d in data]
    return data, labels


def _number_group_by_date(question, data_ids, params):
    """Number question grouped by date."""
    repeat_agg = params.get("repeat_agg", "average")
    agg_func = AGG_FUNCS.get(repeat_agg, Avg)
    date_qname = params.get("date_question_name")

    if date_qname:
        data = []
        for data_id in data_ids:
            date_answer = MVAnswerDenormalized.objects.filter(
                data_id=data_id,
                question_name=date_qname,
            ).first()
            if not date_answer or not date_answer.answer_name:
                continue
            val_result = MVAnswerDenormalized.objects.filter(
                data_id=data_id,
                question_name=question.name,
                answer_value__isnull=False,
            ).aggregate(agg_value=agg_func("answer_value"))
            if val_result["agg_value"] is not None:
                date_str = format_date_group(
                    date_answer.answer_name
                )
                data.append({
                    "value": round(
                        val_result["agg_value"], 2
                    ),
                    "label": date_str,
                    "group": date_str,
                })
    else:
        results = MVAnswerDenormalized.objects.filter(
            data_id__in=data_ids,
            question_name=question.name,
            answer_value__isnull=False,
        ).annotate(
            date=TruncDate("data_created"),
        ).values("date").annotate(
            agg_value=agg_func("answer_value"),
        ).order_by("date")
        data = [
            {
                "value": round(r["agg_value"], 2),
                "label": format_date_group(r["date"]),
                "group": format_date_group(r["date"]),
            }
            for r in results
        ]

    data.sort(key=lambda x: x["group"])
    return _finalize_date(data, params)


def _number_group_by_month(
    question, data_ids, agg_func, value_type, params
):
    """Number question grouped by month.

    When date_question_id is provided, bucket by the month of that
    date answer (via a Subquery) instead of FormData.created so the
    x-axis aligns with the filter's date dimension.
    """
    date_qname = params.get("date_question_name")

    base = MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
        answer_value__isnull=False,
    )

    if date_qname:
        date_sq = _date_answer_sq(date_qname)
        results = base.annotate(
            date_name=Subquery(date_sq),
        ).filter(
            date_name__isnull=False,
        ).annotate(
            month_key=Substr("date_name", 1, 7),
        ).values("month_key").annotate(
            agg_value=agg_func("answer_value"),
        ).order_by("month_key")
        data = [
            {
                "value": round(r["agg_value"], 2),
                "label": format_month_label(r["month_key"]),
                "group": r["month_key"],
            }
            for r in results if r["agg_value"] is not None
        ]
    else:
        results = base.annotate(
            month=TruncMonth("data_created"),
        ).values("month").annotate(
            agg_value=agg_func("answer_value"),
        ).order_by("month")
        data = [
            {
                "value": round(r["agg_value"], 2),
                "label": format_month_label(r["month"]),
                "group": format_month_group(r["month"]),
            }
            for r in results
        ]

    if value_type == "percentage":
        total = sum(d["value"] for d in data)
        if total > 0:
            for d in data:
                d["value"] = round(
                    d["value"] / total * 100, 2
                )

    return _finalize_month(data, params)


# -- Stack handlers --

def handle_stack_by_option(
    question, options, data_ids,
    qs, is_latest, params
):
    """Handle stack_by=option: stacked bar charts."""
    group_by = params.get("group_by")
    value_type = params.get("value_type", "number")

    opt_labels = [o.label for o in options]
    opt_colors = [o.color for o in options]

    if group_by == "month":
        return _stack_option_by_month(
            question, options, data_ids,
            opt_labels, opt_colors, value_type, params
        )

    if group_by == "parent_id":
        return _stack_option_by_parent(
            question, options, data_ids,
            qs, is_latest, opt_labels, opt_colors
        )

    return {
        "data": [], "labels": [],
        "stack_labels": [], "colors": [],
    }


def _stack_option_by_month(
    question, options, data_ids,
    opt_labels, opt_colors, value_type, params
):
    """Stack by option, grouped by month.

    Fetches answers once and buckets in Python — O(N) instead of
    O(months × options) queries. Honors date_question_id when
    provided so the month bucket aligns with the filter dimension.
    """
    date_qname = params.get("date_question_name")
    option_values = {o.value for o in options}

    base = MVAnswerDenormalized.objects.filter(
        data_id__in=data_ids,
        question_name=question.name,
    )

    if date_qname:
        date_sq = _date_answer_sq(date_qname)
        rows = base.annotate(
            date_name=Subquery(date_sq),
        ).filter(
            date_name__isnull=False,
        ).annotate(
            month_key=Substr("date_name", 1, 7),
        ).values("month_key", "answer_options")
        get_key = lambda r: r["month_key"]  # noqa: E731
        get_label = lambda k: format_month_label(k)  # noqa: E731
    else:
        rows = base.annotate(
            month=TruncMonth("data_created"),
        ).values("month", "answer_options")
        get_key = lambda r: format_month_group(r["month"])  # noqa: E731
        get_label = lambda k: format_month_label(k)  # noqa: E731

    buckets = defaultdict(lambda: defaultdict(int))
    for r in rows:
        key = get_key(r)
        if not key:
            continue
        for v in (r["answer_options"] or []):
            if v in option_values:
                buckets[key][v] += 1

    data = []
    for key in sorted(buckets.keys()):
        row = {"group": key, "label": get_label(key)}
        total_in_month = 0
        for opt in options:
            count = buckets[key].get(opt.value, 0)
            row[opt.label] = count
            total_in_month += count
        if value_type == "percentage" and total_in_month > 0:
            for opt in options:
                row[opt.label] = round(
                    row[opt.label] / total_in_month * 100, 2,
                )
        data.append(row)

    labels = [d["label"] for d in data]
    return {
        "data": data,
        "labels": labels,
        "stack_labels": opt_labels,
        "colors": opt_colors,
    }


def _stack_option_by_parent_from_mv(
    agg_data, parent_ids, qs, options, opt_labels, opt_colors
):
    """Build stack data from mv_parent_aggregates.

    Single query — O(1) instead of O(P × M) queries.
    """
    parent_options = {
        row['parent_id']: row['option_values'] or []
        for row in agg_data
    }
    parent_names = {p.id: p.name for p in qs.only('id', 'name')}

    data = []
    for parent_id in parent_ids:
        opts = parent_options.get(parent_id, [])
        row = {"label": parent_names.get(parent_id, ""), "group": parent_id}
        for opt in options:
            row[opt.label] = opts.count(opt.value)
        data.append(row)

    return {
        "data": data,
        "labels": [d["label"] for d in data],
        "stack_labels": opt_labels,
        "colors": opt_colors,
    }


def _stack_option_by_parent_legacy(
    question, options, data_ids,
    qs, is_latest, opt_labels, opt_colors
):
    """Original _stack_option_by_parent implementation.

    Used as fallback when MV is not available or empty.

    Handles three data shapes:
      - is_latest=True: qs rows are parent FormData with a `latest_id`
        annotation pointing to each parent's most-recent monitoring
        submission. Answer counts are read from that single submission.
      - is_latest=False, monitoring-form query: data_ids reference
        monitoring submissions; parents are derived via their parent_id.
        Answer counts aggregate all matching submissions per parent.
      - is_latest=False, REGISTRATION-form query (akvo-mis-9d8): data_ids
        ARE registration submissions themselves (parent__isnull=True).
        Parents = qs directly; p_data_ids = [parent.id].
    """
    is_registration_form = False
    if is_latest:
        parents = qs  # FormData qs with .latest_id and .name
    else:
        parent_ids = list(
            MVAnswerDenormalized.objects.filter(
                data_id__in=data_ids,
                parent_id__isnull=False,
            ).values_list("parent_id", flat=True).distinct()
        )
        if parent_ids:
            name_map = dict(
                MVLatestMonitoring.objects.filter(
                    parent_id__in=parent_ids
                ).values_list("parent_id", "parent_name").distinct()
            )
            parents = [
                {"id": pid, "name": name_map.get(pid, "")}
                for pid in parent_ids
            ]
        else:
            parents = qs  # registration-form path
            is_registration_form = True

    data = []
    for parent in parents:
        if is_latest:
            p_data_ids = [parent.latest_id]
            p_name = parent.name
            parent_id_val = parent.id
        elif is_registration_form:
            p_data_ids = [parent.id]
            p_name = parent.name
            parent_id_val = parent.id
        else:
            # parent is a dict {"id": ..., "name": ...}
            p_data_ids = list(
                MVAnswerDenormalized.objects.filter(
                    data_id__in=data_ids,
                    parent_id=parent["id"],
                ).values_list("data_id", flat=True).distinct()
            )
            p_name = parent["name"]
            parent_id_val = parent["id"]

        row = {"label": p_name, "group": parent_id_val}
        for opt in options:
            count = MVAnswerDenormalized.objects.filter(
                data_id__in=p_data_ids,
                question_name=question.name,
                answer_options__contains=[opt.value],
            ).count()
            row[opt.label] = count
        data.append(row)

    labels = [d["label"] for d in data]
    return {
        "data": data,
        "labels": labels,
        "stack_labels": opt_labels,
        "colors": opt_colors,
    }


def _stack_option_by_parent(
    question, options, data_ids,
    qs, is_latest, opt_labels, opt_colors
):
    """Stack by option, grouped by parent_id.

    OPTIMIZED: Uses mv_parent_aggregates when is_latest=True to replace
    the N+1 query pattern (P parents × M options) with a single MV lookup.
    Falls back to _stack_option_by_parent_legacy when MV has no data or
    when is_latest=False (all-submissions path, not covered by the MV).
    """
    if is_latest and data_ids:
        first_data = (
            MVAnswerDenormalized.objects
            .filter(data_id__in=data_ids[:1])
            .values('form_id')
            .first()
        )
        if first_data:
            form_id = first_data['form_id']
            parent_ids = list(qs.values_list('id', flat=True))
            agg_data = list(
                MVParentAggregates.objects.filter(
                    form_id=form_id,
                    question_name=question.name,
                    parent_id__in=parent_ids,
                ).values('parent_id', 'option_values')
            )
            if agg_data:
                return _stack_option_by_parent_from_mv(
                    agg_data, parent_ids, qs, options, opt_labels, opt_colors
                )

    return _stack_option_by_parent_legacy(
        question, options, data_ids, qs, is_latest, opt_labels, opt_colors
    )


def handle_stack_by_parent(
    question, qs, is_latest, data_ids, params
):
    """Handle stack_by=parent_id: multi-line charts."""
    group_by = params.get("group_by")
    repeat_agg = params.get("repeat_agg", "average")
    agg_func = AGG_FUNCS.get(repeat_agg, Avg)

    if is_latest:
        parents = list(
            qs.values("id", "name", "latest_id")
        )
    else:
        parent_id_list = list(
            MVAnswerDenormalized.objects.filter(
                data_id__in=data_ids,
                parent_id__isnull=False,
            ).values_list("parent_id", flat=True).distinct()
        )
        name_map = dict(
            MVLatestMonitoring.objects.filter(
                parent_id__in=parent_id_list
            ).values_list("parent_id", "parent_name").distinct()
        )
        parents = [
            {
                "id": pid,
                "name": name_map.get(pid, ""),
                "data_ids": list(
                    MVAnswerDenormalized.objects.filter(
                        data_id__in=data_ids,
                        parent_id=pid,
                    ).values_list("data_id", flat=True).distinct()
                ),
            }
            for pid in parent_id_list
        ]

    parent_names = [p["name"] for p in parents]

    if group_by == "date":
        return _stack_parent_by_date(
            question, parents, is_latest,
            parent_names, agg_func, params
        )

    if group_by == "month":
        return _stack_parent_by_month(
            question, parents, is_latest,
            parent_names, agg_func, params
        )

    return {"data": [], "labels": [], "stack_labels": []}


def _stack_parent_by_date(
    question, parents, is_latest,
    parent_names, agg_func, params
):
    """Stack by parent_id, grouped by date.

    Prefetches date keys and aggregated values per data_id in two
    bulk queries instead of N+1 per-point queries.
    """
    date_qname = params.get("date_question_name")

    all_data_ids = []
    for p in parents:
        if is_latest:
            all_data_ids.append(p["latest_id"])
        else:
            all_data_ids.extend(p["data_ids"])

    if date_qname:
        date_rows = MVAnswerDenormalized.objects.filter(
            data_id__in=all_data_ids,
            question_name=date_qname,
            answer_name__isnull=False,
        ).values("data_id", "answer_name")
        date_map = {
            r["data_id"]: format_date_group(r["answer_name"])
            for r in date_rows
        }
    else:
        mv_rows = MVAnswerDenormalized.objects.filter(
            data_id__in=all_data_ids,
        ).values("data_id", "data_created").distinct()
        date_map = {
            r["data_id"]: format_date_group(r["data_created"])
            for r in mv_rows
        }

    val_rows = MVAnswerDenormalized.objects.filter(
        data_id__in=all_data_ids,
        question_name=question.name,
        answer_value__isnull=False,
    ).values("data_id").annotate(
        agg_value=agg_func("answer_value"),
    )
    val_map = {
        r["data_id"]: r["agg_value"]
        for r in val_rows
        if r["agg_value"] is not None
    }

    all_rows = {}
    for p in parents:
        p_ids = (
            [p["latest_id"]] if is_latest
            else p["data_ids"]
        )
        for data_id in p_ids:
            date_key = date_map.get(data_id)
            agg_val = val_map.get(data_id)
            if not date_key or agg_val is None:
                continue
            if date_key not in all_rows:
                all_rows[date_key] = {"date": date_key}
            all_rows[date_key][p["name"]] = round(agg_val, 2)

    data = [all_rows[k] for k in sorted(all_rows.keys())]
    labels = sorted(all_rows.keys())
    return {
        "data": data,
        "labels": labels,
        "stack_labels": parent_names,
    }


def _stack_parent_by_month(
    question, parents, is_latest,
    parent_names, agg_func, params
):
    """Stack by parent_id, grouped by month.

    When date_question_id is provided, buckets by the month of that
    date answer (via Subquery) instead of FormData.created.
    """
    date_qname = params.get("date_question_name")
    all_rows = {}

    for p in parents:
        p_ids = (
            [p["latest_id"]] if is_latest
            else p["data_ids"]
        )

        base = MVAnswerDenormalized.objects.filter(
            data_id__in=p_ids,
            question_name=question.name,
            answer_value__isnull=False,
        )

        if date_qname:
            date_sq = _date_answer_sq(date_qname)
            results = base.annotate(
                date_name=Subquery(date_sq),
            ).filter(
                date_name__isnull=False,
            ).annotate(
                month_key=Substr("date_name", 1, 7),
            ).values("month_key").annotate(
                agg_value=agg_func("answer_value"),
            ).order_by("month_key")
            for r in results:
                if r["agg_value"] is None:
                    continue
                month_key = r["month_key"]
                if month_key not in all_rows:
                    all_rows[month_key] = {
                        "month": format_month_label(month_key),
                    }
                all_rows[month_key][p["name"]] = round(
                    r["agg_value"], 2,
                )
        else:
            results = base.annotate(
                month=TruncMonth("data_created"),
            ).values("month").annotate(
                agg_value=agg_func("answer_value"),
            ).order_by("month")
            for r in results:
                month_key = format_month_group(r["month"])
                if month_key not in all_rows:
                    all_rows[month_key] = {
                        "month": format_month_label(
                            r["month"]
                        ),
                    }
                all_rows[month_key][p["name"]] = round(
                    r["agg_value"], 2,
                )

    data = [all_rows[k] for k in sorted(all_rows.keys())]
    labels = [d["month"] for d in data]
    return {
        "data": data,
        "labels": labels,
        "stack_labels": parent_names,
    }

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