16070961563

Committed 04 Jul 2025 09:48AM UTC coverage: 94.796% (+4.0%) from 90.817%

Build # 16070961563

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1361 from rdmorganiser/2.3.2

RDMO 2.3.2 🐛🐛

Run Details

2086 of 2191 branches covered (95.21%)

22353 of 23580 relevant lines covered (94.8%)

3.79 hits per line

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

98.06

rdmo/projects/progress.py

from collections import defaultdict
from itertools import chain

from django.db.models import Exists, OuterRef, Q

from rdmo.conditions.models import Condition
from rdmo.core.utils import markdown2html
from rdmo.questions.models import Page, Question, QuestionSet


def get_catalog_conditions(catalog):
    pages_conditions_subquery = Page.objects.filter_by_catalog(catalog) \
                                            .filter(conditions=OuterRef('pk'))
    questionsets_conditions_subquery = QuestionSet.objects.filter_by_catalog(catalog) \
                                                          .filter(conditions=OuterRef('pk'))
    questions_conditions_subquery = Question.objects.filter_by_catalog(catalog) \
                                                    .filter(conditions=OuterRef('pk'))

    return Condition.objects.annotate(
        has_page=Exists(pages_conditions_subquery),
        has_questionset=Exists(questionsets_conditions_subquery),
        has_question=Exists(questions_conditions_subquery)
    ).filter(
        Q(has_page=True) | Q(has_questionset=True) | Q(has_question=True)
    ).distinct().select_related('source', 'target_option')


def resolve_conditions(catalog, values, sets):
    # resolve all conditions and return for each condition the set_prefix and set_index for which it resolved true
    conditions = defaultdict(set)
    if sets:
        for condition in get_catalog_conditions(catalog):
            conditions[condition.id] = {
                (set_prefix, set_index)
                for set_prefix, set_index in chain.from_iterable(sets.values())
                if condition.resolve(values, set_prefix=set_prefix, set_index=set_index)
            }

    return conditions


def compute_sets(values):
    # compute sets from values (including empty values)
    sets = defaultdict(set)
    for attribute, set_prefix, set_index in values.distinct_list():
        sets[attribute].add((set_prefix, set_index))
    return sets


def compute_next_relevant_page(current_page, direction, catalog, resolved_conditions):
    # recursively compute the next relevant page based on resolved conditions.
    # first, get the next page from the catalog based on the specified direction
    next_page = (
        catalog.get_prev_page(current_page) if direction == 'prev'
        else catalog.get_next_page(current_page)
    )

    # if there is no next page, return None
    if not next_page:
        return None

    # check if the next page meets the conditions
    if compute_show_page(next_page, resolved_conditions):
        return next_page

    # recursive step: check the next page
    return compute_next_relevant_page(next_page, direction, catalog, resolved_conditions)


def compute_show_page(page, conditions):
    # determine if a page should be shown based on resolved conditions
    # show only pages with resolved conditions, but show all pages without conditions
    pages_conditions = {page.id for page in page.conditions.all()}

    if pages_conditions:
        # check if any valuesets for set_prefix = '' resolved
        # for non collection pages restrict further to set_index = 0

        return any(
            (set_prefix == '') and (page.is_collection or set_index == 0)
            for page_condition in pages_conditions
            for set_prefix, set_index in conditions[page_condition]
        )
    else:
        return True


def compute_navigation(section, project, snapshot=None):
    # get all values for this project and snapshot
    values = project.values.filter(snapshot=snapshot).select_related('attribute', 'option')

    # compute sets from values (including empty values)
    sets = compute_sets(values)

    # resolve all conditions to get a dict mapping conditions to set_indexes
    conditions = resolve_conditions(project.catalog, values, sets)

    # compute sets anew, but without empty optional values
    sets = compute_sets(values.exclude_empty_optional(project.catalog))

    # query distinct, non empty set values
    values_list = values.exclude_empty().distinct_list()

    navigation = []
    for catalog_section in project.catalog.elements:
        navigation_section = {
            'id': catalog_section.id,
            'uri': catalog_section.uri,
            'title': markdown2html(catalog_section.short_title or catalog_section.title),
            'first': catalog_section.elements[0].id if catalog_section.elements else None,
            'count': 0,
            'total': 0
        }
        if section is not None and catalog_section.id == section.id:
            navigation_section['pages'] = []

        for page in catalog_section.elements:

            # determine if a page should be shown or not
            show = compute_show_page(page, conditions)

            # count the total number of questions, taking sets and conditions into account
            counts = count_questions(page, sets, conditions)

            # filter the values_list for the attributes, and compute the total sum of counts
            count = sum(1 for value in values_list if value[0] in counts)
            total = sum(counts.values())

            navigation_section['count'] += count
            navigation_section['total'] += total

            if 'pages' in navigation_section:
                navigation_section['pages'].append({
                    'id': page.id,
                    'uri': page.uri,
                    'title': markdown2html(page.short_title or page.title),
                    'show': show,
                    'count': count,
                    'total': total
                })

        navigation.append(navigation_section)

    return navigation


def compute_progress(project, snapshot=None):
    # get all values for this project and snapshot
    values = project.values.filter(snapshot=snapshot).select_related('attribute', 'option')

    # compute sets from values (including empty values)
    sets = compute_sets(values)

    # resolve all conditions to get a dict mapping conditions to set_indexes
    conditions = resolve_conditions(project.catalog, values, sets)

    # query distinct, non empty set values
    values_list = values.exclude_empty().distinct_list()

    # compute sets anew, but without empty optional values
    sets = compute_sets(values.exclude_empty_optional(project.catalog))

    # count the total number of questions, taking sets and conditions into account
    counts = count_questions(project.catalog, sets, conditions)

    # filter the values_list for the attributes, and compute the total sum of counts
    count = sum(1 for value in values_list if value[0] in counts)
    total = sum(counts.values())

    return count, total


def count_questions(element, sets, conditions):
    counts = defaultdict(int)

    if isinstance(element, (Page, QuestionSet)):
        # obtain the maximum number of set-distinct values for the questions in this page or
        # question set. this number is how often each question is displayed and we will use
        # this number to determine how often a question needs to be counted
        element_sets = set()

        # count the sets for the id attribute of the page or question
        if element.attribute is not None:
            # nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
            for set_prefix, set_index in sets[element.attribute.id]:
                element_sets.add((set_prefix, set_index))

        # count the sets for the questions in the page or question
        for child in element.elements:
            if isinstance(child, Question):
                if child.attribute is not None:
                    # nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
                    for set_prefix, set_index in sets[child.attribute.id]:
                        element_sets.add((set_prefix, set_index))

        # look for the elements conditions
        element_conditions = {condition.id for condition in element.conditions.all()}

        # if this element has conditions: check if those conditions resolve for the found sets
        if element_conditions:
            # compute the intersection of the conditions of this child with the full set of conditions
            element_condition_intersection = {
                (set_prefix, set_index)
                for condition_id, condition in conditions.items()
                for set_prefix, set_index in conditions[condition_id]
                if condition_id in element_conditions
            }

            resolved_sets = element_sets.intersection(element_condition_intersection)
            if resolved_sets:
                set_count = len(resolved_sets) if element.is_collection else 1
            else:
                # return immediately and do not consider children, this page/questionset is hidden
                return counts
        else:
            set_count = len(element_sets) if element.is_collection else 1

    # loop over all children of this element
    for child in element.elements:
        # look for the child element's conditions
        if isinstance(child, (Page, QuestionSet, Question)):
            child_conditions = {condition.id for condition in child.conditions.all()}
        else:
            child_conditions = set()

        # compute the intersection of the conditions of this child with the full set of conditions
        child_condition_intersection = {
            (set_prefix, set_index)
            for condition_id, condition in conditions.items()
            for set_prefix, set_index in conditions[condition_id]
            if condition_id in child_conditions
        }

        # check if the element either has no condition or its conditions intersect with the full set of conditions
        if not child_conditions or child_condition_intersection:
            if isinstance(child, Question):
                # for regular questions add the set_count to the counts dict, since the
                # question should be answered in every set
                # for optional questions add just the number of present answers, so that
                # only answered questions count for the progress/navigation
                # use the max function, since the same attribute could appear twice in the tree
                if child.attribute is not None:
                    if child.is_optional:
                        child_count = len(sets[child.attribute.id])
                        counts[child.attribute.id] = max(counts[child.attribute.id], child_count)
                    else:
                        if child_condition_intersection:
                            # update the set_count for the current question (child element)
                            # count only the sets that have conditions resolved to true
                            current_count = len(element_sets.intersection(child_condition_intersection))
                        else:
                            current_count = set_count

                        counts[child.attribute.id] = max(counts[child.attribute.id], current_count)
            else:
                # for everything else, call this function recursively
                counts.update(count_questions(child, sets, conditions))

    return counts

1	from collections import defaultdict	4✔
2	from itertools import chain	4✔
3
4	from django.db.models import Exists, OuterRef, Q	4✔
5
6	from rdmo.conditions.models import Condition	4✔
7	from rdmo.core.utils import markdown2html	4✔
8	from rdmo.questions.models import Page, Question, QuestionSet	4✔
9
10
11	def get_catalog_conditions(catalog):	4✔
12	pages_conditions_subquery = Page.objects.filter_by_catalog(catalog) \	4✔
13	.filter(conditions=OuterRef('pk'))
14	questionsets_conditions_subquery = QuestionSet.objects.filter_by_catalog(catalog) \	4✔
15	.filter(conditions=OuterRef('pk'))
16	questions_conditions_subquery = Question.objects.filter_by_catalog(catalog) \	4✔
17	.filter(conditions=OuterRef('pk'))
18
19	return Condition.objects.annotate(	4✔
20	has_page=Exists(pages_conditions_subquery),
21	has_questionset=Exists(questionsets_conditions_subquery),
22	has_question=Exists(questions_conditions_subquery)
23	).filter(
24	Q(has_page=True) \| Q(has_questionset=True) \| Q(has_question=True)
25	).distinct().select_related('source', 'target_option')
26
27
28	def resolve_conditions(catalog, values, sets):	4✔
29	# resolve all conditions and return for each condition the set_prefix and set_index for which it resolved true
30	conditions = defaultdict(set)	4✔
31	if sets:	4✔
32	for condition in get_catalog_conditions(catalog):	4✔
33	conditions[condition.id] = {	4✔
34	(set_prefix, set_index)
35	for set_prefix, set_index in chain.from_iterable(sets.values())
36	if condition.resolve(values, set_prefix=set_prefix, set_index=set_index)
37	}
38
39	return conditions	4✔
40
41
42	def compute_sets(values):	4✔
43	# compute sets from values (including empty values)
44	sets = defaultdict(set)	4✔
45	for attribute, set_prefix, set_index in values.distinct_list():	4✔
46	sets[attribute].add((set_prefix, set_index))	4✔
47	return sets	4✔
48
49
50	def compute_next_relevant_page(current_page, direction, catalog, resolved_conditions):	4✔
51	# recursively compute the next relevant page based on resolved conditions.
52	# first, get the next page from the catalog based on the specified direction
53	next_page = (	4✔
54	catalog.get_prev_page(current_page) if direction == 'prev'
55	else catalog.get_next_page(current_page)
56	)
57
58	# if there is no next page, return None
59	if not next_page:	4✔
60	return None	×
61
62	# check if the next page meets the conditions
63	if compute_show_page(next_page, resolved_conditions):	4✔
64	return next_page	4✔
65
66	# recursive step: check the next page
67	return compute_next_relevant_page(next_page, direction, catalog, resolved_conditions)	4✔
68
69
70	def compute_show_page(page, conditions):	4✔
71	# determine if a page should be shown based on resolved conditions
72	# show only pages with resolved conditions, but show all pages without conditions
73	pages_conditions = {page.id for page in page.conditions.all()}	4✔
74
75	if pages_conditions:	4✔
76	# check if any valuesets for set_prefix = '' resolved
77	# for non collection pages restrict further to set_index = 0
78
79	return any(	4✔
80	(set_prefix == '') and (page.is_collection or set_index == 0)
81	for page_condition in pages_conditions
82	for set_prefix, set_index in conditions[page_condition]
83	)
84	else:
85	return True	4✔
86
87
88	def compute_navigation(section, project, snapshot=None):	4✔
89	# get all values for this project and snapshot
90	values = project.values.filter(snapshot=snapshot).select_related('attribute', 'option')	4✔
91
92	# compute sets from values (including empty values)
93	sets = compute_sets(values)	4✔
94
95	# resolve all conditions to get a dict mapping conditions to set_indexes
96	conditions = resolve_conditions(project.catalog, values, sets)	4✔
97
98	# compute sets anew, but without empty optional values
99	sets = compute_sets(values.exclude_empty_optional(project.catalog))	4✔
100
101	# query distinct, non empty set values
102	values_list = values.exclude_empty().distinct_list()	4✔
103
104	navigation = []	4✔
105	for catalog_section in project.catalog.elements:	4✔
106	navigation_section = {	4✔
107	'id': catalog_section.id,
108	'uri': catalog_section.uri,
109	'title': markdown2html(catalog_section.short_title or catalog_section.title),
110	'first': catalog_section.elements[0].id if catalog_section.elements else None,
111	'count': 0,
112	'total': 0
113	}
114	if section is not None and catalog_section.id == section.id:	4✔
115	navigation_section['pages'] = []	4✔
116
117	for page in catalog_section.elements:	4✔
118
119	# determine if a page should be shown or not
120	show = compute_show_page(page, conditions)	4✔
121
122	# count the total number of questions, taking sets and conditions into account
123	counts = count_questions(page, sets, conditions)	4✔
124
125	# filter the values_list for the attributes, and compute the total sum of counts
126	count = sum(1 for value in values_list if value[0] in counts)	4✔
127	total = sum(counts.values())	4✔
128
129	navigation_section['count'] += count	4✔
130	navigation_section['total'] += total	4✔
131
132	if 'pages' in navigation_section:	4✔
133	navigation_section['pages'].append({	4✔
134	'id': page.id,
135	'uri': page.uri,
136	'title': markdown2html(page.short_title or page.title),
137	'show': show,
138	'count': count,
139	'total': total
140	})
141
142	navigation.append(navigation_section)	4✔
143
144	return navigation	4✔
145
146
147	def compute_progress(project, snapshot=None):	4✔
148	# get all values for this project and snapshot
149	values = project.values.filter(snapshot=snapshot).select_related('attribute', 'option')	4✔
150
151	# compute sets from values (including empty values)
152	sets = compute_sets(values)	4✔
153
154	# resolve all conditions to get a dict mapping conditions to set_indexes
155	conditions = resolve_conditions(project.catalog, values, sets)	4✔
156
157	# query distinct, non empty set values
158	values_list = values.exclude_empty().distinct_list()	4✔
159
160	# compute sets anew, but without empty optional values
161	sets = compute_sets(values.exclude_empty_optional(project.catalog))	4✔
162
163	# count the total number of questions, taking sets and conditions into account
164	counts = count_questions(project.catalog, sets, conditions)	4✔
165
166	# filter the values_list for the attributes, and compute the total sum of counts
167	count = sum(1 for value in values_list if value[0] in counts)	4✔
168	total = sum(counts.values())	4✔
169
170	return count, total	4✔
171
172
173	def count_questions(element, sets, conditions):	4✔
174	counts = defaultdict(int)	4✔
175
176	if isinstance(element, (Page, QuestionSet)):	4✔
177	# obtain the maximum number of set-distinct values for the questions in this page or
178	# question set. this number is how often each question is displayed and we will use
179	# this number to determine how often a question needs to be counted
180	element_sets = set()	4✔
181
182	# count the sets for the id attribute of the page or question
183	if element.attribute is not None:	4✔
184	# nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
185	for set_prefix, set_index in sets[element.attribute.id]:	4✔
186	element_sets.add((set_prefix, set_index))	4✔
187
188	# count the sets for the questions in the page or question
189	for child in element.elements:	4✔
190	if isinstance(child, Question):	4✔
191	if child.attribute is not None:	4✔
192	# nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
193	for set_prefix, set_index in sets[child.attribute.id]:	4✔
194	element_sets.add((set_prefix, set_index))	4✔
195
196	# look for the elements conditions
197	element_conditions = {condition.id for condition in element.conditions.all()}	4✔
198
199	# if this element has conditions: check if those conditions resolve for the found sets
200	if element_conditions:	4✔
201	# compute the intersection of the conditions of this child with the full set of conditions
202	element_condition_intersection = {	4✔
203	(set_prefix, set_index)
204	for condition_id, condition in conditions.items()
205	for set_prefix, set_index in conditions[condition_id]
206	if condition_id in element_conditions
207	}
208
209	resolved_sets = element_sets.intersection(element_condition_intersection)	4✔
210	if resolved_sets:	4✔
211	set_count = len(resolved_sets) if element.is_collection else 1	4✔
212	else:
213	# return immediately and do not consider children, this page/questionset is hidden
214	return counts	4✔
215	else:
216	set_count = len(element_sets) if element.is_collection else 1	4✔
217
218	# loop over all children of this element
219	for child in element.elements:	4✔
220	# look for the child element's conditions
221	if isinstance(child, (Page, QuestionSet, Question)):	4✔
222	child_conditions = {condition.id for condition in child.conditions.all()}	4✔
223	else:
224	child_conditions = set()	4✔
225
226	# compute the intersection of the conditions of this child with the full set of conditions
227	child_condition_intersection = {	4✔
228	(set_prefix, set_index)
229	for condition_id, condition in conditions.items()
230	for set_prefix, set_index in conditions[condition_id]
231	if condition_id in child_conditions
232	}
233
234	# check if the element either has no condition or its conditions intersect with the full set of conditions
235	if not child_conditions or child_condition_intersection:	4✔
236	if isinstance(child, Question):	4✔
237	# for regular questions add the set_count to the counts dict, since the
238	# question should be answered in every set
239	# for optional questions add just the number of present answers, so that
240	# only answered questions count for the progress/navigation
241	# use the max function, since the same attribute could appear twice in the tree
242	if child.attribute is not None:	4✔
243	if child.is_optional:	4✔
244	child_count = len(sets[child.attribute.id])	4✔
245	counts[child.attribute.id] = max(counts[child.attribute.id], child_count)	4✔
246	else:
247	if child_condition_intersection:	4✔
248	# update the set_count for the current question (child element)
249	# count only the sets that have conditions resolved to true
250	current_count = len(element_sets.intersection(child_condition_intersection))	×
251	else:
252	current_count = set_count	4✔
253
254	counts[child.attribute.id] = max(counts[child.attribute.id], current_count)	4✔
255	else:
256	# for everything else, call this function recursively
257	counts.update(count_questions(child, sets, conditions))	4✔
258
259	return counts	4✔

rdmorganiser / rdmo / 16070961563

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