<commit_before># ActivitySim
# See full license in LICENSE.txt.

import logging
import itertools

import numpy as np
import pandas as pd
from zbox import toolz as tz, gen

from activitysim.core.simulate import eval_variables
from activitysim.core.simulate import compute_utilities

from activitysim.core import chunk
from activitysim.core import logit
from activitysim.core import tracing

logger = logging.getLogger(__name__)

# FIXME - this allows us to turn some dev debug table dump code on and off - eventually remove?
# DUMP = False

_persons_index_ = 'PERID'
_hh_index_ = 'HHID'
_hh_size_ = 'hhsize'

_hh_id_ = 'household_id'
_ptype_ = 'ptype'
_age_ = 'age'

# For clarity, the named constant MAX_HHSIZE refers to the cdap 5 person threshold figure.
MAX_HHSIZE = 5

MAX_INTERACTION_CARDINALITY = 3

WORKER_PTYPES = [1, 2]
CHILD_PTYPES = [6, 7, 8]


def set_hh_index(df):

    # index on household_id, not person_id
    df.set_index(_hh_id_, inplace=True)
    df.index.name = _hh_index_


def add_pn(col, pnum):
    """
    return the canonical column name for the indiv_util column or columns
    in merged hh_chooser df for individual with cdap_rank pnum

    e.g. M_p1, ptype_p2 but leave _hh_id_ column unchanged
    """
    if type(col) is str:
        return col if col == _hh_id_ else '%s_p%s' % (col, pnum)
    elif isinstance(col, (list, tuple)):
        return [c if c == _hh_id_ else '%s_p%s' % (c, pnum) for c in col]
    else:
        raise RuntimeError("add_pn col not list or str")


def assign_cdap_rank(persons, trace_hh_id=None, trace_label=None):
    """
    Assign an integer index, cdap_rank, to each household member. (Starting with 1, not 0)

    Modifies persons df in place

    The cdap_rank order is important, because cdap only assigns activities to the first
    MAX_HHSIZE persons in each household.

    This will preferentially be two working adults and the three youngest children.

    Rank is assigned starting at 1. This necessitates some care indexing, but is preferred as
    it follows the convention of 1-based pnums in expression files.

    According to the documentation of reOrderPersonsForCdap in mtctm2.abm.ctramp
    HouseholdCoordinatedDailyActivityPatternModel:

    "Method reorders the persons in the household for use with the CDAP model,
    which only explicitly models the interaction of five persons in a HH. Priority
    in the reordering is first given to full time workers (up to two), then to
    part time workers (up to two workers, of any type), then to children (youngest
    to oldest, up to three). If the method is called for a household with less
    than 5 people, the cdapPersonArray is the same as the person array."

    We diverge from the above description in that a cdap_rank is assigned to all persons,
    including 'extra' householld members, whose activity is assigned subsequently.
    The pair _hh_id_, cdap_rank will uniquely identify each household member.

    Parameters
    ----------
    persons : pandas.DataFrame
        Table of persons data. Must contain columns _hh_size_, _hh_id_, _ptype_, _age_

    Returns
    -------
    cdap_rank : pandas.Series
        integer cdap_rank of every person, indexed on _persons_index_
    """

    # transient categories used to categorize persons in cdap_rank before assigning final rank
    RANK_WORKER = 1
    RANK_CHILD = 2
    RANK_BACKFILL = 3
    RANK_UNASSIGNED = 9
    persons['cdap_rank'] = RANK_UNASSIGNED

    # choose up to 2 workers, preferring full over part, older over younger
    workers = \
        persons.loc[persons[_ptype_].isin(WORKER_PTYPES), [_hh_id_, _ptype_]]\
        .sort_values(by=[_hh_id_, _ptype_], ascending=[True, True])\
        .groupby(_hh_id_).head(2)
    # tag the selected workers
    persons.loc[workers.index, 'cdap_rank'] = RANK_WORKER
    del workers

    # choose up to 3, preferring youngest
    children = \
        persons.loc[persons[_ptype_].isin(CHILD_PTYPES), [_hh_id_, _ptype_, _age_]]\
        .sort_values(by=[_hh_id_, _ptype_], ascending=[True, True])\
        .groupby(_hh_id_).head(3)
    # tag the selected children
    persons.loc[children.index, 'cdap_rank'] = RANK_CHILD
    del children

    # choose up to MAX_HHSIZE, preferring anyone already chosen
    others = \
        persons[[_hh_id_, 'cdap_rank']]\
        .sort_values(by=[_hh_id_, 'cdap_rank'], ascending=[True, True])\
        .groupby(_hh_id_).head(MAX_HHSIZE)
    # tag the backfilled persons
    persons.loc[others[others.cdap_rank == RANK_UNASSIGNED].index, 'cdap_rank'] \
        = RANK_BACKFILL
    del others

    # assign person number in cdapPersonArray preference order
    # i.e. convert cdap_rank from category to index in order of category rank within household
    # groupby rank() is slow, so we compute rank artisanally
    # save time by sorting only the columns we need (persons is big, and sort moves data)
    p = persons[[_hh_id_, 'cdap_rank', _age_]]\
        .sort_values(by=[_hh_id_, 'cdap_rank', _age_], ascending=[True, True, True])
    rank = p.groupby(_hh_id_).size().map(range)
    rank = [item+1 for sublist in rank for item in sublist]
    p['cdap_rank'] = rank
    persons['cdap_rank'] = p['cdap_rank']  # assignment aligns on index values

    # if DUMP:
    #     tracing.trace_df(persons, '%s.DUMP.cdap_person_array' % trace_label,
    #                      transpose=False, slicer='NONE')

    if trace_hh_id:
        tracing.trace_df(persons, '%s.cdap_rank' % trace_label)

    return persons['cdap_rank']


def individual_utilities(
        persons,
        cdap_indiv_spec,
        locals_d,
        trace_hh_id=None, trace_label=None):
    """
    Calculate CDAP utilities for all individuals.

    Parameters
    ----------
    persons : pandas.DataFrame
        DataFrame of individual persons data.
    cdap_indiv_spec : pandas.DataFrame
        CDAP spec applied to individuals.

    Returns
    -------
    utilities : pandas.DataFrame
        Will have index of `persons` and columns for each of the alternatives.
        plus some 'useful columns' [_hh_id_, _ptype_, 'cdap_rank', _hh_size_]

    """

    # calculate single person utilities
    individual_vars = eval_variables(cdap_indiv_spec.index, persons, locals_d)
    indiv_utils = compute_utilities(individual_vars, cdap_indiv_spec)

    # add columns from persons to facilitate building household interactions
    useful_columns = [_hh_id_, _ptype_, 'cdap_rank', _hh_size_]
    indiv_utils[useful_columns] = persons[useful_columns]

    # if DUMP:
    #     tracing.trace_df(indiv_utils, '%s.DUMP.indiv_utils' % trace_label,
    #                      transpose=False, slicer='NONE')

    if trace_hh_id:
        tracing.trace_df(individual_vars, '%s.individual_vars' % trace_label,
                         column_labels=['expression', 'person'])
        tracing.trace_df(indiv_utils, '%s.indiv_utils' % trace_label,
                         column_labels=['activity', 'person'])

    return indiv_utils


def preprocess_interaction_coefficients(interaction_coefficients):
    """
    The input cdap_interaction_coefficients.csv file has three columns:

    activity
        A single character activity type name (M, N, or H)

    interaction_ptypes
        List of ptypes in the interaction (in order of increasing ptype)
        Stars (\***) instead of ptypes means the interaction applies to all ptypes in that size hh

    coefficient
        The coefficient to apply for all hh interactions for this activity and set of ptypes

    To facilitate building the spec for a given hh ssize, we add two additional columns:

    cardinality
        the number of persons in the interaction (e.g. 3 for a 3-way interaction)

    slug
        a human friendly efficient name so we can dump a readable spec trace file for debugging
        this slug is then replaced with the numerical coefficient value prior to evaluation
    """

    # make a copy
    coefficients = interaction_coefficients.copy()

    if not coefficients['activity'].isin(['M', 'N', 'H']).all():
        msg = "Error in cdap_interaction_coefficients at row %s. Expect only M, N, or H!" \
              % coefficients[~coefficients['activity'].isin(['M', 'N', 'H'])].index.values
        raise RuntimeError(msg)

    coefficients['cardinality'] = coefficients['interaction_ptypes'].astype(str).str.len()

    wildcards = coefficients.interaction_ptypes == coefficients.cardinality.map(lambda x: x*'*')
    coefficients.loc[wildcards, 'interaction_ptypes'] = ''
<commit_msg>pcodestype didn't like escape sequence in a comment<commit_after># ActivitySim
# See full license in LICENSE.txt.

import logging
import itertools

import numpy as np
import pandas as pd
from zbox import toolz as tz, gen

from activitysim.core.simulate import eval_variables
from activitysim.core.simulate import compute_utilities

from activitysim.core import chunk
from activitysim.core import logit
from activitysim.core import tracing

logger = logging.getLogger(__name__)

# FIXME - this allows us to turn some dev debug table dump code on and off - eventually remove?
# DUMP = False

_persons_index_ = 'PERID'
_hh_index_ = 'HHID'
_hh_size_ = 'hhsize'

_hh_id_ = 'household_id'
_ptype_ = 'ptype'
_age_ = 'age'

# For clarity, the named constant MAX_HHSIZE refers to the cdap 5 person threshold figure.
MAX_HHSIZE = 5

MAX_INTERACTION_CARDINALITY = 3

WORKER_PTYPES = [1, 2]
CHILD_PTYPES = [6, 7, 8]


def set_hh_index(df):

    # index on household_id, not person_id
    df.set_index(_hh_id_, inplace=True)
    df.index.name = _hh_index_


def add_pn(col, pnum):
    """
    return the canonical column name for the indiv_util column or columns
    in merged hh_chooser df for individual with cdap_rank pnum

    e.g. M_p1, ptype_p2 but leave _hh_id_ column unchanged
    """
    if type(col) is str:
        return col if col == _hh_id_ else '%s_p%s' % (col, pnum)
    elif isinstance(col, (list, tuple)):
        return [c if c == _hh_id_ else '%s_p%s' % (c, pnum) for c in col]
    else:
        raise RuntimeError("add_pn col not list or str")


def assign_cdap_rank(persons, trace_hh_id=None, trace_label=None):
    """
    Assign an integer index, cdap_rank, to each household member. (Starting with 1, not 0)

    Modifies persons df in place

    The cdap_rank order is important, because cdap only assigns activities to the first
    MAX_HHSIZE persons in each household.

    This will preferentially be two working adults and the three youngest children.

    Rank is assigned starting at 1. This necessitates some care indexing, but is preferred as
    it follows the convention of 1-based pnums in expression files.

    According to the documentation of reOrderPersonsForCdap in mtctm2.abm.ctramp
    HouseholdCoordinatedDailyActivityPatternModel:

    "Method reorders the persons in the household for use with the CDAP model,
    which only explicitly models the interaction of five persons in a HH. Priority
    in the reordering is first given to full time workers (up to two), then to
    part time workers (up to two workers, of any type), then to children (youngest
    to oldest, up to three). If the method is called for a household with less
    than 5 people, the cdapPersonArray is the same as the person array."

    We diverge from the above description in that a cdap_rank is assigned to all persons,
    including 'extra' householld members, whose activity is assigned subsequently.
    The pair _hh_id_, cdap_rank will uniquely identify each household member.

    Parameters
    ----------
    persons : pandas.DataFrame
        Table of persons data. Must contain columns _hh_size_, _hh_id_, _ptype_, _age_

    Returns
    -------
    cdap_rank : pandas.Series
        integer cdap_rank of every person, indexed on _persons_index_
    """

    # transient categories used to categorize persons in cdap_rank before assigning final rank
    RANK_WORKER = 1
    RANK_CHILD = 2
    RANK_BACKFILL = 3
    RANK_UNASSIGNED = 9
    persons['cdap_rank'] = RANK_UNASSIGNED

    # choose up to 2 workers, preferring full over part, older over younger
    workers = \
        persons.loc[persons[_ptype_].isin(WORKER_PTYPES), [_hh_id_, _ptype_]]\
        .sort_values(by=[_hh_id_, _ptype_], ascending=[True, True])\
        .groupby(_hh_id_).head(2)
    # tag the selected workers
    persons.loc[workers.index, 'cdap_rank'] = RANK_WORKER
    del workers

    # choose up to 3, preferring youngest
    children = \
        persons.loc[persons[_ptype_].isin(CHILD_PTYPES), [_hh_id_, _ptype_, _age_]]\
        .sort_values(by=[_hh_id_, _ptype_], ascending=[True, True])\
        .groupby(_hh_id_).head(3)
    # tag the selected children
    persons.loc[children.index, 'cdap_rank'] = RANK_CHILD
    del children

    # choose up to MAX_HHSIZE, preferring anyone already chosen
    others = \
        persons[[_hh_id_, 'cdap_rank']]\
        .sort_values(by=[_hh_id_, 'cdap_rank'], ascending=[True, True])\
        .groupby(_hh_id_).head(MAX_HHSIZE)
    # tag the backfilled persons
    persons.loc[others[others.cdap_rank == RANK_UNASSIGNED].index, 'cdap_rank'] \
        = RANK_BACKFILL
    del others

    # assign person number in cdapPersonArray preference order
    # i.e. convert cdap_rank from category to index in order of category rank within household
    # groupby rank() is slow, so we compute rank artisanally
    # save time by sorting only the columns we need (persons is big, and sort moves data)
    p = persons[[_hh_id_, 'cdap_rank', _age_]]\
        .sort_values(by=[_hh_id_, 'cdap_rank', _age_], ascending=[True, True, True])
    rank = p.groupby(_hh_id_).size().map(range)
    rank = [item+1 for sublist in rank for item in sublist]
    p['cdap_rank'] = rank
    persons['cdap_rank'] = p['cdap_rank']  # assignment aligns on index values

    # if DUMP:
    #     tracing.trace_df(persons, '%s.DUMP.cdap_person_array' % trace_label,
    #                      transpose=False, slicer='NONE')

    if trace_hh_id:
        tracing.trace_df(persons, '%s.cdap_rank' % trace_label)

    return persons['cdap_rank']


def individual_utilities(
        persons,
        cdap_indiv_spec,
        locals_d,
        trace_hh_id=None, trace_label=None):
    """
    Calculate CDAP utilities for all individuals.

    Parameters
    ----------
    persons : pandas.DataFrame
        DataFrame of individual persons data.
    cdap_indiv_spec : pandas.DataFrame
        CDAP spec applied to individuals.

    Returns
    -------
    utilities : pandas.DataFrame
        Will have index of `persons` and columns for each of the alternatives.
        plus some 'useful columns' [_hh_id_, _ptype_, 'cdap_rank', _hh_size_]

    """

    # calculate single person utilities
    individual_vars = eval_variables(cdap_indiv_spec.index, persons, locals_d)
    indiv_utils = compute_utilities(individual_vars, cdap_indiv_spec)

    # add columns from persons to facilitate building household interactions
    useful_columns = [_hh_id_, _ptype_, 'cdap_rank', _hh_size_]
    indiv_utils[useful_columns] = persons[useful_columns]

    # if DUMP:
    #     tracing.trace_df(indiv_utils, '%s.DUMP.indiv_utils' % trace_label,
    #                      transpose=False, slicer='NONE')

    if trace_hh_id:
        tracing.trace_df(individual_vars, '%s.individual_vars' % trace_label,
                         column_labels=['expression', 'person'])
        tracing.trace_df(indiv_utils, '%s.indiv_utils' % trace_label,
                         column_labels=['activity', 'person'])

    return indiv_utils


def preprocess_interaction_coefficients(interaction_coefficients):
    """
    The input cdap_interaction_coefficients.csv file has three columns:

    activity
        A single character activity type name (M, N, or H)

    interaction_ptypes
        List of ptypes in the interaction (in order of increasing ptype)
        Stars (***) instead of ptypes means the interaction applies to all ptypes in that size hh.

    coefficient
        The coefficient to apply for all hh interactions for this activity and set of ptypes

    To facilitate building the spec for a given hh ssize, we add two additional columns:

    cardinality
        the number of persons in the interaction (e.g. 3 for a 3-way interaction)

    slug
        a human friendly efficient name so we can dump a readable spec trace file for debugging
        this slug is then replaced with the numerical coefficient value prior to evaluation
    """

    # make a copy
    coefficients = interaction_coefficients.copy()

    if not coefficients['activity'].isin(['M', 'N', 'H']).all():
        msg = "Error in cdap_interaction_coefficients at row %s. Expect only M, N, or H!" \
              % coefficients[~coefficients['activity'].isin(['M', 'N', 'H'])].index.values
        raise RuntimeError(msg)

    coefficients['cardinality'] = coefficients['interaction_ptypes'].astype(str).str.len()

    wildcards = coefficients.interaction_ptypes == coefficients.cardinality.map(lambda x: x*'*')
    coefficients.loc[wildcards, 'interaction_ptypes'] = ''

{'fraction_non_alphanumeric': 0.08367538407411751, 'fraction_numerical': 0.0031664125718306557, 'mean_word_length': 3.2467629482071714, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6419928', 'n_tokens_mistral': 5327, 'n_tokens_neox': 5145, 'n_words': 1735}