<commit_before># Python
import datetime
import dateutil.parser
import dateutil.relativedelta
import uuid
import random
from functools import wraps
import logging

# 3rd party
import numpy
import scipy.stats
import validictory
import json

# django core
from django.utils.translation import ugettext as _
import django.contrib.gis.db.models as gismodels
from django.db import models
from django.contrib.gis.geos import Point
from django.db.models import Q


# django 3rd party
from rest_framework.reverse import reverse
from phonenumber_field.modelfields import PhoneNumberField
from jsonfield import JSONField
from django_extensions.db.fields import (
    AutoSlugField, CreationDateTimeField, ModificationDateTimeField,
)

# Internal
from fields import UniqueActiveField


logger = logging.getLogger(__name__)


class TeamMember(models.Model):
    MALE = 'M'
    FEMALE = 'F'
    GENDER_CHOICES = (
        (MALE, _('Male')),
        (FEMALE, _('Female')),
    )
    member_id = AutoSlugField(blank=False,
                              unique=True,
                              populate_from="id",
                              separator='')
    first_name = models.CharField(blank=False, max_length=50)
    last_name = models.CharField(blank=False, max_length=50)
    mobile = PhoneNumberField(blank=True)
    email = models.EmailField(blank=True)
    birth_year = models.IntegerField(blank=False)
    gender = models.CharField(choices=GENDER_CHOICES, max_length=3, blank=True)

    created = CreationDateTimeField()
    modified = ModificationDateTimeField()

    _last_survey = None

    def last_survey(self):
        # TODO: maybe get the reverse relationship instead

        # Poor mans cache
        if self._last_survey:
            return self._last_survey

        survey = (HouseholdSurveyJSON.objects.filter(
            Q(team_lead=self)
            | Q(team_assistant=self)
            | Q(team_anthropometrist=self)
        ).order_by('-id') or [None])[0]

        self._last_survey = survey
        return survey

    def last_survey_position(self):
        last_survey = self.last_survey()
        if last_survey:
            for position in ['team_lead', 'team_assistant', 'team_anthropometrist']:
                if self.id == getattr(last_survey, position).id:
                    return position

            raise "TeamMember has last_survey but no position, this shouldn't happen."

    def last_survey_created(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.get_start_time()

    def last_survey_cluster_name(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.json['cluster_name']

    def last_survey_cluster(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.json['cluster']

    class Meta:
        get_latest_by = 'modified'
        ordering = ('-modified', '-created',)

    @property
    def random_id(self):
        return random.randint(10000, 100000)

    def __unicode__(self):
        return u'%s-%s %s' % (self.id, self.first_name, self.last_name)

    def get_absolute_url(self):
        return reverse('teammember-detail', args=[str(self.member_id)])


def validate_json(spec_file):
    survey_schema = None
    try:
        survey_schema = json.load(open(spec_file))
    except IOError as e:
        # log the error but it's ok if the spec is missing
        # if the validator is never called
        logger.exception(
            "Could not load the json spec %s. \nThis is ok if you are just setting up.\n%s" % (spec_file, e.message))

    @wraps(validate_json)
    def wrapped(value):
        assert survey_schema, "Trying to validate a non existant JSON schema file %s" % spec_file
        # It really should exist by now.
        validictory.validate(value, survey_schema, required_by_default=False)
    return wrapped


class BaseHouseholdMember(models.Model):

    GENDER = (
        ('M', 'Male'),
        ('F', 'Female'),
        ('O', 'Other'),
    )
    gender = models.CharField(max_length=1, choices=GENDER, blank=True)

    first_name = models.TextField(blank=True)
    index = models.SmallIntegerField(blank=True)
    birthdate = models.DateField(blank=True, null=True)

    muac = models.SmallIntegerField(null=True)  # in millimeter ?
    weight = models.FloatField(null=True)       # probably in kilograms ?
    height = models.FloatField(null=True)       # probably in centimeters ?
    edema = models.NullBooleanField()

    class Meta:
        abstract = True


class HouseholdMember(BaseHouseholdMember):
    household_survey = models.ForeignKey('HouseholdSurveyJSON', related_name='members')


class BaseHouseholdSurveyJSON(gismodels.Model):

    class Meta:
        verbose_name = 'household survey'
        abstract = True

    team_assistant = models.ForeignKey('TeamMember', related_name='%(class)s_surveys_as_team_assistant')
    team_anthropometrist = models.ForeignKey('TeamMember', related_name='%(class)s_surveys_as_team_anthropometrist')
    household_number = models.SmallIntegerField()

    json = JSONField(
        # validators=[validate_json(settings.BOWER_COMPONENTS_ROOT
        #                          + '/data-models/schemas/NutritionSurvey.json')],
        null=False,
        blank=False,
        help_text='A JSON document containing data acquired from one '
                  'household.  Typically not edited here but uploaded from a '
                  'mobile application used by a team of surveyors in the '
                  'field.  If in doubt, do not edit.'
    )

    second_admin_level = models.CharField(max_length=60, blank=True)
    first_admin_level = models.CharField(max_length=60, blank=True)
    cluster = models.IntegerField(blank=True, null=True)
    cluster_name = models.CharField(max_length=60, blank=True)
    start_time = models.DateTimeField(blank=True, null=True)
    end_time = models.DateTimeField(blank=True, null=True)
    uuid = models.CharField(
        max_length=255, unique=True,
        help_text='A unique identifier of an individual household survey.  '
                  'Typically assigned by a mobile application before the data '
                  'is uploaded to the server.  If in doubt, do no edit.'
    )

    def parse_team(self, position):

        # TODO this still depends on self.json, while aiming to replace it.

        team_members = self.json['team']['members']
        for m in team_members:
            designation = m['designation']
            if designation == position:
                return m

    def parse_and_set_team_members(self):

        def make_team_member(parsed):
            tm, created = TeamMember.objects.get_or_create(
                member_id=parsed['memberID'],
                defaults={
                    'member_id': parsed['memberID'],
                    'gender': parsed['gender'],
                    'first_name': parsed['firstName'],
                    'last_name': parsed['lastName'],
                    'mobile': parsed['mobile'],
                    'email': parsed['email'],
                    'birth_year': parsed['birthYear']
                }
            )
            return tm

        lead = self.parse_team('Team Leader')
        self.team_lead = make_team_member(lead)

        assistant = self.parse_team('Assistant')
        self.team_assistant = make_team_member(assistant)

        anthro = self.parse_team('Anthropometrist')
        self.team_anthropometrist = make_team_member(anthro)

    def __unicode__(self):
        # Try to build a name describing a survey.
        if not self.json:
            # If field json is invalid then there is no way to compute the name.
            return u'invalid {}'.format(self._meta.verbose_name)
        if 'householdID' in self.json:
            household = self.json['householdID']
        else:
            household = None
        if 'cluster' in self.json:
            cluster = self.json['cluster']
        else:
            cluster = None
        if 'startTime' in self.json:
            start_time = self.json['startTime']
        else:
            start_time = None
        team_name = self.get_team_name()

        return u'cluster: {}; household: {}; team: {}; start time: {}'.format(
            cluster, household, team_name, start_time
        )

    def _get_time_stamp(self, key):
        try:
            time_string = self.json[key]
        except KeyError:
            return None
        try:
            time_stamp = dateutil.parser.parse(time_string)
        except TypeError:
            return None
        else:
            return time_stamp

    def get_start_time(self):
        return self._get_time_stamp('startTime')

    def get_end_time(self):
        return self._get_time_stamp('endTime')

    def get_survey_duration(self):
        """This method calculates duration of the survey in minutes (i.e. the
        difference between the survey end time and its start time).  If any of
        the two time stamps is missing or invalid or the difference is less
        than zero, it returns None.
        """
        start_time = self.get_start_time()
        end_time = self.get_end_time()
        if isinstance(start_time, datetime.datetime) and \
                isinstance(end_time, datetime.datetime):
            delta = end_time - start_time
            if delta.total_seconds() >= 0:
                # Return the survey duration in minutes.
                return delta.total_seconds() / 60.0
            else:
                return None
        else:
            return None

    @classmethod
    def find_all_surveys_by_team(cls, team_id):
        """This method finds all teams buy the team leader id. This is after
        clarification that the team_id is based on the id of th team leader.
        """
        docs = cls.objects.filter(team_lead__id=team_id)
        return docs

    def find_all_surveys_by_this_team(self):
        """This method finds all instances of HouseholdSurveyJSON which were created
        by team who created this instance.  If this document is does not have
        a valid team/team_id or the team_id is None (in both cases the document
        is not a valid survey data structure), it returns an empty list.
        """
        team_id = self.get_team_id()
        if team_id is None:
            return []
        else:
            return self.find_all_surveys_by_team(team_id)

    def get_team_id(self):
        try:
            team_id = self.json['team']['teamID']
        except TypeError:
            return None
        else:
            return team_id

    def get_location(self):
        try:
            location = self.json['location']
        except TypeError:
            return None
        else:
            return location

    def get_cluster_id(self):
        try:
            location = self.json['cluster']
        except TypeError:
            return None
        else:
            return location

    def get_team_leader_name(self):
        """Returns a team leader name or None if no team leader name found.
        """
        last_name = None
        try:
            team_members = self.json['team']['members']
            for m in team_members:
                designation = m['designation']
                if designation is not None:
                    if 'Leader' in designation:
                        last_name = m['lastName']
                        break
        except KeyError:
            return None
        if last_name:
            return last_name
        else:
            return None

    def get_team_name(self):
        """Returns a team leader name or team ID if no team leader name found or
         'UNNAMED'.
        """
        team_leader = self.get_team_leader_name()
        if team_leader is not None:
            return team_leader
        team_id = self.get_team_id()
        if team_id is not None:
            return str(team_id)
        else:
            return 'UNNAMED'

    def _get_records_for_survey_types(self, survey_types):
        if not isinstance(survey_types, (list, tuple, set)) and \
                isinstance(survey_types, basestring):
            survey_types = (survey_types,)
        output = []
        if 'members' in self.json:
            # Get the survey date to calculate age based on date of birth.
            if 'startTime' in self.json:
                survey_start_time = self.json['startTime']
            else:
                survey_start_time = None
            for member in self.json['members']:
                if 'surveyType' in member and member['surveyType'] in survey_types:
                    member['survey_start_time'] = survey_start_time
                    output.append(member)
        return output

    def get_child_records(self):
        return self._get_records_for_survey_types(survey_types='child')

    def get_women_records(self):
        return self._get_records_for_survey_types(survey_types='women')

    def get_women_and_child_records(self):
        return self._get_records_for_survey_types(
            survey_types=('child', 'women')
        )

    @staticmethod
    def get_household_member_gender(household_member):
        if 'gender' in household_member:
            return household_member['gender']
        else:
            return None

    @staticmethod
    def get_household_members_age_in_years(household_member):
        """This function returns a recorded age in years as integer or None if
        no valid age was recorded.
        """
        if 'age' in household_member:
            try:
                age = int(household_member['age'])
            except ValueError:
                return None
            else:
                return age
        else:
            return None

    @staticmethod
    def get_childs_age_in_months(child):
        """This function tries to calculate the child's age in months based on
        their date of birth and the date when the survey was conducted.  If the
        recorded data does not allow that, the function tries to use the age in
        months recorded during the survey.

        N.B. The age in years is not used by this function as a fallback even
        if it was recorded by the surveying team.

        On success, it returns an integer representing the child's age in
        months.
        On failure, it returns None.
        """
        if 'survey' not in child:
            return None
        # Start by trying to calculate the child's age based on their date of
        # birth, if it has been recorder, as it has higher priority than the
        # recorder age in months.
        # On success, the following if-statement FINISHES the function execution
        # and returns a computed value.
        if 'birthDate' in child['survey']:
            dob = child['survey']['birthDate']
            # Try to get the survey start date to use it later for the child's
            # age computations.
            survey_start_time_string = child['survey_start_time']
            try:
                survey_date = dateutil.parser.parse(survey_start_time_string)
            except TypeError:
                # It is impossible to compute the age at the time without
                # knowing when the survey was conducted.
                survey_date = None
            else:
                # if it worked, get rid of the TZ information and time
                survey_date = survey_date.date()

            if survey_date is not None:
                try:
                    dob = dateutil.parser.parse(dob)
                except TypeError:
                    # It is impossible to compute the age at the time without
                    # knowing the person's date of birth.
                    dob = None
                else:
                    # if it worked, get rid of the TZ info and time
                    dob = dob.date()

            if isinstance(dob, datetime.date) and \
                    isinstance(survey_date, datetime.date):
                delta = dateutil.relativedelta.relativedelta(survey_date, dob)
                # return the computed age in months
                return delta.years * 12 + delta.months

        # The following lines should be executed only if the attempt to
        # calculate the child's age based on their date of birth (above) was
        # unsuccessful.
        if 'ageInMonths' not in child['survey']:
            return None

        try:
            age_in_months = int(child['survey']['ageInMonths'])
        except ValueError:
            return None

        return age_in_months

    def get_uuid(self):
        try:
            uuid = self.json['uuid']
        except KeyError:
            return None
        else:
            return uuid


class HouseholdSurveyJSON(BaseHouseholdSurveyJSON):
    team_lead = models.ForeignKey('TeamMember', related_name='%(class)s_as_team_lead')
    point = gismodels.PointField()


class Alert(models.Model):

    """
    New alerts are only created if there is no unarchived alert having the
    same content of the json field.  The user may still create alerts manually
    using the admin interface.
    """

    team_lead = models.ForeignKey('dashboard.TeamMember', null=True, on_delete=models.CASCADE)
    survey = models.ForeignKey('dashboard.HouseholdSurveyJSON', null=True, on_delete=models.CASCADE)
    text = models.TextField()

    json = JSONField(
        null=True, blank=True,
        help_text='A JSON document containing data for one alert.'
    )

    category = models.CharField(max_length=255, default='general')

    created = models.DateTimeField(auto_now_add=True)
    last_modified = models.DateTimeField(auto_now=True)

    completed = models.BooleanField(default=False)
    archived = models.BooleanField(default=False)

    def cluster_id(self):
        return self.json.get('cluster_id')

    def location(self):
        return self.json.get('location')

    def type(self):
        return self.json.get('type')

    def survey_id(self):
        return self.json.get('survey_id')

    def __unicode__(self):
        if self.archived:
            archived = u', archived'
        else:
            archived = u''
        return u'{} (alert #{}{})'.format(self.text, self.pk, archived)

    def get_absolute_url(self):
        return reverse('alert-detail', args=[str(self.id)])

    @classmethod
    def get_or_create_alert(cls, **values):

        # TODO yes, this still contains a race condition, we should really
        # think about how we can use .get_or_create here.

        filter_values = {'archived': False}
        filter_values.update(values)
        filter_values.pop('json')

        if not cls.objects.filter(**filter_values):
            cls.objects.create(**values)

    @classmethod
    def run_alert_checks_on_document(cls, household_survey):
        """This method runs all the defined alert checks which leads to
        relevant alerts being created in case they are triggered by data stored
        in household_survey.
        """
        cls.mapping_check_alert(household_survey)
        cls.sex_ratio_alert(household_survey)
        cls.child_age_in_months_ratio_alert(household_survey)
        cls.child_age_displacement_alert(household_survey)
        cls.woman_age_14_15_displacement_alert(household_survey)
        cls.woman_age_4549_5054_displacement_alert(household_survey)
        cls.digit_preference_alert(household_survey)
        cls.data_collection_time_alert(household_survey)

    @classmethod
    def mapping_check_alert(cls, household_survey):
        """These are four different alerts related to location and cluster id.
        mapping_check_missing_cluster_id: Cluster ID is missing.
        mapping_check_missing_location: Location is missing.
        mapping_check_unknown_cluster: Cluster ID is given, but it's unknown to
            the server
        mapping_check_wrong_location: Cluster ID and location are given and known
            to the server, but the location is not inside the boundaries of the
            cluster that the data supposedly comes from.
        """
        cluster_id = household_survey.get_cluster_id()
        location = household_survey.get_location()
        team_lead = household_survey.team_lead
        team_id = household_survey.get_team_id()
        team_name = household_survey.get_team_name()

        if cluster_id is None:

            alert_text = 'No cluster ID for survey of team {} (survey {})'.format(
                team_id,
                household_survey.uuid)

            alert_json = {
                'type': 'mapping_check_missing_cluster_id',
                'survey_id': household_survey.id,
            }

            if location:
                alert_json['location'] = location

            cls.get_or_create_alert(
                team_lead=team_lead,
                survey=household_survey,
                text=alert_text,
                json=alert_json,
                category='map'
            )

        if location is None:

            alert_text = 'No location for survey of team {} (survey {})'.format(
                team_id,
                household_survey.id)

            alert_json = {
                'type': 'mapping_check_missing_location',
                'team_name': team_name,
                'team_id': team_id,
                'survey_id': household_survey.id,
            }

            if cluster_id:
                alert_json['cluster_id'] = cluster_id

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

        if not (cluster_id and location):
            return

        # get cluster data
        cluster = Clusters.get_cluster_from_active(cluster_id)

        # if cluster data not found, assume location incorrect
        if cluster is None:

            alert_text = 'Unknown cluster ID {} for team {} (survey {})'.format(
                cluster_id,
                team_id,
                household_survey.id)

            alert_json = {
                'type': 'mapping_check_unknown_cluster',
                'team_name': team_name,
                'team_id': team_id,
                'cluster_id': cluster_id,
                'survey_id': household_survey.id,
                'location': location
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

            return

        # if cluster data found, get first and second admin level
        second_admin_level_name = cluster.get('second_admin_level_name', None)
        first_admin_level_name = cluster.get('first_admin_level_name', None)

        # if first and second admin level names not found, assume database inconsistencies and abort
        if not (first_admin_level_name and second_admin_level_name):
            return False

        # if first and second admin level found, check the location
        second_admin_level = SecondAdminLevel.find_second_admin_level(
            name=second_admin_level_name, first_admin_level_name=first_admin_level_name)

        if second_admin_level is None:
            # if no second admin level found, assume database inconsistencies and abort
            return False

        if not second_admin_level.contains_location(location):
            alert_text = 'Wrong location for team {} (survey {})'.format(team_id,
                                                                         household_survey.id)
            alert_json = {
                'type': 'mapping_check_wrong_location',
                'team_name': team_name,
                'team_id': team_id,
                'cluster_id': cluster_id,
                'second_admin_level_name': second_admin_level_name,
                'first_admin_level_name': first_admin_level_name,
                'survey_id': household_survey.id,
                'location': location
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

    @classmethod
    def sex_ratio_alert(cls, household_survey, test='chi-squared'):
        """In the data of children from 0-59 months of age, we expect to find
        an even number boys and girls. If chi-square of sex ratio < 0.001 then
        report an alert on dashboard "Sex-ratio issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """

        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []

        for survey in surveys:
            children.extend(survey.get_child_records())

        boys = 0
        girls = 0
        for child in children:
            age = HouseholdSurveyJSON.get_childs_age_in_months(child)
            if age is None:
                continue
            elif age < 60:
                gender = HouseholdSurveyJSON.get_household_member_gender(child)
                if gender == 'M':
                    boys += 1
                elif gender == 'F':
                    girls += 1

        if boys + girls == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return

        if test == 'binomial':
            p = scipy.stats.binom_test([boys, girls], p=0.5)
        else:
            expected = (boys + girls) / 2.0
            chi2, p = scipy.stats.chisquare([boys, girls], [expected, expected])

        if p < 0.001:

            team_name = household_survey.get_team_name()

            team_id = household_survey.get_team_id()

            alert_text = 'Sex ratio issue in team {}'.format(team_id)

            alert_json = {
                'type': 'sex_ratio',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='sex',
            )

    @classmethod
    def child_age_in_months_ratio_alert(cls, household_survey,
                                        test='chi-squared'):
        """In the data of children from 6-59 months of age, we expect an age
        ratio of 6-29 months to 30-59 months to be around 0.85. If a chi-square
        test of the age ratio 6-29 months / 30-59 months is significantly
        < 0.001 from expected 0.85 then report an alert on dashboard "Age ratio
        issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """

        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []

        for survey in surveys:
            children.extend(survey.get_child_records())

        age6to29 = 0
        age30to59 = 0
        for child in children:
            age = HouseholdSurveyJSON.get_childs_age_in_months(child)
            if age is None:
                continue
            elif 5 < age < 30:
                age6to29 += 1
            elif 29 < age < 60:
                age30to59 += 1

        # The theoretical probability of the 'positive' outcome (i.e. the
        # child being 6 to 29 months old).
        p629 = 0.45945945945945954
        # The theoretical probability of the 'negative' outcome (i.e. the
        # child being 30 to 59 months old).
        # p3059 = 1 - p629 = 0.5405405405405405

        if age6to29 + age30to59 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age6to29, age30to59], p=p629)
        else:
            expected6to29 = (age6to29 + age30to59) * p629
            expected30to59 = (age6to29 + age30to59) - expected6to29
            chi2, p = scipy.stats.chisquare(
                [age6to29, age30to59],
                [expected6to29, expected30to59]
            )

        if p < 0.001:

            team_name = household_survey.get_team_name()

            team_id = household_survey.get_team_id()

            alert_text = 'Age ratio issue in team {}'.format(team_id)

            alert_json = {
                'type': 'child_age_in_months_ratio',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution'
            )

    @classmethod
    def child_age_displacement_alert(cls, household_survey, test='chi-squared'):
        """Ratio of children aged 5 years / children aged 4 years is expected
        to equal 1. If the chi-square of this ratio is significantly < 0.001
        from expected 1 then report an alert on dashboard "Child age
        displacement issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []
        for survey in surveys:
            children.extend(survey.get_child_records())
        age4 = 0
        age5 = 0
        for child in children:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(child)
            if age is None:
                continue
            elif age == 4:
                age4 += 1
            elif age == 5:
                age5 += 1

        if age4 + age5 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age4, age5], p=0.5)
        else:
            expected = (age4 + age5) / 2.0
            chi2, p = scipy.stats.chisquare([age4, age5], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = 'Child age displacement issue in team {}'.format(team_id)
            alert_json = {
                'type': 'child_age_displacement',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution'
            )

    @classmethod
    def woman_age_14_15_displacement_alert(
            cls, household_survey, test='chi-squared'):
        """Ratio of women aged 14 / women aged 15 is expected to equal 1. If
        the chi-square of this ratio is significantly < 0.001 from expected 1
        then report an alert on dashboard "Woman age displacement issue (14/15)
        in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        women = []
        for survey in surveys:
            women.extend(survey.get_women_records())
        age14 = 0
        age15 = 0
        for woman in women:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(woman)
            if age is None:
                continue
            elif age == 14:
                age14 += 1
            elif age == 15:
                age15 += 1

        if age14 + age15 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age14, age15], p=0.5)
        else:
            expected = (age14 + age15) / 2.0
            chi2, p = scipy.stats.chisquare(
                [age14, age15], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = \
                'Woman age displacement issue (14/15) in team {}'.format(team_id)
            alert_json = {
                'type': 'woman_age_14_15_displacement',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution',
            )

    @classmethod
    def woman_age_4549_5054_displacement_alert(cls, household_survey,
                                               test='chi-squared'):
        """Ratio of women aged 45-49 / women aged 50-54 is expected to equal 1.
        If the chi-square of this ratio is significantly < 0.001 from expected
        1 then report an alert on dashboard "Woman age displacement issue
        (45-49/50-54) in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        women = []
        for survey in surveys:
            women.extend(survey.get_women_records())
        age4549 = 0
        age5054 = 0

        for woman in women:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(woman)
            if age is None:
                continue
            elif 44 < age < 50:
                age4549 += 1
            elif 49 < age < 55:
                age5054 += 1

        if age4549 + age5054 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age4549, age5054], p=0.5)
        else:
            expected = (age4549 + age5054) / 2.0
            chi2, p = scipy.stats.chisquare(
                [age4549, age5054], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = 'Woman age displacement issue (45-49/50-54) in team ' \
                         '{}'.format(team_id)
            alert_json = {
                'type': 'woman_age_4549_5054_displacement',
                'team_id': team_id,
                'team_name': team_name
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution',
            )

    @classmethod
    def digit_preference_alert(cls, household_survey):
        """If terminal digit preference score for weight, height or MUAC > 20,
        then report alert on dashboard "Digit preference issue in Team NAME".
        N.B. this function calculates and checks TDPS for each variable (i.e.
        weight, height and MUAC) independently.  If only one of the calculated
        terminal digit preferences scores satisfies the condition then the
        alert is triggered.
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        data_points = {'muac': [], 'weight': [], 'height': []}
        for survey in surveys:
            subjects = survey.get_women_and_child_records()
            for subject in subjects:
                # If there is no survey section then there is no data for this
                # subject.
                if 'survey' not in subject:
                    continue
                # Add the subject's muac, weight and height to data_points.
                # Append None where the value is missing.
                for k in data_points:
                    data_points[k].append(subject['survey'].get(k, None))

        # Extract and count terminal digits for each variable of interest.
        terminal_digit_counts = {
            'muac': [0] * 10,
            'weight': [0] * 10,
            'height': [0] * 10
        }
        terminal_digit_preference_score = {
            'muac': 0,
            'weight': 0,
            'height': 0
        }
        for k in terminal_digit_counts:
            for v in data_points[k]:
                try:
                    v = float(v)
                except (TypeError, ValueError):
                    # Get rid of all None objects and other invalid values.
                    continue
                else:
                    terminal_digit = int(
                        ('{:.1f}'.format(float(int(v * 10) / 10.0)))[-1]
                    )
                    terminal_digit_counts[k][terminal_digit] += 1
        # Calculate terminal digit preference scores
        for k in terminal_digit_preference_score:
            total_number_of_digits = sum(terminal_digit_counts[k])
            if total_number_of_digits > 0:
                expected = total_number_of_digits / 10.0
                chi2 = sum(
                    [
                        (i - expected) ** 2 / expected
                        for i in terminal_digit_counts[k]
                    ]
                )
                terminal_digit_preference_score[k] =\
                    100 * (chi2 / (9 * total_number_of_digits)) ** 0.5

        # Check if any of the computed scores triggers the alert.
        for k in terminal_digit_preference_score:
            if terminal_digit_preference_score[k] > 20:
                team_name = household_survey.get_team_name()
                team_id = household_survey.get_team_id()
                alert_text = 'Digit preference issue in team {}'.format(team_id)
                alert_json = {
                    'type': 'digit_preference',
                    'team_id': team_id,
                    'team_name': team_name
                }

                cls.get_or_create_alert(
                    team_lead=team_lead,
                    text=alert_text,
                    json=alert_json,
                    category='number_distribution'
                )

                # Only one alert should be emitted so no need to finish the
                # loop.
                break

    @classmethod
    def data_collection_time_alert(cls, household_survey):
        """If timestamp for any data point in data collection is <07:00 hours
        or >20:00 hours then report alert on dashboard "Data collection time
        issue in team NAME (survey: UUID)".
        If any of the time stamps is not a valid date, the alert is triggered
        too.
        N.B. This function performs these checks for both the start time and
        the end time.  The alert is triggered if any of them satisfies the
        condition mentioned above.
        """
        t700h = datetime.time(7)
        t2000h = datetime.time(20)
        start = household_survey.get_start_time()
        end = household_survey.get_end_time()
        triggered = False

        for t in (start, end):
            if t is None:
                triggered = True
                break
            elif t.time() < t700h or t.time() > t2000h:
                triggered = True
                break

        if triggered:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            team_lead = household_survey.team_lead
            alert_text = u'Data collection time issue in team {} (survey: {})'.\
                format(team_id, household_survey.id)
<commit_msg>6:00 AM is the new 7:00 AM (spec adjustment)<commit_after># Python
import datetime
import dateutil.parser
import dateutil.relativedelta
import uuid
import random
from functools import wraps
import logging

# 3rd party
import numpy
import scipy.stats
import validictory
import json

# django core
from django.utils.translation import ugettext as _
import django.contrib.gis.db.models as gismodels
from django.db import models
from django.contrib.gis.geos import Point
from django.db.models import Q


# django 3rd party
from rest_framework.reverse import reverse
from phonenumber_field.modelfields import PhoneNumberField
from jsonfield import JSONField
from django_extensions.db.fields import (
    AutoSlugField, CreationDateTimeField, ModificationDateTimeField,
)

# Internal
from fields import UniqueActiveField


logger = logging.getLogger(__name__)


class TeamMember(models.Model):
    MALE = 'M'
    FEMALE = 'F'
    GENDER_CHOICES = (
        (MALE, _('Male')),
        (FEMALE, _('Female')),
    )
    member_id = AutoSlugField(blank=False,
                              unique=True,
                              populate_from="id",
                              separator='')
    first_name = models.CharField(blank=False, max_length=50)
    last_name = models.CharField(blank=False, max_length=50)
    mobile = PhoneNumberField(blank=True)
    email = models.EmailField(blank=True)
    birth_year = models.IntegerField(blank=False)
    gender = models.CharField(choices=GENDER_CHOICES, max_length=3, blank=True)

    created = CreationDateTimeField()
    modified = ModificationDateTimeField()

    _last_survey = None

    def last_survey(self):
        # TODO: maybe get the reverse relationship instead

        # Poor mans cache
        if self._last_survey:
            return self._last_survey

        survey = (HouseholdSurveyJSON.objects.filter(
            Q(team_lead=self)
            | Q(team_assistant=self)
            | Q(team_anthropometrist=self)
        ).order_by('-id') or [None])[0]

        self._last_survey = survey
        return survey

    def last_survey_position(self):
        last_survey = self.last_survey()
        if last_survey:
            for position in ['team_lead', 'team_assistant', 'team_anthropometrist']:
                if self.id == getattr(last_survey, position).id:
                    return position

            raise "TeamMember has last_survey but no position, this shouldn't happen."

    def last_survey_created(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.get_start_time()

    def last_survey_cluster_name(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.json['cluster_name']

    def last_survey_cluster(self):
        last_survey = self.last_survey()
        if last_survey:
            return last_survey.json['cluster']

    class Meta:
        get_latest_by = 'modified'
        ordering = ('-modified', '-created',)

    @property
    def random_id(self):
        return random.randint(10000, 100000)

    def __unicode__(self):
        return u'%s-%s %s' % (self.id, self.first_name, self.last_name)

    def get_absolute_url(self):
        return reverse('teammember-detail', args=[str(self.member_id)])


def validate_json(spec_file):
    survey_schema = None
    try:
        survey_schema = json.load(open(spec_file))
    except IOError as e:
        # log the error but it's ok if the spec is missing
        # if the validator is never called
        logger.exception(
            "Could not load the json spec %s. \nThis is ok if you are just setting up.\n%s" % (spec_file, e.message))

    @wraps(validate_json)
    def wrapped(value):
        assert survey_schema, "Trying to validate a non existant JSON schema file %s" % spec_file
        # It really should exist by now.
        validictory.validate(value, survey_schema, required_by_default=False)
    return wrapped


class BaseHouseholdMember(models.Model):

    GENDER = (
        ('M', 'Male'),
        ('F', 'Female'),
        ('O', 'Other'),
    )
    gender = models.CharField(max_length=1, choices=GENDER, blank=True)

    first_name = models.TextField(blank=True)
    index = models.SmallIntegerField(blank=True)
    birthdate = models.DateField(blank=True, null=True)

    muac = models.SmallIntegerField(null=True)  # in millimeter ?
    weight = models.FloatField(null=True)       # probably in kilograms ?
    height = models.FloatField(null=True)       # probably in centimeters ?
    edema = models.NullBooleanField()

    class Meta:
        abstract = True


class HouseholdMember(BaseHouseholdMember):
    household_survey = models.ForeignKey('HouseholdSurveyJSON', related_name='members')


class BaseHouseholdSurveyJSON(gismodels.Model):

    class Meta:
        verbose_name = 'household survey'
        abstract = True

    team_assistant = models.ForeignKey('TeamMember', related_name='%(class)s_surveys_as_team_assistant')
    team_anthropometrist = models.ForeignKey('TeamMember', related_name='%(class)s_surveys_as_team_anthropometrist')
    household_number = models.SmallIntegerField()

    json = JSONField(
        # validators=[validate_json(settings.BOWER_COMPONENTS_ROOT
        #                          + '/data-models/schemas/NutritionSurvey.json')],
        null=False,
        blank=False,
        help_text='A JSON document containing data acquired from one '
                  'household.  Typically not edited here but uploaded from a '
                  'mobile application used by a team of surveyors in the '
                  'field.  If in doubt, do not edit.'
    )

    second_admin_level = models.CharField(max_length=60, blank=True)
    first_admin_level = models.CharField(max_length=60, blank=True)
    cluster = models.IntegerField(blank=True, null=True)
    cluster_name = models.CharField(max_length=60, blank=True)
    start_time = models.DateTimeField(blank=True, null=True)
    end_time = models.DateTimeField(blank=True, null=True)
    uuid = models.CharField(
        max_length=255, unique=True,
        help_text='A unique identifier of an individual household survey.  '
                  'Typically assigned by a mobile application before the data '
                  'is uploaded to the server.  If in doubt, do no edit.'
    )

    def parse_team(self, position):

        # TODO this still depends on self.json, while aiming to replace it.

        team_members = self.json['team']['members']
        for m in team_members:
            designation = m['designation']
            if designation == position:
                return m

    def parse_and_set_team_members(self):

        def make_team_member(parsed):
            tm, created = TeamMember.objects.get_or_create(
                member_id=parsed['memberID'],
                defaults={
                    'member_id': parsed['memberID'],
                    'gender': parsed['gender'],
                    'first_name': parsed['firstName'],
                    'last_name': parsed['lastName'],
                    'mobile': parsed['mobile'],
                    'email': parsed['email'],
                    'birth_year': parsed['birthYear']
                }
            )
            return tm

        lead = self.parse_team('Team Leader')
        self.team_lead = make_team_member(lead)

        assistant = self.parse_team('Assistant')
        self.team_assistant = make_team_member(assistant)

        anthro = self.parse_team('Anthropometrist')
        self.team_anthropometrist = make_team_member(anthro)

    def __unicode__(self):
        # Try to build a name describing a survey.
        if not self.json:
            # If field json is invalid then there is no way to compute the name.
            return u'invalid {}'.format(self._meta.verbose_name)
        if 'householdID' in self.json:
            household = self.json['householdID']
        else:
            household = None
        if 'cluster' in self.json:
            cluster = self.json['cluster']
        else:
            cluster = None
        if 'startTime' in self.json:
            start_time = self.json['startTime']
        else:
            start_time = None
        team_name = self.get_team_name()

        return u'cluster: {}; household: {}; team: {}; start time: {}'.format(
            cluster, household, team_name, start_time
        )

    def _get_time_stamp(self, key):
        try:
            time_string = self.json[key]
        except KeyError:
            return None
        try:
            time_stamp = dateutil.parser.parse(time_string)
        except TypeError:
            return None
        else:
            return time_stamp

    def get_start_time(self):
        return self._get_time_stamp('startTime')

    def get_end_time(self):
        return self._get_time_stamp('endTime')

    def get_survey_duration(self):
        """This method calculates duration of the survey in minutes (i.e. the
        difference between the survey end time and its start time).  If any of
        the two time stamps is missing or invalid or the difference is less
        than zero, it returns None.
        """
        start_time = self.get_start_time()
        end_time = self.get_end_time()
        if isinstance(start_time, datetime.datetime) and \
                isinstance(end_time, datetime.datetime):
            delta = end_time - start_time
            if delta.total_seconds() >= 0:
                # Return the survey duration in minutes.
                return delta.total_seconds() / 60.0
            else:
                return None
        else:
            return None

    @classmethod
    def find_all_surveys_by_team(cls, team_id):
        """This method finds all teams buy the team leader id. This is after
        clarification that the team_id is based on the id of th team leader.
        """
        docs = cls.objects.filter(team_lead__id=team_id)
        return docs

    def find_all_surveys_by_this_team(self):
        """This method finds all instances of HouseholdSurveyJSON which were created
        by team who created this instance.  If this document is does not have
        a valid team/team_id or the team_id is None (in both cases the document
        is not a valid survey data structure), it returns an empty list.
        """
        team_id = self.get_team_id()
        if team_id is None:
            return []
        else:
            return self.find_all_surveys_by_team(team_id)

    def get_team_id(self):
        try:
            team_id = self.json['team']['teamID']
        except TypeError:
            return None
        else:
            return team_id

    def get_location(self):
        try:
            location = self.json['location']
        except TypeError:
            return None
        else:
            return location

    def get_cluster_id(self):
        try:
            location = self.json['cluster']
        except TypeError:
            return None
        else:
            return location

    def get_team_leader_name(self):
        """Returns a team leader name or None if no team leader name found.
        """
        last_name = None
        try:
            team_members = self.json['team']['members']
            for m in team_members:
                designation = m['designation']
                if designation is not None:
                    if 'Leader' in designation:
                        last_name = m['lastName']
                        break
        except KeyError:
            return None
        if last_name:
            return last_name
        else:
            return None

    def get_team_name(self):
        """Returns a team leader name or team ID if no team leader name found or
         'UNNAMED'.
        """
        team_leader = self.get_team_leader_name()
        if team_leader is not None:
            return team_leader
        team_id = self.get_team_id()
        if team_id is not None:
            return str(team_id)
        else:
            return 'UNNAMED'

    def _get_records_for_survey_types(self, survey_types):
        if not isinstance(survey_types, (list, tuple, set)) and \
                isinstance(survey_types, basestring):
            survey_types = (survey_types,)
        output = []
        if 'members' in self.json:
            # Get the survey date to calculate age based on date of birth.
            if 'startTime' in self.json:
                survey_start_time = self.json['startTime']
            else:
                survey_start_time = None
            for member in self.json['members']:
                if 'surveyType' in member and member['surveyType'] in survey_types:
                    member['survey_start_time'] = survey_start_time
                    output.append(member)
        return output

    def get_child_records(self):
        return self._get_records_for_survey_types(survey_types='child')

    def get_women_records(self):
        return self._get_records_for_survey_types(survey_types='women')

    def get_women_and_child_records(self):
        return self._get_records_for_survey_types(
            survey_types=('child', 'women')
        )

    @staticmethod
    def get_household_member_gender(household_member):
        if 'gender' in household_member:
            return household_member['gender']
        else:
            return None

    @staticmethod
    def get_household_members_age_in_years(household_member):
        """This function returns a recorded age in years as integer or None if
        no valid age was recorded.
        """
        if 'age' in household_member:
            try:
                age = int(household_member['age'])
            except ValueError:
                return None
            else:
                return age
        else:
            return None

    @staticmethod
    def get_childs_age_in_months(child):
        """This function tries to calculate the child's age in months based on
        their date of birth and the date when the survey was conducted.  If the
        recorded data does not allow that, the function tries to use the age in
        months recorded during the survey.

        N.B. The age in years is not used by this function as a fallback even
        if it was recorded by the surveying team.

        On success, it returns an integer representing the child's age in
        months.
        On failure, it returns None.
        """
        if 'survey' not in child:
            return None
        # Start by trying to calculate the child's age based on their date of
        # birth, if it has been recorder, as it has higher priority than the
        # recorder age in months.
        # On success, the following if-statement FINISHES the function execution
        # and returns a computed value.
        if 'birthDate' in child['survey']:
            dob = child['survey']['birthDate']
            # Try to get the survey start date to use it later for the child's
            # age computations.
            survey_start_time_string = child['survey_start_time']
            try:
                survey_date = dateutil.parser.parse(survey_start_time_string)
            except TypeError:
                # It is impossible to compute the age at the time without
                # knowing when the survey was conducted.
                survey_date = None
            else:
                # if it worked, get rid of the TZ information and time
                survey_date = survey_date.date()

            if survey_date is not None:
                try:
                    dob = dateutil.parser.parse(dob)
                except TypeError:
                    # It is impossible to compute the age at the time without
                    # knowing the person's date of birth.
                    dob = None
                else:
                    # if it worked, get rid of the TZ info and time
                    dob = dob.date()

            if isinstance(dob, datetime.date) and \
                    isinstance(survey_date, datetime.date):
                delta = dateutil.relativedelta.relativedelta(survey_date, dob)
                # return the computed age in months
                return delta.years * 12 + delta.months

        # The following lines should be executed only if the attempt to
        # calculate the child's age based on their date of birth (above) was
        # unsuccessful.
        if 'ageInMonths' not in child['survey']:
            return None

        try:
            age_in_months = int(child['survey']['ageInMonths'])
        except ValueError:
            return None

        return age_in_months

    def get_uuid(self):
        try:
            uuid = self.json['uuid']
        except KeyError:
            return None
        else:
            return uuid


class HouseholdSurveyJSON(BaseHouseholdSurveyJSON):
    team_lead = models.ForeignKey('TeamMember', related_name='%(class)s_as_team_lead')
    point = gismodels.PointField()


class Alert(models.Model):

    """
    New alerts are only created if there is no unarchived alert having the
    same content of the json field.  The user may still create alerts manually
    using the admin interface.
    """

    team_lead = models.ForeignKey('dashboard.TeamMember', null=True, on_delete=models.CASCADE)
    survey = models.ForeignKey('dashboard.HouseholdSurveyJSON', null=True, on_delete=models.CASCADE)
    text = models.TextField()

    json = JSONField(
        null=True, blank=True,
        help_text='A JSON document containing data for one alert.'
    )

    category = models.CharField(max_length=255, default='general')

    created = models.DateTimeField(auto_now_add=True)
    last_modified = models.DateTimeField(auto_now=True)

    completed = models.BooleanField(default=False)
    archived = models.BooleanField(default=False)

    def cluster_id(self):
        return self.json.get('cluster_id')

    def location(self):
        return self.json.get('location')

    def type(self):
        return self.json.get('type')

    def survey_id(self):
        return self.json.get('survey_id')

    def __unicode__(self):
        if self.archived:
            archived = u', archived'
        else:
            archived = u''
        return u'{} (alert #{}{})'.format(self.text, self.pk, archived)

    def get_absolute_url(self):
        return reverse('alert-detail', args=[str(self.id)])

    @classmethod
    def get_or_create_alert(cls, **values):

        # TODO yes, this still contains a race condition, we should really
        # think about how we can use .get_or_create here.

        filter_values = {'archived': False}
        filter_values.update(values)
        filter_values.pop('json')

        if not cls.objects.filter(**filter_values):
            cls.objects.create(**values)

    @classmethod
    def run_alert_checks_on_document(cls, household_survey):
        """This method runs all the defined alert checks which leads to
        relevant alerts being created in case they are triggered by data stored
        in household_survey.
        """
        cls.mapping_check_alert(household_survey)
        cls.sex_ratio_alert(household_survey)
        cls.child_age_in_months_ratio_alert(household_survey)
        cls.child_age_displacement_alert(household_survey)
        cls.woman_age_14_15_displacement_alert(household_survey)
        cls.woman_age_4549_5054_displacement_alert(household_survey)
        cls.digit_preference_alert(household_survey)
        cls.data_collection_time_alert(household_survey)

    @classmethod
    def mapping_check_alert(cls, household_survey):
        """These are four different alerts related to location and cluster id.
        mapping_check_missing_cluster_id: Cluster ID is missing.
        mapping_check_missing_location: Location is missing.
        mapping_check_unknown_cluster: Cluster ID is given, but it's unknown to
            the server
        mapping_check_wrong_location: Cluster ID and location are given and known
            to the server, but the location is not inside the boundaries of the
            cluster that the data supposedly comes from.
        """
        cluster_id = household_survey.get_cluster_id()
        location = household_survey.get_location()
        team_lead = household_survey.team_lead
        team_id = household_survey.get_team_id()
        team_name = household_survey.get_team_name()

        if cluster_id is None:

            alert_text = 'No cluster ID for survey of team {} (survey {})'.format(
                team_id,
                household_survey.uuid)

            alert_json = {
                'type': 'mapping_check_missing_cluster_id',
                'survey_id': household_survey.id,
            }

            if location:
                alert_json['location'] = location

            cls.get_or_create_alert(
                team_lead=team_lead,
                survey=household_survey,
                text=alert_text,
                json=alert_json,
                category='map'
            )

        if location is None:

            alert_text = 'No location for survey of team {} (survey {})'.format(
                team_id,
                household_survey.id)

            alert_json = {
                'type': 'mapping_check_missing_location',
                'team_name': team_name,
                'team_id': team_id,
                'survey_id': household_survey.id,
            }

            if cluster_id:
                alert_json['cluster_id'] = cluster_id

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

        if not (cluster_id and location):
            return

        # get cluster data
        cluster = Clusters.get_cluster_from_active(cluster_id)

        # if cluster data not found, assume location incorrect
        if cluster is None:

            alert_text = 'Unknown cluster ID {} for team {} (survey {})'.format(
                cluster_id,
                team_id,
                household_survey.id)

            alert_json = {
                'type': 'mapping_check_unknown_cluster',
                'team_name': team_name,
                'team_id': team_id,
                'cluster_id': cluster_id,
                'survey_id': household_survey.id,
                'location': location
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

            return

        # if cluster data found, get first and second admin level
        second_admin_level_name = cluster.get('second_admin_level_name', None)
        first_admin_level_name = cluster.get('first_admin_level_name', None)

        # if first and second admin level names not found, assume database inconsistencies and abort
        if not (first_admin_level_name and second_admin_level_name):
            return False

        # if first and second admin level found, check the location
        second_admin_level = SecondAdminLevel.find_second_admin_level(
            name=second_admin_level_name, first_admin_level_name=first_admin_level_name)

        if second_admin_level is None:
            # if no second admin level found, assume database inconsistencies and abort
            return False

        if not second_admin_level.contains_location(location):
            alert_text = 'Wrong location for team {} (survey {})'.format(team_id,
                                                                         household_survey.id)
            alert_json = {
                'type': 'mapping_check_wrong_location',
                'team_name': team_name,
                'team_id': team_id,
                'cluster_id': cluster_id,
                'second_admin_level_name': second_admin_level_name,
                'first_admin_level_name': first_admin_level_name,
                'survey_id': household_survey.id,
                'location': location
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='map'
            )

    @classmethod
    def sex_ratio_alert(cls, household_survey, test='chi-squared'):
        """In the data of children from 0-59 months of age, we expect to find
        an even number boys and girls. If chi-square of sex ratio < 0.001 then
        report an alert on dashboard "Sex-ratio issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """

        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []

        for survey in surveys:
            children.extend(survey.get_child_records())

        boys = 0
        girls = 0
        for child in children:
            age = HouseholdSurveyJSON.get_childs_age_in_months(child)
            if age is None:
                continue
            elif age < 60:
                gender = HouseholdSurveyJSON.get_household_member_gender(child)
                if gender == 'M':
                    boys += 1
                elif gender == 'F':
                    girls += 1

        if boys + girls == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return

        if test == 'binomial':
            p = scipy.stats.binom_test([boys, girls], p=0.5)
        else:
            expected = (boys + girls) / 2.0
            chi2, p = scipy.stats.chisquare([boys, girls], [expected, expected])

        if p < 0.001:

            team_name = household_survey.get_team_name()

            team_id = household_survey.get_team_id()

            alert_text = 'Sex ratio issue in team {}'.format(team_id)

            alert_json = {
                'type': 'sex_ratio',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='sex',
            )

    @classmethod
    def child_age_in_months_ratio_alert(cls, household_survey,
                                        test='chi-squared'):
        """In the data of children from 6-59 months of age, we expect an age
        ratio of 6-29 months to 30-59 months to be around 0.85. If a chi-square
        test of the age ratio 6-29 months / 30-59 months is significantly
        < 0.001 from expected 0.85 then report an alert on dashboard "Age ratio
        issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """

        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []

        for survey in surveys:
            children.extend(survey.get_child_records())

        age6to29 = 0
        age30to59 = 0
        for child in children:
            age = HouseholdSurveyJSON.get_childs_age_in_months(child)
            if age is None:
                continue
            elif 5 < age < 30:
                age6to29 += 1
            elif 29 < age < 60:
                age30to59 += 1

        # The theoretical probability of the 'positive' outcome (i.e. the
        # child being 6 to 29 months old).
        p629 = 0.45945945945945954
        # The theoretical probability of the 'negative' outcome (i.e. the
        # child being 30 to 59 months old).
        # p3059 = 1 - p629 = 0.5405405405405405

        if age6to29 + age30to59 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age6to29, age30to59], p=p629)
        else:
            expected6to29 = (age6to29 + age30to59) * p629
            expected30to59 = (age6to29 + age30to59) - expected6to29
            chi2, p = scipy.stats.chisquare(
                [age6to29, age30to59],
                [expected6to29, expected30to59]
            )

        if p < 0.001:

            team_name = household_survey.get_team_name()

            team_id = household_survey.get_team_id()

            alert_text = 'Age ratio issue in team {}'.format(team_id)

            alert_json = {
                'type': 'child_age_in_months_ratio',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution'
            )

    @classmethod
    def child_age_displacement_alert(cls, household_survey, test='chi-squared'):
        """Ratio of children aged 5 years / children aged 4 years is expected
        to equal 1. If the chi-square of this ratio is significantly < 0.001
        from expected 1 then report an alert on dashboard "Child age
        displacement issue in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        children = []
        for survey in surveys:
            children.extend(survey.get_child_records())
        age4 = 0
        age5 = 0
        for child in children:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(child)
            if age is None:
                continue
            elif age == 4:
                age4 += 1
            elif age == 5:
                age5 += 1

        if age4 + age5 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age4, age5], p=0.5)
        else:
            expected = (age4 + age5) / 2.0
            chi2, p = scipy.stats.chisquare([age4, age5], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = 'Child age displacement issue in team {}'.format(team_id)
            alert_json = {
                'type': 'child_age_displacement',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution'
            )

    @classmethod
    def woman_age_14_15_displacement_alert(
            cls, household_survey, test='chi-squared'):
        """Ratio of women aged 14 / women aged 15 is expected to equal 1. If
        the chi-square of this ratio is significantly < 0.001 from expected 1
        then report an alert on dashboard "Woman age displacement issue (14/15)
        in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        women = []
        for survey in surveys:
            women.extend(survey.get_women_records())
        age14 = 0
        age15 = 0
        for woman in women:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(woman)
            if age is None:
                continue
            elif age == 14:
                age14 += 1
            elif age == 15:
                age15 += 1

        if age14 + age15 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age14, age15], p=0.5)
        else:
            expected = (age14 + age15) / 2.0
            chi2, p = scipy.stats.chisquare(
                [age14, age15], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = \
                'Woman age displacement issue (14/15) in team {}'.format(team_id)
            alert_json = {
                'type': 'woman_age_14_15_displacement',
                'team_name': team_name,
                'team_id': team_id
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution',
            )

    @classmethod
    def woman_age_4549_5054_displacement_alert(cls, household_survey,
                                               test='chi-squared'):
        """Ratio of women aged 45-49 / women aged 50-54 is expected to equal 1.
        If the chi-square of this ratio is significantly < 0.001 from expected
        1 then report an alert on dashboard "Woman age displacement issue
        (45-49/50-54) in team NAME".
        This method implements both binomial two-tailed test and chi-square
        test.  The latter is the default (as per client's request).
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        women = []
        for survey in surveys:
            women.extend(survey.get_women_records())
        age4549 = 0
        age5054 = 0

        for woman in women:
            age = HouseholdSurveyJSON.get_household_members_age_in_years(woman)
            if age is None:
                continue
            elif 44 < age < 50:
                age4549 += 1
            elif 49 < age < 55:
                age5054 += 1

        if age4549 + age5054 == 0:
            # Chi-square impossible to compute (expected value is 0 which would
            # cause division by zero) and binomial equals 1 so no alert
            # necessary.
            return
        if test == 'binomial':
            p = scipy.stats.binom_test([age4549, age5054], p=0.5)
        else:
            expected = (age4549 + age5054) / 2.0
            chi2, p = scipy.stats.chisquare(
                [age4549, age5054], [expected, expected])
        if p < 0.001:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            alert_text = 'Woman age displacement issue (45-49/50-54) in team ' \
                         '{}'.format(team_id)
            alert_json = {
                'type': 'woman_age_4549_5054_displacement',
                'team_id': team_id,
                'team_name': team_name
            }

            cls.get_or_create_alert(
                team_lead=team_lead,
                text=alert_text,
                json=alert_json,
                category='age_distribution',
            )

    @classmethod
    def digit_preference_alert(cls, household_survey):
        """If terminal digit preference score for weight, height or MUAC > 20,
        then report alert on dashboard "Digit preference issue in Team NAME".
        N.B. this function calculates and checks TDPS for each variable (i.e.
        weight, height and MUAC) independently.  If only one of the calculated
        terminal digit preferences scores satisfies the condition then the
        alert is triggered.
        """
        surveys = household_survey.find_all_surveys_by_this_team()
        team_lead = household_survey.team_lead
        data_points = {'muac': [], 'weight': [], 'height': []}
        for survey in surveys:
            subjects = survey.get_women_and_child_records()
            for subject in subjects:
                # If there is no survey section then there is no data for this
                # subject.
                if 'survey' not in subject:
                    continue
                # Add the subject's muac, weight and height to data_points.
                # Append None where the value is missing.
                for k in data_points:
                    data_points[k].append(subject['survey'].get(k, None))

        # Extract and count terminal digits for each variable of interest.
        terminal_digit_counts = {
            'muac': [0] * 10,
            'weight': [0] * 10,
            'height': [0] * 10
        }
        terminal_digit_preference_score = {
            'muac': 0,
            'weight': 0,
            'height': 0
        }
        for k in terminal_digit_counts:
            for v in data_points[k]:
                try:
                    v = float(v)
                except (TypeError, ValueError):
                    # Get rid of all None objects and other invalid values.
                    continue
                else:
                    terminal_digit = int(
                        ('{:.1f}'.format(float(int(v * 10) / 10.0)))[-1]
                    )
                    terminal_digit_counts[k][terminal_digit] += 1
        # Calculate terminal digit preference scores
        for k in terminal_digit_preference_score:
            total_number_of_digits = sum(terminal_digit_counts[k])
            if total_number_of_digits > 0:
                expected = total_number_of_digits / 10.0
                chi2 = sum(
                    [
                        (i - expected) ** 2 / expected
                        for i in terminal_digit_counts[k]
                    ]
                )
                terminal_digit_preference_score[k] =\
                    100 * (chi2 / (9 * total_number_of_digits)) ** 0.5

        # Check if any of the computed scores triggers the alert.
        for k in terminal_digit_preference_score:
            if terminal_digit_preference_score[k] > 20:
                team_name = household_survey.get_team_name()
                team_id = household_survey.get_team_id()
                alert_text = 'Digit preference issue in team {}'.format(team_id)
                alert_json = {
                    'type': 'digit_preference',
                    'team_id': team_id,
                    'team_name': team_name
                }

                cls.get_or_create_alert(
                    team_lead=team_lead,
                    text=alert_text,
                    json=alert_json,
                    category='number_distribution'
                )

                # Only one alert should be emitted so no need to finish the
                # loop.
                break

    @classmethod
    def data_collection_time_alert(cls, household_survey):
        """If timestamp for any data point in data collection is < 06:00 AM
        or > 8:00 PM then report alert on dashboard "Data collection time
        issue in team NAME (survey: UUID)".
        If any of the time stamps is not a valid date, the alert is triggered
        too.
        N.B. This function performs these checks for both the start time and
        the end time.  The alert is triggered if any of them satisfies the
        condition mentioned above.
        """
        minimum_time = datetime.time(6) # 6:00 AM
        maximum_time = datetime.time(20) # 8:00 PM
        start = household_survey.get_start_time()
        end = household_survey.get_end_time()
        triggered = False

        for t in (start, end):
            if t is None:
                triggered = True
                break
            elif t.time() < minimum_time or t.time() > maximum_time:
                triggered = True
                break

        if triggered:
            team_name = household_survey.get_team_name()
            team_id = household_survey.get_team_id()
            team_lead = household_survey.team_lead
            alert_text = u'Data collection time issue in team {} (survey: {})'.\
                format(team_id, household_survey.id)

{'fraction_non_alphanumeric': 0.0765605724154947, 'fraction_numerical': 0.013138415988156921, 'mean_word_length': 1.8581855364761468, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6689957', 'n_tokens_mistral': 23416, 'n_tokens_neox': 21691, 'n_words': 6260}