Compliance

Models

class compliance.models.ObservationComplianceQueryset(model=None, query=None, using=None, hints=None)

subquery_for_observations() → ObservationComplianceQueryset

Filters subquery based on outer ObservationQueryset.

for example, to annotate observation with compliance value:

CustomObservation.objects.annotate(
    compliance=ObservationCompliance.objects.subquery_for_observations().values('observation_compliance')[:1],
)

for_form_observations(form_id: int, observation_ids: Iterable[int]) → ObservationComplianceQueryset: Gets compliances for observations by id where all observations belong to the same form

for_observation_querysets(*querysets: QuerySet) → ObservationComplianceQueryset

Filter compliance objects by given observation queryset(s)

Pass multiple queryset objects if using multiple different observation models. If all observations are of the same model, it is advised to build a single queryset.

Parameters:: querysets – any number of observation querysets
Returns:: queryset representing compliance objects for the passed observations.

for_observations(observations: Iterable[Observation]) → ObservationComplianceQueryset

Filter compliance objects by observations. Does not verify that all observations have a corresponding compliance object.

Parameters:: observations – queryset or any iterable of observations.

calc_average_compliance(weighted: bool = True) → WeighedValue

Average compliance value in all observations in this queryset.

Parameters:: weighted – Whether to take observation compliance weight into account

get_field_compliance(*field_names: str) → dict[str, audit_builder.utils.WeighedValue]

Gets average compliance for given field(s) from the queryset.

Parameters:: field_names – names of fields to get compliance from
Returns:: a dictionary mapping field name to its average compliance (weighed according to number of compliances)

get_subform_compliance(*subform_names: str) → dict[str, audit_builder.utils.WeighedValue]

Average compliance for each subform

Parameters:: subform_names – names of the subform to calculate average compliance from
Returns:: a dictionary mapping subform name to its compliance and weighted according to number of occurences

iterate_subobservation_field_compliance(audit_form: AuditForm, field_names: set[str] | None = None) → Iterator[tuple[int, str, Optional[float], float]]

Yields field compliance for each sub-observation (stored in compliance data). This is subform compliance pre weighting decision (e.g. whether to propegate wieght or use subform weight or default 1).

Parameters:

audit_form – name of the audit form, used to fetch field weights
field_names – which fields to use when calculating the field compliance

get_subobservation_field_compliance_dataframe(audit_form: AuditForm, field_names: set[str] | None = None) → DataFrame: Creates a dataframe of sub-observation field compliances

class compliance.models.ObservationCompliance(id, order, publish, uid, created, modified, form, observation_id, schema_hash, observation_hash, observation_compliance, observation_weight, compliance_data, field_compliance, subform_compliance)

field_compliance: dict[str, Optional[float]]: maps field names to compliance

subform_compliance: dict[str, Optional[float]]: maps subform name to compliance

property weighed_compliance: WeighedValue: Weighed compliance of the observation

exception DoesNotExist

exception MultipleObjectsReturned

Compliance getter

class compliance.compliance_getter.ComplianceGetter

A compliance calculator that uses ObservationCompliance model instead of computing compliance each time. It is an equivalent to ComplianceCalculator, but is not limited to working with one form at a time.

The class can be used as a callable:

getter = ComplianceGetter()
compliance: WeighedValue = getter(observation)

Or by explicitly calling relevant methods:

getter = ComplianceGetter()
compliance: WighedValue = getter.get_observation_compliance(observation)
compliance: WighedValue = getter.get_observations_compliance([observation])

required_compliance_fields: dict[int, set[str]]: names of fields with compliance_calculation=REQUIRED in each form

get_form(form_id: int) → AuditForm: Gets audit form by id. The form instance is cached in the class level

get_fields(form_id: int) → dict[str, Union[audit_builder.models.CustomField, django.db.models.fields.Field]]: Fields in given audit form, mapped by their field name

get_required_fields(form_id: int) → set[str]: set of fields with compliance=required in given form

get_field_weights(form_id: int) → dict[str, float]

Returns dictionary mapping field name to field’s weight.

If a field is missing in the dictionary, it is implied that it’s weigh is `FIELD_WEIGHT_DEFAULT. This applies in particular to hardcoded forms where each field is weighed equally.

get_subform_weights(form_id: int) → dict[str, float]: Returns a dictionary mapping subform name to its weight. By default, subforms inherit weight from their answered questions, but this can be overridden in custom subforms.

get_field_compliance(form_id: int, data: Iterable[tuple[str, Optional[float]]], field_names: Collection[str]) → WeighedValue

Extracts compliance data for a set of fields

Parameters:

form_id – AuditForm’s primary key
data – a collection of compliance data where field name is mapped to compliance value
field_names – field names to collect compliance for

Returns:

average compliance value for given fields

get_subform_compliance(form_id: int, data: Iterable[tuple[str, list[dict[str, Optional[float]]]]], field_names: set[str]) → WeighedValue

Calculates subform compliance using the provided fields

Parameters:

form_id – pk of the AuditForm
data – tuples mapping subform name to a list of answer compliances

Returns:

a weighted value representing average of passed compliance data

get_observations_compliance(observations: Iterable[BaseAuditModel], field_names: Collection[str] | None = None) → WeighedValue

Gets average compliance for given observations.

if field names are specified, the compliance is re-calculated for a subset of fields from field compliance data

get_average_compliance_for_compliance_data(form_id: int, compliance_data: dict[str, Optional[float]] | dict[str, list[dict[str, Optional[float]]]], field_names: Collection[str]) → WeighedValue

Get subform compliance with ObservationCompliance compliance_data field for given field names

Parameters:

form_id – pk of the AuditForm
compliance_data – dict mapping fields & subform names to compliance for a subform.
field_names – fields to be used for compliance calculation

Returns:

weighted average compliance for compliance_data (an observation) for the specified fields

get_average_compliance(compliances: ObservationComplianceQueryset, field_names: Collection[str] | None = None) → WeighedValue

Gets average compliance for given compliance objects and a subset of fields.

if field names are specified, the compliance is re-calculated for a subset of fields from field compliance data

get_compliance_dataframe(observations: Iterable[BaseAuditModel], field_names: Sequence[str] = None) → DataFrame

Creates a pandas Dataframe containing compliance value and weight for each observation.

The compliance weight column represents the weight of the observation rather than sum of selected fields.

Parameters:

observations – any iterable of observations
field_names – fields to be used for compliance calculation

Returns:

dataframe containing compliance value and compliance weight, indexed by (form_id, observation_id)

Utilities

compliance.utils.trigger_compliance_chunked(form_id: int, observation_ids: Iterable[int], batch_size: int = 200, priority: int = 10, **kwargs)

Triggers update_observation_compliance() celery jobs for compliance calculation in chunks.

Parameters:

form_id – form pk
observation_ids – any iterable of observation ids belonging to the form
batch_size – how many observations should be calculated in each job
priority – override priority of the celery task
kwargs – any additional arguments for update_observation_compliance()

compliance.utils.trigger_compliance_calc(observations: Iterable[Observation], **kwargs)

Triggers update_observation_compliance() job after current transaction is complete. The function does not always trigger job right away, it waits until current transaction is committed and celery job will be able to access the updated observation objects.

The calculation will be triggered in batches, but grouping the observations by their form id, with larges batch size being set by COMPLIANCE_BATCH_SIZE.

Parameters:

observations – observations to recalculate compliance for, can be any iterable
kwargs – Any additional kwargs for the celery job update_observation_compliance()

compliance.utils.extract_subform_averages(subform_compliances: list[dict[str, Optional[float]]]) → tuple[Optional[float], dict[str, Optional[float]]]: Extracts average subobservation compliance and a dictionary and fields mapped to their average answer compliance

compliance.utils.extract_field_compliances(compliance_data: dict[str, Optional[float]] | dict[str, list[dict[str, Optional[float]]]], field_name: str) → Iterator[float | None]

Given a compliance map for flat for, or form with subforms, extracts compliance data for a single field. Compliance value is found by field mapped to compliance, or is observation has subforms, by iterating those subforms and looking for one that has the field

Parameters:

compliance_data – Dict mapping field name to its compliance, or a subform name to a list of field compliances
field_name – name of the field

Yields:

compliance values extracted from the map. Typically, a single value, but if form allows multiple answers, multiple compliances are returned.

compliance.utils.generate_observation_compliance_fields(calc: ComplianceCalculator, observation: Observation) → Iterable[tuple[str, Compliance | ComplianceMap | AnswerComplianceMap]]

Generates field values for ObservationCompliance.

Yields:: tuples containing field name and value to be assigned to the field

compliance.utils.parse_number_range_expression(expression: str) → list[tuple[Callable, float]]

Parses a number range expression and returns a list of tuples. Each tuple contains a comparison operator function and a float number. Supports the following comparison operators: > < >= <=.

Parameters:: expression – A string containing the equality expression to parse. The expression should contain one or more components, where each component is a comparison operator followed by a number.
Returns:: A list of tuples containing a comparison operator function from the operator module (gt, lt, ge, le) and a float number.

Example:

>>> parse_number_range_expression('>=10<=100')
>>> [(ge, 10.0), (le, 100.0)]

compliance.utils.parse_compliance_from_answer_values(answer_values: dict[str, Compliance], range_answer_values: list[tuple[str, Compliance]], answer: Answer) → Compliance

Parses compliance from answer values for a given number. Checks answer_values and if it satisfies any of the equality expressions in answer_values.

Parameters:

answer_values – A dict mapping answer values to compliance.
range_answer_values – A dict mapping answer range values to compliance like >=10<=100.
answer – An answer to check compliance against.

Returns:

The compliance value.

compliance.compliance_calculator.COMPLIANCE_CACHE_VERSION = 1: Cache schema version Increment this number when changing the schema of object being saved to cache

class compliance.compliance_calculator.ComplianceCalculator(audit_form: AuditForm, *, use_cache=True)

A class responsible for calculating compliance in the context of given audit form/config. Computes compliance by reading answers and the form schema.

Note

This class implements calculation logic from answers. This is slow and should not be used within application view logic. If you need to work with observation compliance data, use ComplianceGetter which uses pre-computed compliance data.

See also

Compliance

use_cache: bool: Whether this calculator instance should use cache Disabling cache means compliance will be computed every time, and none of the computed values will be stored in cache for subsequent calls, even if they use cache.

observation_model

Base Observation model for the current model.

Returns:: Observation model, or base observation class for hand hygiene uk and us variants

property weigh_sub_observations: bool: Whether subform weight should be used when calculating compliance between multiple subforms. If true, each sub-observation should have a weight equal sum of weight of questions answered within. If false, each subform has equal weight - 1

weigh_observations: Whether observation weight should be used when calculating compliance between observations. If True, each observation will have a weight equal sum of weights of answered questions. If false, each observation has the same weight - 1.

custom_fields: All custom fields in this audit

hardcoded_fields: Hardcoded fields in this audit and all subforms

subform_weights: Maps subform ids to the subform weight

compliance_field_names: Set of names of fields used for compliance calculation

compliance_field_names_including_ignored: Set of names of fields used for compliance calculation

hardcoded_field_values: Dict mapping field name to a dict mapping value to its compliance value

hardcoded_field_compliance_calculations: A dictionary mapping a hardcoded field name to its compliance calculation option (use, require, ignore)

get_fields(observation: BaseAuditModel | BaseSubObservation, field_names: Collection[str] = None, default_compliance_fields=True) → Iterator[CustomField | Field]

Gets hardcoded and custom fields for given sub/observation. By default returns all compliance fields

Parameters:

observation – observation or subobservation instance
field_names – optional list of field names to return
default_compliance_fields – if fields are not explicitly selected, whether to return only compliance fields with compliance=use. Else all fields with compliance logic are returned

get_hardcoded_subforms(field_names: set[str] | None = None) → Iterator[Type[BaseSubObservation]]: Filters hardcoded subforms based on selected

Calculates compliance for a given answer to a given field

Raises:: Incompliant – if field is configured to compliance == REQUIRE as this should invalidate compliance for entire observation

calculate_field_compliance(observation: BaseAuditModel | BaseSubObservation, field: CustomField | Field, raise_incompliant=True) → WeighedValue

Calculates compliance for a single field in an observation.

Extracts answer from the observation and uses calculate_answer_compliance() to compute and return value.

Parameters:

observation – observation containing the value of the field
field – the field
raise_incompliant – whether to re-railse Incompliant exception if field is marked as

Returns:

field’s compliance based on answer in the observation, weighed according to field’s properties

Raises:

Incompliant – if field is configured to compliance == REQUIRE and raise_incompliant = True

calculate_observation_field_compliance(observation: BaseAuditModel | BaseSubObservation, field_names: Collection[str] | None = None) → WeighedValue: Calculates compliance for questions inside sub/observation - common logic used for observations and sub-observations

calculate_subobservation_compliance(sub_observation: BaseSubObservation, field_names: Collection[str] = None) → WeighedValue: Calculate compliance for sub-observation

calculate_observation_compliance(observation: BaseAuditModel, field_names: Collection[str] = None) → WeighedValue: Calculate weighed compliance for observation

build_compliance_tree(observation: BaseAuditModel) → dict[str, Optional[float]] | dict[str, list[dict[str, Optional[float]]]]

For a given observation, builds a dictionary mapping field name to its compliance, and subform name to a list of dicts representing the same (+ empty string mapped to subobservation compliance).

Note:: Compliance weight information is not included in this structure, but is taken into account when calculating subform compliance.

Example:

{
    'field1': 1.0,
    'field2': None,
    'subform1': [
        {
            '': 0.95,
            'field3': 0.95
        },
    ],
}

calculate_observations_compliances(observations: Iterable[BaseAuditModel], field_names: Collection[str] = None, key_attribute: str | None = None) → dict[typing.Union[str, int, ForwardRef('BaseAuditModel')], audit_builder.utils.WeighedValue]

Calculates compliance for multiple observations and returns dict mapping each observation to its compliance

Parameters:: key_attribute – defines the observation attribute used to create the mapping key in the return dict.

calculate_observations_compliance(observations: Iterable[BaseAuditModel], field_names: Collection[str] = None) → WeighedValue: Calculates compliance for multiple observations. Returns weighed average.

calculate_subobservations_compliance(sub_observations: Iterable[BaseSubObservation], field_names: Collection[str] = None) → WeighedValue: Calculates compliance for multiple sub-observations

static generate_mixed_compliances(observations: Iterable[BaseAuditModel], field_names: Collection[str] = None) → Iterator[tuple['BaseAuditModel', audit_builder.utils.WeighedValue]]: Calculate observation compliance in bulk and stream results as an iterator

static calculate_mixed_compliance(observations: Iterable[BaseAuditModel], field_names: Collection[str] = None) → WeighedValue: Calculate compliance for a mixed group of observations from multiple observations using multiple calculators

aggregate_observation_compliance(observations: Iterable[BaseAuditModel], aggregate: Callable[[BaseAuditModel], AggregateKey], compliance_fields: Collection[str] = None) → Dict[AggregateKey, WeighedValue]

Calculate observation compliancem but break observations into groups based on result from the provided aggregate function

Parameters:

observations – any iterable of observations
aggregate – a function returning key used for grouping observations. Key can be any hashable object.
compliance_fields – fields used for compliance calculation

Returns:

dictionary mapping keys to average compliance of observations matching that key

Celery tasks

(celery task)compliance.tasks.update_observation_compliance(form_id: int, observation_ids: Iterable[int], check_hash: bool = True, create_missing: bool = True)

Updates (or optionally creates) compliance data for given observations.

Parameters:

form_id – PK of the AuditForm
observation_ids – pks of observations, model depends on audit form settings
check_hash – whether to skip compliance calculation if observation hash is already up-to-date
create_missing – whether to create missing compliance objects. False will only update existing observation compliances.

(celery task)compliance.tasks.fill_missing_compliances(form_id: int): Creates compliance data for observations within given form that do not have compliance data.

(celery task)compliance.tasks.handle_form_schema_change(form_id: int, *, new_hash: str) → bool

Job triggered by schema changes. It validates that the schema that triggered it is still up-to-date, and triggers observation schema re-calculation.

Observations with missing compliance object will not be affected

Parameters:

form_id – PK of the form whose schema has changed
new_hash – hash of the new schema after change. This hash is used to verify that there were no further changes since the job was triggered
observation_age_days – only trigger update for most recent observations, at most this old.

Returns:

boolean True if further job was triggered to update observation compliances, False if compliance