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],
)
as_json_subquery_for_observations() → ObservationComplianceQueryset

Returns a Subquery yielding a JSON payload containing “compliance” and “weight”.

example:

{
  "compliance": 1.0,
  "weight": 1.0
}
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, 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)

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, float | None]

maps field names to compliance

subform_compliance: dict[str, float | None]

maps subform name to compliance

property weighed_compliance: WeighedValue

Weighed compliance of the observation

exception DoesNotExist
exception MultipleObjectsReturned
class compliance.models.SubformComplianceQuerySet(model=None, query=None, using=None, hints=None)
for_observation_querysets(*querysets: QuerySet) → SubformComplianceQuerySet

Filters to SubformCompliance rows whose parent ObservationCompliance matches observations from the given queryset(s).

Parameters:

querysets – one or more observation querysets

correlated_to_outer_observation(subform_names: Iterable[str]) → SubformComplianceQuerySet

Filters to answered (non-null) rows of the given subforms belonging to the OUTER query’s observation, grouped per observation. Base for the compliance/weight subqueries used by the over-time widgets.

Parameters:

subform_names – subform names to include.

weighted_compliance_for_outer_observation(subform_names: Iterable[str]) → Subquery

Per-instance weighted average compliance of the selected subforms for the outer observation. Each row’s effective weight (answered field weights, or custom subform weight) is used.

Parameters:

subform_names – subform names to include.

weight_total_for_outer_observation(subform_names: Iterable[str]) → Subquery

Sum of the selected subforms’ weights for the outer observation. Weights each observation across time by only the selected subforms, so filtered-out subforms don’t inflate the cross-observation weighting.

Parameters:

subform_names – subform names to include.

class compliance.models.SubformCompliance(*args, **kwargs)

Stores per-instance compliance data for sub-observations within an observation.

Each row represents one instance of a subform on a single observation, with its resolved compliance score and effective weight.

property weighed_compliance: WeighedValue

Weighed compliance of this sub-observation instance.

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])
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, CustomField | Field]

Fields in given audit form, mapped by their field name

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.

field_weights_for(form_id: int, field_names: Collection[str]) → dict[str, float]

Builds a {field_name: weight} map for the requested fields on a form, to be passed to weighted_average(). Fields without a configured weight fall back to FIELD_WEIGHT_DEFAULT.

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(compliances: ObservationComplianceQueryset, field_names: Collection[str] | None = None) → WeighedValue

Average compliance across observations, optionally for a subset of fields.

When field_names is None, delegates to calc_average_compliance() which computes the weighted average of pre-computed observation compliance entirely in SQL.

When field_names is provided, the multi-field compliance rules apply (see Multi-field Compliance in the docs). Compliance is calculated per observation in SQL from the pre-computed ObservationCompliance.field_compliance map via weighted_average(), then averaged across observations.

field_compliance already stores per-field scores with multiple answers (e.g. the same field across several subform instances) pre-averaged into a single value per observation, so a single SQL expression covers both top-level and subform-nested fields without touching SubformCompliance or the heavy compliance_data JSON.

Note

Per the documented multi-field rules, the field-subset path applies field weights only and deliberately ignores subform weights, subform instance counts, the compliance_calculation=require rule, and form-level compliance weighting. Each observation contributes equally to the average. This diverges from the legacy Python calculation, which re-derived these from compliance_data; the divergence is intentional and pending confirmation with the compliance util author (see #32979).

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

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

The compliance weight column carries the observation’s own weight (observation_weight), not the sum of the selected field weights. Downstream consumers (e.g. the ward compliance widget) use it to weight each observation when aggregating field-subset compliance.

When field_names is provided, the per-observation compliance value is calculated in SQL from the pre-computed field_compliance map via weighted_average(), following the documented multi-field rules (field weights only). See get_average_compliance() for the full semantics.

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, float | None]]) → tuple[float | None, dict[str, float | None]]

Extracts average subobservation compliance and a dictionary and fields mapped to their average answer compliance

compliance.utils.extract_field_compliances(compliance_data: dict[str, float | None] | dict[str, list[dict[str, float | None]]], 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.build_subform_compliance_objects(observation_compliance: ObservationCompliance, compliance_data: dict[str, float | None] | dict[str, list[dict[str, float | None]]], field_weights: dict[str, float], subform_weights: dict[str, float], propagate_weight: bool) → list[SubformCompliance]

Builds unsaved SubformCompliance instances from compliance data.

Parameters:

  • observation_compliance – parent observation compliance record

  • compliance_data – compliance tree from build_compliance_tree()

  • field_weights – maps field name to configured compliance weight

  • subform_weights – maps subform name to hardcoded compliance weight (only subforms with explicit weight)

  • propagate_weight – whether to use cumulative question weight for subforms with hardcoded weight

Returns:

list of unsaved SubformCompliance instances

compliance.utils.sync_subform_compliances(compliance_ids: list[int], new_objects: list[SubformCompliance]) → None

Syncs SubformCompliance rows for multiple observations in bulk, updating existing rows in place and only creating/deleting when instance counts change per (observation_compliance, subform_name) group.

Rows with unchanged values are still included in the bulk_update — Postgres will execute the UPDATE but skip writing a new tuple if no column values differ (HOT update). A Python-side equality check could avoid the query entirely but adds complexity for minimal gain.

Parameters:
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 = 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

calculate_answer_compliance(field: CustomField | Field, answer: str | int | bool | list | float | None) → WeighedValue

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 = None) → WeighedValue

Calculate compliance for sub-observation

calculate_observation_compliance(observation: BaseAuditModel, field_names: Collection[str] | None = None) → WeighedValue

Calculate weighed compliance for observation

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

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 = None, key_attribute: str | None = None) → dict[str | int | BaseAuditModel, 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 = None) → WeighedValue

Calculates compliance for multiple observations. Returns weighed average.

calculate_subobservations_compliance(sub_observations: Iterable[BaseSubObservation], field_names: Collection[str] | None = None) → WeighedValue

Calculates compliance for multiple sub-observations

static generate_mixed_compliances(observations: Iterable[BaseAuditModel], field_names: Collection[str] | None = None) → Iterator[tuple[BaseAuditModel, WeighedValue]]

Calculate observation compliance in bulk and stream results as an iterator

static calculate_mixed_compliance(observations: Iterable[BaseAuditModel], field_names: Collection[str] | None = 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 = 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.backfill_subform_compliances_for_form(form_id: int)

Backfills SubformCompliance rows for a single form.

Skips ObservationCompliance records that already have associated SubformCompliance rows, making re-runs safe.

Parameters:

form_id – PK of the AuditForm to backfill

(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