14404330126

Committed 11 Apr 2025 01:31PM UTC coverage: 90.789% (+0.3%) from 90.478%

Build # 14404330126

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1195 from rdmorganiser/2.3.0

RDMO 2.3.0 ⭐

Run Details

989 of 1076 branches covered (91.91%)

9176 of 10107 relevant lines covered (90.79%)

3.63 hits per line

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

98.02

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)

    # 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()

    # 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	# query distinct, non empty set values
99	values_list = values.exclude_empty().distinct_list()	4✔
100
101	navigation = []	4✔
102	for catalog_section in project.catalog.elements:	4✔
103	navigation_section = {	4✔
104	'id': catalog_section.id,
105	'uri': catalog_section.uri,
106	'title': markdown2html(catalog_section.short_title or catalog_section.title),
107	'first': catalog_section.elements[0].id if catalog_section.elements else None,
108	'count': 0,
109	'total': 0
110	}
111	if section is not None and catalog_section.id == section.id:	4✔
112	navigation_section['pages'] = []	4✔
113
114	for page in catalog_section.elements:	4✔
115
116	# determine if a page should be shown or not
117	show = compute_show_page(page, conditions)	4✔
118
119	# count the total number of questions, taking sets and conditions into account
120	counts = count_questions(page, sets, conditions)	4✔
121
122	# filter the values_list for the attributes, and compute the total sum of counts
123	count = sum(1 for value in values_list if value[0] in counts)	4✔
124	total = sum(counts.values())	4✔
125
126	navigation_section['count'] += count	4✔
127	navigation_section['total'] += total	4✔
128
129	if 'pages' in navigation_section:	4✔
130	navigation_section['pages'].append({	4✔
131	'id': page.id,
132	'uri': page.uri,
133	'title': markdown2html(page.short_title or page.title),
134	'show': show,
135	'count': count,
136	'total': total
137	})
138
139	navigation.append(navigation_section)	4✔
140
141	return navigation	4✔
142
143
144	def compute_progress(project, snapshot=None):	4✔
145	# get all values for this project and snapshot
146	values = project.values.filter(snapshot=snapshot).select_related('attribute', 'option')	4✔
147
148	# compute sets from values (including empty values)
149	sets = compute_sets(values)	4✔
150
151	# resolve all conditions to get a dict mapping conditions to set_indexes
152	conditions = resolve_conditions(project.catalog, values, sets)	4✔
153
154	# query distinct, non empty set values
155	values_list = values.exclude_empty().distinct_list()	4✔
156
157	# count the total number of questions, taking sets and conditions into account
158	counts = count_questions(project.catalog, sets, conditions)	4✔
159
160	# filter the values_list for the attributes, and compute the total sum of counts
161	count = sum(1 for value in values_list if value[0] in counts)	4✔
162	total = sum(counts.values())	4✔
163
164	return count, total	4✔
165
166
167	def count_questions(element, sets, conditions):	4✔
168	counts = defaultdict(int)	4✔
169
170	if isinstance(element, (Page, QuestionSet)):	4✔
171	# obtain the maximum number of set-distinct values for the questions in this page or
172	# question set. this number is how often each question is displayed and we will use
173	# this number to determine how often a question needs to be counted
174	element_sets = set()	4✔
175
176	# count the sets for the id attribute of the page or question
177	if element.attribute is not None:	4✔
178	# nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
179	for set_prefix, set_index in sets[element.attribute.id]:	4✔
180	element_sets.add((set_prefix, set_index))	4✔
181
182	# count the sets for the questions in the page or question
183	for child in element.elements:	4✔
184	if isinstance(child, Question):	4✔
185	if child.attribute is not None:	4✔
186	# nested loop over the separate set_prefix, set_index lists in sets[element.attribute.id]
187	for set_prefix, set_index in sets[child.attribute.id]:	4✔
188	element_sets.add((set_prefix, set_index))	4✔
189
190	# look for the elements conditions
191	element_conditions = {condition.id for condition in element.conditions.all()}	4✔
192
193	# if this element has conditions: check if those conditions resolve for the found sets
194	if element_conditions:	4✔
195	# compute the intersection of the conditions of this child with the full set of conditions
196	element_condition_intersection = {	4✔
197	(set_prefix, set_index)
198	for condition_id, condition in conditions.items()
199	for set_prefix, set_index in conditions[condition_id]
200	if condition_id in element_conditions
201	}
202
203	resolved_sets = element_sets.intersection(element_condition_intersection)	4✔
204	if resolved_sets:	4✔
205	set_count = len(resolved_sets) if element.is_collection else 1	4✔
206	else:
207	# return immediately and do not consider children, this page/questionset is hidden
208	return counts	4✔
209	else:
210	set_count = len(element_sets) if element.is_collection else 1	4✔
211
212	# loop over all children of this element
213	for child in element.elements:	4✔
214	# look for the child element's conditions
215	if isinstance(child, (Page, QuestionSet, Question)):	4✔
216	child_conditions = {condition.id for condition in child.conditions.all()}	4✔
217	else:
218	child_conditions = set()	4✔
219
220	# compute the intersection of the conditions of this child with the full set of conditions
221	child_condition_intersection = {	4✔
222	(set_prefix, set_index)
223	for condition_id, condition in conditions.items()
224	for set_prefix, set_index in conditions[condition_id]
225	if condition_id in child_conditions
226	}
227
228	# check if the element either has no condition or its conditions intersect with the full set of conditions
229	if not child_conditions or child_condition_intersection:	4✔
230	if isinstance(child, Question):	4✔
231	# for regular questions add the set_count to the counts dict, since the
232	# question should be answered in every set
233	# for optional questions add just the number of present answers, so that
234	# only answered questions count for the progress/navigation
235	# use the max function, since the same attribute could appear twice in the tree
236	if child.attribute is not None:	4✔
237	if child.is_optional:	4✔
238	child_count = len(sets[child.attribute.id])	4✔
239	counts[child.attribute.id] = max(counts[child.attribute.id], child_count)	4✔
240	else:
241	if child_condition_intersection:	4✔
242	# update the set_count for the current question (child element)
243	# count only the sets that have conditions resolved to true
244	current_count = len(element_sets.intersection(child_condition_intersection))	×
245	else:
246	current_count = set_count	4✔
247
248	counts[child.attribute.id] = max(counts[child.attribute.id], current_count)	4✔
249	else:
250	# for everything else, call this function recursively
251	counts.update(count_questions(child, sets, conditions))	4✔
252
253	return counts	4✔

rdmorganiser / rdmo / 14404330126

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