Strategies

AssignmentStrategy

Base class for assignment strategies.

Strategies perform straightforward assignments based on context.

Source code in may/attribute_assignment/strategies.py

class AssignmentStrategy:
    """
    Base class for assignment strategies.

    Strategies perform straightforward assignments based on context.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        """
        Initialize strategy.

        Args:
            config: Strategy configuration from YAML
            data_manager: DataSourceManager instance for data lookups
        """
        self.config = config
        self.data_manager = data_manager
        self.strategy_type = config.get('strategy')

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """
        Assign attribute value to a person.

        Args:
            person: Person object to assign to
            household: Household (venue) object
            context: Assignment context with state information

        Returns:
            Assigned attribute value
        """
        raise NotImplementedError("Subclasses must implement assign()")

    def _fail(self, person, reason: str):
        """Abort assignment loudly. No fallbacks."""
        raise RuntimeError(
            f"Strategy '{self.strategy_type}' could not assign person {person.id}: "
            f"{reason}. No fallbacks. Fix the data/config, or express the "
            "alternative as explicit primary logic."
        )

    def _weighted_draw(self, probs: Dict[str, float], person, *, size=None):
        """
        Draw value(s) from a {value: weight} distribution.

        The single weighted-draw code path shared by every draw-family strategy:
        clamp negative weights to zero, normalize, then `np.random.choice`. With
        `size=None` returns one value; with `size=n` returns n values (batch).
        An empty or zero-total distribution fails loudly rather than returning None.
        """
        if not probs:
            self._fail(person, "data source returned no distribution")
        values = list(probs.keys())
        weights = np.asarray(list(probs.values()), dtype=float)
        if np.any(weights < 0):
            logger.warning(f"Negative weight(s) clamped to 0 for person {person.id}")
            weights = np.clip(weights, 0.0, None)
        total = weights.sum()
        if total <= 0:
            self._fail(person, "distribution has zero total weight")
        return np.random.choice(values, size=size, p=weights / total)

    def _marginal_assign(self, person, household, context: Dict[str, Any]) -> Any:
        """
        Assign from the configured marginal distribution.

        This is explicit primary logic for the defined case where a strategy has
        nothing to condition on (e.g. inheritance with no parent values). The
        marginal source is named by `marginal_source`; absence of that key means
        the case is not expected and is a hard error.
        """
        marginal_source = self.config.get('marginal_source')
        if not marginal_source:
            self._fail(
                person,
                "no value to condition on and no 'marginal_source' configured",
            )
        strat = ProbabilisticStrategy(
            {'strategy': 'probabilistic', 'data_source': marginal_source},
            self.data_manager,
        )
        return strat.assign(person, household, context)

    def _get_person_by_role(self, context: Dict[str, Any], role_name: str):
        """
        Get person by role name from context.

        Args:
            context: Assignment context
            role_name: Role name (e.g., "primary_adult")

        Returns:
            Person object or None
        """
        person_key = f"{role_name}_person"
        return context.get(person_key)

    def _get_attribute_value(self, person, attribute_name: str) -> Any:
        """
        Get attribute value from person.

        Delegates to the shared get_person_attribute utility which handles
        dot-notation, properties dict, and residence prefix.

        Args:
            person: Person object
            attribute_name: Name of attribute

        Returns:
            Attribute value or None
        """
        from may.utils.attribute_access import get_person_attribute
        return get_person_attribute(person, attribute_name)

__init__(config, data_manager)

Initialize strategy.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Strategy configuration from YAML	required
data_manager		DataSourceManager instance for data lookups	required

Source code in may/attribute_assignment/strategies.py

def __init__(self, config: Dict[str, Any], data_manager):
    """
    Initialize strategy.

    Args:
        config: Strategy configuration from YAML
        data_manager: DataSourceManager instance for data lookups
    """
    self.config = config
    self.data_manager = data_manager
    self.strategy_type = config.get('strategy')

assign(person, household, context)

Assign attribute value to a person.

Parameters:

Name	Type	Description	Default
person		Person object to assign to	required
household		Household (venue) object	required
context	Dict[str, Any]	Assignment context with state information	required

Returns:

Type	Description
Any	Assigned attribute value

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """
    Assign attribute value to a person.

    Args:
        person: Person object to assign to
        household: Household (venue) object
        context: Assignment context with state information

    Returns:
        Assigned attribute value
    """
    raise NotImplementedError("Subclasses must implement assign()")

CommutingLikelihoodStrategy

Bases: AssignmentStrategy

Assigns workplace location based on origin-destination commuting flows.

Samples from an origin-destination matrix weighted by likelihood. Can assign multiple attributes (e.g., workplace_location and work_mode).

Supports batch assignment to reduce repeated lookups.

Source code in may/attribute_assignment/strategies.py

class CommutingLikelihoodStrategy(AssignmentStrategy):
    """
    Assigns workplace location based on origin-destination commuting flows.

    Samples from an origin-destination matrix weighted by likelihood.
    Can assign multiple attributes (e.g., workplace_location and work_mode).

    Supports batch assignment to reduce repeated lookups.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        """Initialize commuting likelihood strategy."""
        super().__init__(config, data_manager)
        self.data_source_name = config.get('data_source')
        self.outputs = config.get('outputs', {})
        # Optional: restrict the destination draw to O-D rows whose metadata
        # matches an attribute already assigned to the person. Lets a coarser
        # matrix answer only "where", once a finer source has answered "whether".
        self.condition = config.get('condition')

    def _condition_allowed(self, person) -> Optional[set]:
        """Metadata values this person's destinations may carry, or None if unconditioned."""
        if not self.condition:
            return None
        attribute = self.condition['attribute']
        value = self._get_attribute_value(person, attribute)
        mapping = self.condition['map']
        if value not in mapping:
            raise ValueError(
                f"commuting_likelihood condition on '{attribute}': value {value!r} "
                f"is not in the map ({sorted(mapping)}). No fallbacks."
            )
        return set(mapping[value])

    def _filter_by_condition(self, destinations, allowed, origin_code):
        """Keep O-D rows whose condition metadata is allowed, renormalised to 1.0."""
        if allowed is None:
            return destinations
        key = self.condition['metadata_key']
        kept = [(d, m, l) for (d, m, l) in destinations if m.get(key) in allowed]
        total = sum(l for _, _, l in kept)
        if not kept or total <= 0:
            raise ValueError(
                f"commuting_likelihood: origin '{origin_code}' has no destinations "
                f"with {key} in {sorted(allowed)}. The O-D matrix and the "
                "conditioning attribute disagree. No fallbacks."
            )
        return [(d, m, l / total) for (d, m, l) in kept]

    def _resolve_origin_code(self, person) -> Optional[str]:
        """
        Resolve person's origin geographical unit to the correct level for O-D matrix lookup.

        This handles complex data source configurations like ancestor lookups.

        Args:
            person: Person object

        Returns:
            Origin code string, or None if resolution fails
        """
        origin_geo_unit = getattr(person, 'geographical_unit', None)
        if not origin_geo_unit:
            return None

        source = self.data_manager.get_source(self.data_source_name)
        if not source:
            return None

        if hasattr(source, '_file_configs') and source._file_configs:
            file_config = source._file_configs[0]
            key_columns = file_config.get('key_columns', {})

            if key_columns:
                first_key_config = list(key_columns.values())[0]

                if isinstance(first_key_config, dict):
                    lookup_type = first_key_config.get('type')
                    if lookup_type == 'ancestor_lookup':
                        level = first_key_config.get('level')
                        property_name = first_key_config.get('property', 'name')

                        ancestor = origin_geo_unit.get_ancestor_by_level(level)
                        if ancestor:
                            return getattr(ancestor, property_name)
                        else:
                            return None
                    else:
                        return origin_geo_unit.name
                else:
                    return origin_geo_unit.name
            else:
                return origin_geo_unit.name
        else:
            return origin_geo_unit.name

    def assign_batch(self, people_list: List, households_list: List, contexts_list: List[Dict[str, Any]]) -> List[Any]:
        """
        Batch assignment to minimize repeated O-D matrix lookups.

        Groups people by origin code and processes each group together.

        Args:
            people_list: List of Person objects
            households_list: List of Household objects (parallel to people_list)
            contexts_list: List of context dicts (parallel to people_list)

        Returns:
            List of assigned values (parallel to people_list)
            - If single output: list of values
            - If multiple outputs: list of dicts
        """
        from collections import defaultdict

        source = self.data_manager.get_source(self.data_source_name)
        if not source:
            logger.warning(f"Data source '{self.data_source_name}' not found")
            return [self._fail(person, "no commuting-flow row for this origin")
                    for person, household, context in zip(people_list, households_list, contexts_list)]

        # Group by origin and, when conditioning, by the condition value too:
        # every member of a group then shares one filtered distribution.
        origin_groups = defaultdict(list)

        for i, person in enumerate(people_list):
            origin_code = self._resolve_origin_code(person)
            if origin_code:
                allowed = self._condition_allowed(person)
                origin_groups[(origin_code,
                               frozenset(allowed) if allowed else None)].append(i)

        results = [None] * len(people_list)

        for (origin_code, allowed), indices in origin_groups.items():
            destinations = source.lookup(origin_code)
            if destinations:
                destinations = self._filter_by_condition(destinations, allowed, origin_code)
            if not destinations:
                logger.warning(f"No destinations found for origin {origin_code}")
                for idx in indices:
                    person = people_list[idx]
                    household = households_list[idx]
                    context = contexts_list[idx]
                    results[idx] = self._fail(person, "no commuting-flow row for this origin")
                continue

            # destinations is List[(destination, metadata_dict, likelihood)]
            dest_codes = [dest for dest, meta, lik in destinations]
            likelihoods = [lik for dest, meta, lik in destinations]
            metadata_list = [meta for dest, meta, lik in destinations]

            n_samples = len(indices)
            sampled_indices = np.random.choice(len(dest_codes), size=n_samples, p=likelihoods)

            for idx, sampled_idx in zip(indices, sampled_indices):
                sampled_dest = dest_codes[sampled_idx]
                sampled_metadata = metadata_list[sampled_idx]
                results[idx] = self._build_output(sampled_dest, sampled_metadata)

            # Mark redistributed assignments. A worker from this origin was
            # bounced back in-boundary with probability = the origin's
            # out-of-boundary mass; a per-person Bernoulli reproduces that fraction.
            flag = getattr(source, '_redistributed_flag', None)
            fraction = getattr(source, '_redistributed_fraction', {}).get(origin_code, 0.0)
            if flag and fraction > 0.0:
                marks = np.random.random(len(indices)) < fraction
                for idx, marked in zip(indices, marks):
                    if marked:
                        results[idx] = self._with_redistributed_flag(results[idx], flag)

        # Headcount of out-of-boundary assignments. Under the 'outside'
        # policy the sentinel is a normal destination value, so report how many
        # people drew it, and the per-step exclusion downstream reports the rest.
        outside_value = getattr(source, '_outside_value', None)
        if outside_value:
            # The sentinel lands on whichever output is wired to 'destination'.
            dest_attr = next(
                (attr for attr, src in self.outputs.items() if src == 'destination'),
                None,
            )
            n_outside = sum(
                1 for r in results
                if (r.get(dest_attr) if isinstance(r, dict) else r) == outside_value
            )
            if n_outside:
                logger.info(
                    f"  [out_of_boundary] {n_outside}/{len(results)} people assigned the "
                    f"'{outside_value}' sentinel (no in-world destination)."
                )

        return results

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """
        Assign workplace location and work mode based on commuting flows.

        Args:
            person: Person object
            household: Household object (optional, may be None for person-level assignment)
            context: Assignment context

        Returns:
            If single output: returns the assigned value
            If multiple outputs: returns dict with all assigned values
        """
        origin_code = self._resolve_origin_code(person)
        if not origin_code:
            logger.warning(f"Could not resolve origin code for person {person.id}")
            return self._fail(person, "no commuting-flow row for this origin")

        source = self.data_manager.get_source(self.data_source_name)
        if not source:
            logger.warning(f"Data source '{self.data_source_name}' not found")
            return self._fail(person, "no commuting-flow row for this origin")

        destinations = source.lookup(origin_code)
        if destinations:
            destinations = self._filter_by_condition(
                destinations, self._condition_allowed(person), origin_code
            )
        if not destinations:
            logger.warning(f"No destinations found for origin {origin_code}")
            return self._fail(person, "no commuting-flow row for this origin")

        # destinations is List[(destination, metadata_dict, likelihood)]
        dest_codes = [dest for dest, meta, lik in destinations]
        likelihoods = [lik for dest, meta, lik in destinations]
        metadata_list = [meta for dest, meta, lik in destinations]

        idx = np.random.choice(len(dest_codes), p=likelihoods)
        sampled_dest = dest_codes[idx]
        sampled_metadata = metadata_list[idx]

        result = self._build_output(sampled_dest, sampled_metadata)
        if not isinstance(result, dict):
            return result
        logger.debug(f"Commuting: {person.id} -> {result}")
        return result

    def _with_redistributed_flag(self, result, flag: str):
        """
        Attach the config-named redistributed flag to a result.

        Dict results gain the flag key. A scalar (single-output) result is promoted
        to a dict so the flag can ride alongside it. If there is no 'destination'
        output to anchor that promotion, we fail loud.
        """
        if isinstance(result, dict):
            result[flag] = True
            return result
        dest_attr = next(
            (attr for attr, src in self.outputs.items() if src == 'destination'),
            None,
        )
        if dest_attr is None:
            raise ValueError(
                f"Strategy '{self.strategy_type}': cannot attach redistributed flag "
                f"'{flag}'. No output is wired to 'destination'."
            )
        return {dest_attr: result, flag: True}

    def _build_output(self, sampled_dest, sampled_metadata):
        """
        Build return value from sampled destination and metadata.

        Returns:
            Single value (if one output configured) or dict (if multiple).
        """
        if len(self.outputs) == 1:
            output_attr, output_source = list(self.outputs.items())[0]
            if output_source == 'destination':
                return sampled_dest
            if output_source in sampled_metadata:
                return sampled_metadata[output_source]
            raise ValueError(
                f"Output source '{output_source}' not found in metadata keys "
                f"{list(sampled_metadata.keys())}. Check outputs config."
            )

        result = {}
        for output_attr, output_source in self.outputs.items():
            if output_source == 'destination':
                result[output_attr] = sampled_dest
            elif output_source in sampled_metadata:
                result[output_attr] = sampled_metadata[output_source]
            else:
                raise ValueError(
                    f"Output source '{output_source}' for attribute '{output_attr}' "
                    f"not found in metadata keys {list(sampled_metadata.keys())}. "
                    f"Check outputs config."
                )
        return result

__init__(config, data_manager)

Initialize commuting likelihood strategy.

Source code in may/attribute_assignment/strategies.py

def __init__(self, config: Dict[str, Any], data_manager):
    """Initialize commuting likelihood strategy."""
    super().__init__(config, data_manager)
    self.data_source_name = config.get('data_source')
    self.outputs = config.get('outputs', {})
    # Optional: restrict the destination draw to O-D rows whose metadata
    # matches an attribute already assigned to the person. Lets a coarser
    # matrix answer only "where", once a finer source has answered "whether".
    self.condition = config.get('condition')

assign(person, household, context)

Assign workplace location and work mode based on commuting flows.

Parameters:

Name	Type	Description	Default
person		Person object	required
household		Household object (optional, may be None for person-level assignment)	required
context	Dict[str, Any]	Assignment context	required

Returns:

Type	Description
Any	If single output: returns the assigned value
Any	If multiple outputs: returns dict with all assigned values

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """
    Assign workplace location and work mode based on commuting flows.

    Args:
        person: Person object
        household: Household object (optional, may be None for person-level assignment)
        context: Assignment context

    Returns:
        If single output: returns the assigned value
        If multiple outputs: returns dict with all assigned values
    """
    origin_code = self._resolve_origin_code(person)
    if not origin_code:
        logger.warning(f"Could not resolve origin code for person {person.id}")
        return self._fail(person, "no commuting-flow row for this origin")

    source = self.data_manager.get_source(self.data_source_name)
    if not source:
        logger.warning(f"Data source '{self.data_source_name}' not found")
        return self._fail(person, "no commuting-flow row for this origin")

    destinations = source.lookup(origin_code)
    if destinations:
        destinations = self._filter_by_condition(
            destinations, self._condition_allowed(person), origin_code
        )
    if not destinations:
        logger.warning(f"No destinations found for origin {origin_code}")
        return self._fail(person, "no commuting-flow row for this origin")

    # destinations is List[(destination, metadata_dict, likelihood)]
    dest_codes = [dest for dest, meta, lik in destinations]
    likelihoods = [lik for dest, meta, lik in destinations]
    metadata_list = [meta for dest, meta, lik in destinations]

    idx = np.random.choice(len(dest_codes), p=likelihoods)
    sampled_dest = dest_codes[idx]
    sampled_metadata = metadata_list[idx]

    result = self._build_output(sampled_dest, sampled_metadata)
    if not isinstance(result, dict):
        return result
    logger.debug(f"Commuting: {person.id} -> {result}")
    return result

assign_batch(people_list, households_list, contexts_list)

Batch assignment to minimize repeated O-D matrix lookups.

Groups people by origin code and processes each group together.

Parameters:

Name	Type	Description	Default
people_list	List	List of Person objects	required
households_list	List	List of Household objects (parallel to people_list)	required
contexts_list	List[Dict[str, Any]]	List of context dicts (parallel to people_list)	required

Returns:

Type	Description
List[Any]	List of assigned values (parallel to people_list)
List[Any]	If single output: list of values
List[Any]	If multiple outputs: list of dicts

Source code in may/attribute_assignment/strategies.py

def assign_batch(self, people_list: List, households_list: List, contexts_list: List[Dict[str, Any]]) -> List[Any]:
    """
    Batch assignment to minimize repeated O-D matrix lookups.

    Groups people by origin code and processes each group together.

    Args:
        people_list: List of Person objects
        households_list: List of Household objects (parallel to people_list)
        contexts_list: List of context dicts (parallel to people_list)

    Returns:
        List of assigned values (parallel to people_list)
        - If single output: list of values
        - If multiple outputs: list of dicts
    """
    from collections import defaultdict

    source = self.data_manager.get_source(self.data_source_name)
    if not source:
        logger.warning(f"Data source '{self.data_source_name}' not found")
        return [self._fail(person, "no commuting-flow row for this origin")
                for person, household, context in zip(people_list, households_list, contexts_list)]

    # Group by origin and, when conditioning, by the condition value too:
    # every member of a group then shares one filtered distribution.
    origin_groups = defaultdict(list)

    for i, person in enumerate(people_list):
        origin_code = self._resolve_origin_code(person)
        if origin_code:
            allowed = self._condition_allowed(person)
            origin_groups[(origin_code,
                           frozenset(allowed) if allowed else None)].append(i)

    results = [None] * len(people_list)

    for (origin_code, allowed), indices in origin_groups.items():
        destinations = source.lookup(origin_code)
        if destinations:
            destinations = self._filter_by_condition(destinations, allowed, origin_code)
        if not destinations:
            logger.warning(f"No destinations found for origin {origin_code}")
            for idx in indices:
                person = people_list[idx]
                household = households_list[idx]
                context = contexts_list[idx]
                results[idx] = self._fail(person, "no commuting-flow row for this origin")
            continue

        # destinations is List[(destination, metadata_dict, likelihood)]
        dest_codes = [dest for dest, meta, lik in destinations]
        likelihoods = [lik for dest, meta, lik in destinations]
        metadata_list = [meta for dest, meta, lik in destinations]

        n_samples = len(indices)
        sampled_indices = np.random.choice(len(dest_codes), size=n_samples, p=likelihoods)

        for idx, sampled_idx in zip(indices, sampled_indices):
            sampled_dest = dest_codes[sampled_idx]
            sampled_metadata = metadata_list[sampled_idx]
            results[idx] = self._build_output(sampled_dest, sampled_metadata)

        # Mark redistributed assignments. A worker from this origin was
        # bounced back in-boundary with probability = the origin's
        # out-of-boundary mass; a per-person Bernoulli reproduces that fraction.
        flag = getattr(source, '_redistributed_flag', None)
        fraction = getattr(source, '_redistributed_fraction', {}).get(origin_code, 0.0)
        if flag and fraction > 0.0:
            marks = np.random.random(len(indices)) < fraction
            for idx, marked in zip(indices, marks):
                if marked:
                    results[idx] = self._with_redistributed_flag(results[idx], flag)

    # Headcount of out-of-boundary assignments. Under the 'outside'
    # policy the sentinel is a normal destination value, so report how many
    # people drew it, and the per-step exclusion downstream reports the rest.
    outside_value = getattr(source, '_outside_value', None)
    if outside_value:
        # The sentinel lands on whichever output is wired to 'destination'.
        dest_attr = next(
            (attr for attr, src in self.outputs.items() if src == 'destination'),
            None,
        )
        n_outside = sum(
            1 for r in results
            if (r.get(dest_attr) if isinstance(r, dict) else r) == outside_value
        )
        if n_outside:
            logger.info(
                f"  [out_of_boundary] {n_outside}/{len(results)} people assigned the "
                f"'{outside_value}' sentinel (no in-world destination)."
            )

    return results

ConstantStrategy

Bases: AssignmentStrategy

Assigns a fixed, constant value.

This is useful for static attributes or default values that apply unconditionally to a given role or household structure.

Source code in may/attribute_assignment/strategies.py

class ConstantStrategy(AssignmentStrategy):
    """
    Assigns a fixed, constant value.

    This is useful for static attributes or default values that apply
    unconditionally to a given role or household structure.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        """Initialize constant strategy."""
        super().__init__(config, data_manager)
        self.value = config.get('value')

    def assign_batch(self, people_list: List, households_list: List, contexts_list: List[Dict[str, Any]]) -> List[Any]:
        """Batch assignment - all receive the same value."""
        if self.value is None:
            return [self.assign(p, h, c) for p, h, c in zip(people_list, households_list, contexts_list)]
        return [self.value] * len(people_list)

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """Assign the constant value."""
        if self.value is None:
            logger.warning(f"ConstantStrategy: No value configured for assignment to person {person.id}")
            return self._fail(person, "constant strategy has no 'value'")

        return self.value

__init__(config, data_manager)

Initialize constant strategy.

Source code in may/attribute_assignment/strategies.py

def __init__(self, config: Dict[str, Any], data_manager):
    """Initialize constant strategy."""
    super().__init__(config, data_manager)
    self.value = config.get('value')

assign(person, household, context)

Assign the constant value.

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """Assign the constant value."""
    if self.value is None:
        logger.warning(f"ConstantStrategy: No value configured for assignment to person {person.id}")
        return self._fail(person, "constant strategy has no 'value'")

    return self.value

assign_batch(people_list, households_list, contexts_list)

Batch assignment - all receive the same value.

Source code in may/attribute_assignment/strategies.py

def assign_batch(self, people_list: List, households_list: List, contexts_list: List[Dict[str, Any]]) -> List[Any]:
    """Batch assignment - all receive the same value."""
    if self.value is None:
        return [self.assign(p, h, c) for p, h, c in zip(people_list, households_list, contexts_list)]
    return [self.value] * len(people_list)

DrawStrategy

Bases: AssignmentStrategy

Weighted single draw from a {value: weight} distribution.

One strategy for every weighted-draw use. The split: the data source owns key-resolution and weight computation. It receives (person, household, context) and returns the distribution; this strategy owns the sampling mechanics (sanitize + draw, via _weighted_draw). The probabilistic, categorical_sampler and geographical_unit_sampler strategies are aliases of this one class.

Source code in may/attribute_assignment/strategies.py

class DrawStrategy(AssignmentStrategy):
    """
    Weighted single draw from a {value: weight} distribution.

    One strategy for every weighted-draw use. The split: the **data source** owns
    key-resolution and weight computation. It receives (person, household,
    context) and returns the distribution; this **strategy** owns the sampling
    mechanics (sanitize + draw, via `_weighted_draw`). The `probabilistic`,
    `categorical_sampler` and `geographical_unit_sampler` strategies are aliases
    of this one class.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        super().__init__(config, data_manager)
        self.data_source_name = config.get('data_source')

    def _source(self):
        source = self.data_manager.get_source(self.data_source_name)
        if not source:
            raise KeyError(
                f"Data source '{self.data_source_name}' is not registered. "
                "No fallbacks."
            )
        return source

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """Draw one value from the source's distribution for this person."""
        probs = self._source().lookup(person, household, context)
        sampled = self._weighted_draw(probs, person)
        logger.debug(f"{self.strategy_type}: {sampled} for {person.id}")
        return sampled

    def assign_batch(self, people_list: List, households_list: List,
                     contexts_list: List[Dict[str, Any]]) -> List[Any]:
        """
        Batch draw. Group people by the distribution their source returns, then
        sample each group in one np.random.choice. Grouping is by the resolved
        distribution, so it needs no knowledge of how the source keys its lookup.
        """
        from collections import defaultdict

        source = self._source()
        groups = defaultdict(list)
        for i, (person, household, context) in enumerate(
            zip(people_list, households_list, contexts_list)
        ):
            probs = source.lookup(person, household, context)
            groups[tuple(sorted(probs.items()))].append((i, person, probs))

        results = [None] * len(people_list)
        for members in groups.values():
            _, person0, probs = members[0]
            sampled = self._weighted_draw(probs, person0, size=len(members))
            for (idx, _person, _probs), value in zip(members, sampled):
                results[idx] = value
        return results

assign(person, household, context)

Draw one value from the source's distribution for this person.

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """Draw one value from the source's distribution for this person."""
    probs = self._source().lookup(person, household, context)
    sampled = self._weighted_draw(probs, person)
    logger.debug(f"{self.strategy_type}: {sampled} for {person.id}")
    return sampled

assign_batch(people_list, households_list, contexts_list)

Batch draw. Group people by the distribution their source returns, then sample each group in one np.random.choice. Grouping is by the resolved distribution, so it needs no knowledge of how the source keys its lookup.

Source code in may/attribute_assignment/strategies.py

def assign_batch(self, people_list: List, households_list: List,
                 contexts_list: List[Dict[str, Any]]) -> List[Any]:
    """
    Batch draw. Group people by the distribution their source returns, then
    sample each group in one np.random.choice. Grouping is by the resolved
    distribution, so it needs no knowledge of how the source keys its lookup.
    """
    from collections import defaultdict

    source = self._source()
    groups = defaultdict(list)
    for i, (person, household, context) in enumerate(
        zip(people_list, households_list, contexts_list)
    ):
        probs = source.lookup(person, household, context)
        groups[tuple(sorted(probs.items()))].append((i, person, probs))

    results = [None] * len(people_list)
    for members in groups.values():
        _, person0, probs = members[0]
        sampled = self._weighted_draw(probs, person0, size=len(members))
        for (idx, _person, _probs), value in zip(members, sampled):
            results[idx] = value
    return results

InheritanceStrategy

Bases: LogicBlockStrategy

Forward inheritance: Parent → Child.

Children inherit attribute values from parents based on declarative logic blocks. Example for ethnicity: same parents → that value (values[0]), differing parents → M (Mixed).

Source code in may/attribute_assignment/strategies.py

class InheritanceStrategy(LogicBlockStrategy):
    """
    Forward inheritance: Parent → Child.

    Children inherit attribute values from parents based on declarative logic
    blocks. Example for ethnicity: same parents → that value (`values[0]`),
    differing parents → `M` (Mixed).
    """

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """Assign a value by inheriting from the configured parent roles."""
        attribute_name = context.get('attribute_name')
        parent_roles = self.inherit_config.get('roles', [])

        parent_values = []
        for role_name in parent_roles:
            parent = self._get_person_by_role(context, role_name)
            if parent:
                value = self._get_attribute_value(parent, attribute_name)
                if value is not None:
                    parent_values.append(value)

        if not parent_values:
            return self._marginal_assign(person, household, context)

        return self._run_logic(parent_values, person, household, context, attribute_name)

assign(person, household, context)

Assign a value by inheriting from the configured parent roles.

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """Assign a value by inheriting from the configured parent roles."""
    attribute_name = context.get('attribute_name')
    parent_roles = self.inherit_config.get('roles', [])

    parent_values = []
    for role_name in parent_roles:
        parent = self._get_person_by_role(context, role_name)
        if parent:
            value = self._get_attribute_value(parent, attribute_name)
            if value is not None:
                parent_values.append(value)

    if not parent_values:
        return self._marginal_assign(person, household, context)

    return self._run_logic(parent_values, person, household, context, attribute_name)

LogicBlockStrategy

Bases: AssignmentStrategy

Base for the inheritance strategies that drive assignment from declarative when/then logic blocks.

A when is a structured predicate (see _classify_when) evaluated against the collected source values and role context. A then is a literal, the values[0] token, a {value: ...} / {copy: ...} block, or a nested strategy block (see _classify_then). The first block whose when matches wins; if none match, assignment fails loudly.

Source code in may/attribute_assignment/strategies.py

class LogicBlockStrategy(AssignmentStrategy):
    """
    Base for the inheritance strategies that drive assignment from declarative
    `when`/`then` logic blocks.

    A `when` is a structured predicate (see `_classify_when`) evaluated against
    the collected source values and role context. A `then` is a literal, the
    `values[0]` token, a `{value: ...}` / `{copy: ...}` block, or a nested
    strategy block (see `_classify_then`). The first block whose `when` matches
    wins; if none match, assignment fails loudly.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        super().__init__(config, data_manager)
        self.inherit_config = config.get('inherit_from', {})
        self.logic_blocks = config.get('logic', [])
        # Precompile blocks once per strategy instance (instances are cached and
        # reused across every person, see assigner._get_or_create_strategy), so
        # the per-call `assign` path only evaluates the precompiled blocks:
        # predicate classification, expression compilation, and nested-strategy
        # construction all happen here, once, up front.
        self._compiled = [self._compile_block(b) for b in self.logic_blocks]

    def _compile_block(self, block: Dict[str, Any]) -> '_CompiledBlock':
        """Classify and pre-build a logic block once (raises on bad shapes)."""
        when = block.get('when')
        then = block.get('then')
        when_kind = _classify_when(when)
        then_kind = _classify_then(then)
        nested = None
        if then_kind == 'strategy' and then.get('strategy') == 'probabilistic':
            nested = ProbabilisticStrategy(then, self.data_manager)
        return _CompiledBlock(when_kind, when, then_kind, then, nested)

    def _resolve_role_attr(self, role: str, attr: str, context: Dict[str, Any]) -> Any:
        """Resolve a `role.attribute` reference to its assigned value, or raise."""
        person = self._get_person_by_role(context, role)
        if person is None:
            raise ValueError(f"logic predicate references role '{role}' absent from context")
        value = self._get_attribute_value(person, attr)
        if value is None:
            raise ValueError(f"logic predicate: role '{role}' has no '{attr}' assigned")
        return value

    def _evaluate_when(self, block: '_CompiledBlock', source_values: List[Any],
                       context: Dict[str, Any]) -> bool:
        """Evaluate a precompiled `when` predicate."""
        kind, when = block.when_kind, block.when
        if kind == 'unique_count':
            return len(set(source_values)) == when['unique_count']
        if kind == 'unique_count_at_least':
            return len(set(source_values)) >= when['unique_count_at_least']
        # role_attr
        value = self._resolve_role_attr(when['role'], when['attr'], context)
        if 'equals' in when:
            return value == when['equals']
        return value in when['in']

    def _resolve_then(self, block: '_CompiledBlock', source_values: List[Any], person,
                      household, context: Dict[str, Any], attribute_name: str) -> Any:
        """Produce the value for a matched precompiled `then` action."""
        kind, then = block.then_kind, block.then
        if kind == 'value':
            return then['value']
        if kind == 'copy':
            spec = then['copy']
            return self._resolve_role_attr(spec['role'], spec['attr'], context)
        if kind == 'strategy':
            return self._run_nested_strategy(block, person, household, context, attribute_name)
        if kind == 'values[0]':
            if not source_values:
                return self._fail(person, "'then: values[0]' but no source values collected")
            return source_values[0]
        # literal
        return then

    def _run_nested_strategy(self, block: '_CompiledBlock', person, household,
                             context: Dict[str, Any], attribute_name: str) -> Any:
        """Run a precompiled nested strategy block, honouring any `exclude` constraint."""
        if block.nested_strategy is None:
            return self._fail(
                person,
                f"unsupported nested strategy '{block.then.get('strategy')}' in logic block",
            )
        excluded_values = self._resolve_exclude_values(
            block.then.get('exclude', []), context, attribute_name
        )
        if excluded_values:
            return self._sample_with_exclusion(person, household, context, block.then, excluded_values)
        return block.nested_strategy.assign(person, household, context)

    def _run_logic(self, source_values: List[Any], person, household,
                   context: Dict[str, Any], attribute_name: str) -> Any:
        """First block whose `when` matches wins; no match fails loudly."""
        for block in self._compiled:
            if self._evaluate_when(block, source_values, context):
                result = self._resolve_then(
                    block, source_values, person, household, context, attribute_name,
                )
                logger.debug(f"{self.strategy_type}: {result} for {person.id}")
                return result
        return self._fail(person, "no logic block matched")

    def _resolve_exclude_values(self, exclude_refs: List[str],
                                context: Dict[str, Any],
                                attribute_name: str) -> set:
        """
        Resolve exclude references like ["primary_elder.ethnicity"] into
        concrete values by looking up the referenced role persons in context.

        A referenced role that is not yet in context (e.g. the first elder when
        assigning the second) contributes nothing to exclude, because there is no value
        to differ from yet. That is primary logic, not a fallback.

        Args:
            exclude_refs: List of "role.attribute" reference strings
            context: Assignment context containing role persons
            attribute_name: Current attribute being assigned

        Returns:
            Set of concrete values to exclude (may be empty)
        """
        excluded = set()
        for ref in exclude_refs:
            # Parse "role_name.attribute_name" format
            parts = ref.split('.', 1)
            if len(parts) == 2:
                role_name, attr_name = parts
            else:
                role_name = ref
                attr_name = attribute_name

            person = self._get_person_by_role(context, role_name)
            if person:
                value = self._get_attribute_value(person, attr_name)
                if value is not None:
                    excluded.add(value)
                    logger.debug(f"Exclude: resolved '{ref}' → '{value}'")
                else:
                    logger.debug(f"Exclude: '{ref}' has no value assigned yet, skipping")
            else:
                logger.debug(f"Exclude: role '{role_name}' not found in context, skipping")

        return excluded

    def _sample_with_exclusion(self, person, household, context: Dict[str, Any],
                                strategy_config: Dict[str, Any],
                                excluded_values: set) -> Any:
        """
        Sample from a probabilistic distribution while excluding specific values.

        Gets the full distribution, removes excluded values, re-normalizes, and
        samples. If every value is excluded, that is a data/config contradiction
        and the sampler fails loudly.

        Args:
            person: Person being assigned
            household: Household venue
            context: Assignment context
            strategy_config: Probabilistic strategy config dict
            excluded_values: Set of values to exclude from sampling

        Returns:
            Sampled attribute value
        """
        data_source_name = strategy_config.get('data_source', 'geo_distribution')
        source = self.data_manager.get_source(data_source_name)
        if not source:
            raise KeyError(
                f"Data source '{data_source_name}' is not registered. No fallbacks."
            )

        # The source resolves its own key and returns the full distribution.
        probs = source.lookup(person, household, context)
        filtered_probs = {k: v for k, v in probs.items() if k not in excluded_values}
        if not filtered_probs:
            return self._fail(
                person,
                f"all values excluded (excluded={excluded_values}, "
                f"available={set(probs.keys())})",
            )

        sampled = self._weighted_draw(filtered_probs, person)
        logger.debug(
            f"Inheritance (with exclusion): {sampled} for {person.id} "
            f"(excluded={excluded_values})"
        )
        return sampled

PartnershipStrategy

Bases: AssignmentStrategy

Partnership-based assignment using pair probabilities.

Given the first person's attribute value, samples the second person's value from conditional probability distribution. Used for couples and family secondary adults.

Source code in may/attribute_assignment/strategies.py

class PartnershipStrategy(AssignmentStrategy):
    """
    Partnership-based assignment using pair probabilities.

    Given the first person's attribute value, samples the second person's value
    from conditional probability distribution. Used for couples and family secondary adults.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        """Initialize partnership strategy."""
        super().__init__(config, data_manager)
        self.data_source_name = config.get('data_source', 'pair_probabilities')
        self.partner_role = config.get('partner_role', 'primary_adult')

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """
        Sample partner attribute value based on first person's attribute value.

        Args:
            person: Person object (the partner being assigned)
            household: Household object
            context: Assignment context (must contain partner_role person)

        Returns:
            Sampled attribute value
        """
        first_person = self._get_person_by_role(context, self.partner_role)
        if not first_person:
            logger.warning(f"Partner role '{self.partner_role}' not found in context")
            return self._marginal_assign(person, household, context)

        attribute_name = context.get('attribute_name')
        first_value = self._get_attribute_value(first_person, attribute_name)
        if first_value is None:
            logger.warning(f"No {attribute_name} found for {self.partner_role}")
            return self._marginal_assign(person, household, context)

        if not household or not household.geographical_unit:
            logger.warning("No geographical unit found for household")
            return self._fail(person, "no geographical_unit available")

        geo_unit = household.geographical_unit.name

        probs = self.data_manager.lookup(self.data_source_name, geo_unit, first_value)
        if not probs:
            logger.warning(f"No pair probabilities for {geo_unit}, {first_value}")
            return self._fail(person, "data source returned no distribution")

        values = list(probs.keys())
        probabilities = list(probs.values())
        sampled = np.random.choice(values, p=probabilities)

        logger.debug(f"Partnership: {sampled} (partner of {first_value}) for {person.id}")
        return sampled

__init__(config, data_manager)

Initialize partnership strategy.

Source code in may/attribute_assignment/strategies.py

def __init__(self, config: Dict[str, Any], data_manager):
    """Initialize partnership strategy."""
    super().__init__(config, data_manager)
    self.data_source_name = config.get('data_source', 'pair_probabilities')
    self.partner_role = config.get('partner_role', 'primary_adult')

assign(person, household, context)

Sample partner attribute value based on first person's attribute value.

Parameters:

Name	Type	Description	Default
person		Person object (the partner being assigned)	required
household		Household object	required
context	Dict[str, Any]	Assignment context (must contain partner_role person)	required

Returns:

Type	Description
Any	Sampled attribute value

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """
    Sample partner attribute value based on first person's attribute value.

    Args:
        person: Person object (the partner being assigned)
        household: Household object
        context: Assignment context (must contain partner_role person)

    Returns:
        Sampled attribute value
    """
    first_person = self._get_person_by_role(context, self.partner_role)
    if not first_person:
        logger.warning(f"Partner role '{self.partner_role}' not found in context")
        return self._marginal_assign(person, household, context)

    attribute_name = context.get('attribute_name')
    first_value = self._get_attribute_value(first_person, attribute_name)
    if first_value is None:
        logger.warning(f"No {attribute_name} found for {self.partner_role}")
        return self._marginal_assign(person, household, context)

    if not household or not household.geographical_unit:
        logger.warning("No geographical unit found for household")
        return self._fail(person, "no geographical_unit available")

    geo_unit = household.geographical_unit.name

    probs = self.data_manager.lookup(self.data_source_name, geo_unit, first_value)
    if not probs:
        logger.warning(f"No pair probabilities for {geo_unit}, {first_value}")
        return self._fail(person, "data source returned no distribution")

    values = list(probs.keys())
    probabilities = list(probs.values())
    sampled = np.random.choice(values, p=probabilities)

    logger.debug(f"Partnership: {sampled} (partner of {first_value}) for {person.id}")
    return sampled

ProbabilisticConditionsStrategy

Bases: AssignmentStrategy

Assigns a set of conditions (e.g. comorbidities) to a person.

Two selection_methods:

• independent_bernoulli: each condition is an independent Bernoulli trial on its marginal probability, using the marginals only.
• gated_conditions: gated hierarchical sampler that honors the joint count structure (no_condition = P(0), has_comorbidity = P(>=1), multiple_morbidities = P(>=2)), then draws which conditions from the per-condition marginals. See _sample_gated_conditions.

Source code in may/attribute_assignment/strategies.py

class ProbabilisticConditionsStrategy(AssignmentStrategy):
    """
    Assigns a set of conditions (e.g. comorbidities) to a person.

    Two `selection_method`s:

    - `independent_bernoulli`: each condition is an independent Bernoulli trial
      on its marginal probability, using the marginals only.
    - `gated_conditions`: gated hierarchical sampler that honors the
      joint count structure (`no_condition` = P(0), `has_comorbidity` = P(>=1),
      `multiple_morbidities` = P(>=2)), then draws which conditions from the
      per-condition marginals. See `_sample_gated_conditions`.
    """

    def __init__(self, config: Dict[str, Any], data_manager):
        """Initialize probabilistic conditions strategy."""
        super().__init__(config, data_manager)
        self.strategy_type = "probabilistic_conditions"
        self.conditions = config.get('conditions', [])
        # A config must declare its selection_method (validated at load by
        # validate_assignment_config); the dispatch in assign() fails loudly if
        # an unknown value reaches it via direct construction.
        self.selection_method = config.get('selection_method')
        self._distribution_cache = {}

    def assign(self, person, household, context: Dict[str, Any]) -> List[str]:
        """
        Assign comorbidities to person.

        Args:
            person: Person object
            household: Household venue (optional)
            context: Assignment context

        Returns:
            List of condition names (e.g., ["cvd", "crd"])
        """
        data_source_name = self.config.get('data_source')
        if not data_source_name:
            self._fail(person, "probabilistic_conditions has no 'data_source'")

        source = self.data_manager.get_source(data_source_name)
        if not source:
            self._fail(person, f"data source '{data_source_name}' not found")

        # Perform lookup (raises on a miss, with no fallbacks)
        probabilities = source.lookup(person, household, context)

        if self.selection_method == 'independent_bernoulli':
            return self._sample_independent_bernoulli(probabilities)
        if self.selection_method == 'gated_conditions':
            return self._sample_gated_conditions(person, probabilities)
        self._fail(person, f"unknown selection_method '{self.selection_method}'")

    def _sample_independent_bernoulli(self, probabilities: Dict[str, float]) -> List[str]:
        """
        Sample conditions independently using Bernoulli trials.

        Each condition is checked independently with its probability.

        Args:
            probabilities: Dict mapping condition names to probabilities

        Returns:
            List of condition names that were sampled
        """
        selected_conditions = []

        for condition in self.conditions:
            condition_name = condition.get('name')
            if not condition_name:
                continue

            probability = probabilities.get(condition_name, 0.0)

            if np.random.random() < probability:
                selected_conditions.append(condition_name)

        return selected_conditions

    def _sample_gated_conditions(self, person, probabilities: Dict[str, float]) -> List[str]:
        """
        Gated hierarchical comorbidity sampler.

        Honors the joint count structure carried by the data exactly:
          P(0)  = no_condition
          P(>=1) = has_comorbidity      (so P(1) = has_comorbidity - multiple)
          P(>=2) = multiple_morbidities

        1. Draw a count tier {0, 1, >=2} from those joint probabilities.
        2. Draw that many *distinct* conditions weighted by the per-condition
           marginals, without replacement.
        3. For the >=2 tier, draw the actual count from the Poisson-binomial
           distribution implied by the per-condition marginals, conditioned on
           >=2. The data fixes P(>=2) but not the upper tail, so this is the
           explicit modelling assumption for the tail shape.

        Missing or contradictory data fails loudly, with no fallbacks.
        """
        dist = self._distribution(probabilities, person)

        drawn_tier = int(dist['tier_cdf'].searchsorted(np.random.random(), side='right'))
        if drawn_tier == 0:
            return []

        margins, valid, weights, weights_cdf = self._condition_weights(
            dist, probabilities, person
        )

        count = 1 if drawn_tier == 1 else self._sample_multi_count(person, dist, margins)
        return self._pick_distinct_conditions(person, valid, weights, weights_cdf, count)

    def _distribution(self, probabilities: Dict[str, float], person) -> Dict[str, Any]:
        """
        Count-tier distribution for one data row, shared by everyone that row covers.

        The data manager hands back one cached dict per demographic key, so
        keying on that dict's identity gives one entry per row. The dict itself
        is held in the entry to keep its id valid.
        """
        entry = self._distribution_cache.get(id(probabilities))
        if entry is not None:
            return entry[1]

        p_none = self._require_prob(probabilities, 'no_condition', person)
        p_any = self._require_prob(probabilities, 'has_comorbidity', person)
        p_multi = self._require_prob(probabilities, 'multiple_morbidities', person)

        p_one = p_any - p_multi
        if p_one < -_PROB_TOL:
            self._fail(
                person,
                f"has_comorbidity ({p_any}) < multiple_morbidities ({p_multi}); "
                "P(exactly 1 condition) would be negative",
            )

        tier = np.clip(np.array([p_none, p_one, p_multi], dtype=float), 0.0, None)
        total = tier.sum()
        if total <= 0:
            self._fail(person, "comorbidity count-tier probabilities sum to zero")
        tier /= total

        dist = {'tier_cdf': _draw_cdf(tier), 'weights': None, 'multi': None}

        # A source that returns a fresh dict per person would otherwise grow
        # this to one entry per person, so cap it. Sources that cache their
        # rows stay well under the cap and keep hitting.
        if len(self._distribution_cache) >= _DISTRIBUTION_CACHE_LIMIT:
            self._distribution_cache.clear()
        self._distribution_cache[id(probabilities)] = (probabilities, dist)
        return dist

    def _condition_weights(self, dist: Dict[str, Any],
                           probabilities: Dict[str, float], person):
        """
        Per-condition marginals for one data row, resolved on first draw above tier 0.

        Returns the marginals, the names carrying positive probability, and the
        normalized weights over those names with their draw CDF.
        """
        cached = dist['weights']
        if cached is not None:
            return cached

        names = [c['name'] for c in self.conditions if c.get('name')]
        margins = np.clip(
            np.array([self._require_prob(probabilities, n, person) for n in names], dtype=float),
            0.0, None,
        )

        positive = margins > 0
        valid = [name for name, ok in zip(names, positive) if ok]

        weights = weights_cdf = None
        if valid:
            weights = margins[positive]
            weights = weights / weights.sum()
            weights_cdf = _draw_cdf(weights)

        cached = (margins, valid, weights, weights_cdf)
        dist['weights'] = cached
        return cached

    def _require_prob(self, probabilities: Dict[str, float], key: str, person) -> float:
        """Read a probability the gated sampler depends on, or fail loudly."""
        if key not in probabilities:
            self._fail(person, f"gated_conditions requires '{key}' from the data source")
        return float(probabilities[key])

    def _sample_multi_count(self, person, dist: Dict[str, Any], margins: np.ndarray) -> int:
        """
        Draw a condition count >=2 from the Poisson-binomial of the per-condition
        marginals, conditioned on >=2.
        """
        cached = dist['multi']
        if cached is None:
            # Poisson-binomial PMF over the marginals: convolve each [1-p, p].
            pmf = np.array([1.0])
            for p in margins:
                pmf = np.convolve(pmf, [1.0 - p, p])

            tail = pmf[2:]
            total = tail.sum()
            if total <= 0:
                self._fail(
                    person,
                    "data implies >=2 comorbidities but the per-condition marginals "
                    "cannot produce two or more",
                )
            cached = (np.arange(2, len(pmf)), _draw_cdf(tail / total))
            dist['multi'] = cached

        counts, cdf = cached
        return int(counts[cdf.searchsorted(np.random.random(), side='right')])

    def _pick_distinct_conditions(self, person, valid: List[str], weights,
                                  weights_cdf, count: int) -> List[str]:
        """Pick `count` distinct conditions weighted by their marginals, no replacement."""
        if count > len(valid):
            self._fail(
                person,
                f"cannot draw {count} distinct conditions from {len(valid)} "
                "with positive probability",
            )

        if count == 1:
            return [valid[weights_cdf.searchsorted(np.random.rand(1), side='right')[0]]]

        chosen = np.random.choice(valid, size=count, replace=False, p=weights)
        return [str(c) for c in chosen]

__init__(config, data_manager)

Initialize probabilistic conditions strategy.

Source code in may/attribute_assignment/strategies.py

def __init__(self, config: Dict[str, Any], data_manager):
    """Initialize probabilistic conditions strategy."""
    super().__init__(config, data_manager)
    self.strategy_type = "probabilistic_conditions"
    self.conditions = config.get('conditions', [])
    # A config must declare its selection_method (validated at load by
    # validate_assignment_config); the dispatch in assign() fails loudly if
    # an unknown value reaches it via direct construction.
    self.selection_method = config.get('selection_method')
    self._distribution_cache = {}

assign(person, household, context)

Assign comorbidities to person.

Parameters:

Name	Type	Description	Default
person		Person object	required
household		Household venue (optional)	required
context	Dict[str, Any]	Assignment context	required

Returns:

Type	Description
List[str]	List of condition names (e.g., ["cvd", "crd"])

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> List[str]:
    """
    Assign comorbidities to person.

    Args:
        person: Person object
        household: Household venue (optional)
        context: Assignment context

    Returns:
        List of condition names (e.g., ["cvd", "crd"])
    """
    data_source_name = self.config.get('data_source')
    if not data_source_name:
        self._fail(person, "probabilistic_conditions has no 'data_source'")

    source = self.data_manager.get_source(data_source_name)
    if not source:
        self._fail(person, f"data source '{data_source_name}' not found")

    # Perform lookup (raises on a miss, with no fallbacks)
    probabilities = source.lookup(person, household, context)

    if self.selection_method == 'independent_bernoulli':
        return self._sample_independent_bernoulli(probabilities)
    if self.selection_method == 'gated_conditions':
        return self._sample_gated_conditions(person, probabilities)
    self._fail(person, f"unknown selection_method '{self.selection_method}'")

ReverseInheritanceStrategy

Bases: LogicBlockStrategy

Reverse inheritance: Child → Parent.

When children are assigned first, infer a parent's attribute value from the child's, via declarative logic blocks. Example for ethnicity: child W/A/B/O → parent copies it; child M → parents differ (nested probabilistic draw, with an optional exclude to force a second parent to differ from the first).

Source code in may/attribute_assignment/strategies.py

class ReverseInheritanceStrategy(LogicBlockStrategy):
    """
    Reverse inheritance: Child → Parent.

    When children are assigned first, infer a parent's attribute value from the
    child's, via declarative logic blocks. Example for ethnicity: child W/A/B/O →
    parent copies it; child M → parents differ (nested probabilistic draw, with an
    optional `exclude` to force a second parent to differ from the first).
    """

    def assign(self, person, household, context: Dict[str, Any]) -> Any:
        """Assign a value by inferring it from the configured child role."""
        attribute_name = context.get('attribute_name')

        child_role = self.inherit_config.get('role')
        if not child_role:
            return self._fail(person, "no child role configured for reverse inheritance")

        child = self._get_person_by_role(context, child_role)
        if not child:
            return self._marginal_assign(person, household, context)

        child_value = self._get_attribute_value(child, attribute_name)
        if child_value is None:
            return self._marginal_assign(person, household, context)

        return self._run_logic([child_value], person, household, context, attribute_name)

assign(person, household, context)

Assign a value by inferring it from the configured child role.

Source code in may/attribute_assignment/strategies.py

def assign(self, person, household, context: Dict[str, Any]) -> Any:
    """Assign a value by inferring it from the configured child role."""
    attribute_name = context.get('attribute_name')

    child_role = self.inherit_config.get('role')
    if not child_role:
        return self._fail(person, "no child role configured for reverse inheritance")

    child = self._get_person_by_role(context, child_role)
    if not child:
        return self._marginal_assign(person, household, context)

    child_value = self._get_attribute_value(child, attribute_name)
    if child_value is None:
        return self._marginal_assign(person, household, context)

    return self._run_logic([child_value], person, household, context, attribute_name)

StrategyFactory

Factory for creating strategy instances.

Maps strategy type strings to strategy classes.

Source code in may/attribute_assignment/strategies.py

class StrategyFactory:
    """
    Factory for creating strategy instances.

    Maps strategy type strings to strategy classes.
    """

    _strategy_map = {
        'probabilistic': ProbabilisticStrategy,
        'partnership': PartnershipStrategy,
        'inheritance': InheritanceStrategy,
        'reverse_inheritance': ReverseInheritanceStrategy,
        'probabilistic_conditions': ProbabilisticConditionsStrategy,
        'commuting_likelihood': CommutingLikelihoodStrategy,
        'geographical_unit_sampler': GUSamplerStrategy,
        'categorical_sampler': CategoricalSamplerStrategy,
        'constant': ConstantStrategy,
    }

    @classmethod
    def create_strategy(cls, config: Dict[str, Any], data_manager) -> AssignmentStrategy:
        """
        Create strategy instance from configuration.

        Args:
            config: Strategy configuration dict
            data_manager: DataSourceManager instance

        Returns:
            Strategy instance

        Raises:
            ValueError: If strategy type is unknown
        """
        validate_assignment_config(config)

        strategy_class = cls._strategy_map.get(config['strategy'])
        return strategy_class(config, data_manager)

create_strategy(config, data_manager) classmethod

Create strategy instance from configuration.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Strategy configuration dict	required
data_manager		DataSourceManager instance	required

Returns:

Type	Description
AssignmentStrategy	Strategy instance

Raises:

Type	Description
ValueError	If strategy type is unknown

Source code in may/attribute_assignment/strategies.py

@classmethod
def create_strategy(cls, config: Dict[str, Any], data_manager) -> AssignmentStrategy:
    """
    Create strategy instance from configuration.

    Args:
        config: Strategy configuration dict
        data_manager: DataSourceManager instance

    Returns:
        Strategy instance

    Raises:
        ValueError: If strategy type is unknown
    """
    validate_assignment_config(config)

    strategy_class = cls._strategy_map.get(config['strategy'])
    return strategy_class(config, data_manager)

validate_assignment_config(config, where='assignment')

Reject assignment config keys no strategy reads.

Raises:

Type	Description
ValueError	naming the unknown keys, the strategy, and the allowed

Source code in may/attribute_assignment/strategies.py

def validate_assignment_config(config: Dict[str, Any], where: str = "assignment") -> None:
    """
    Reject assignment config keys no strategy reads.

    Raises:
        ValueError: naming the unknown keys, the strategy, and the allowed
        set, so a stale key (e.g. `context`) breaks the build at load time.
    """
    if not isinstance(config, dict):
        raise ValueError(f"{where}: assignment must be a mapping, got {type(config).__name__}")

    strategy_type = config.get('strategy')
    if not strategy_type:
        raise ValueError(f"{where}: assignment has no 'strategy' field")

    allowed = STRATEGY_ALLOWED_KEYS.get(strategy_type)
    if allowed is None:
        raise ValueError(
            f"{where}: unknown strategy '{strategy_type}' "
            f"(known: {sorted(STRATEGY_ALLOWED_KEYS)})"
        )

    unknown = set(config) - allowed
    if unknown:
        raise ValueError(
            f"{where}: strategy '{strategy_type}' does not read key(s) "
            f"{sorted(unknown)} — allowed keys are {sorted(allowed)}. "
            f"Remove them (dead config) or fix the typo."
        )

    if strategy_type == 'probabilistic_conditions':
        method = config.get('selection_method')
        if method is None:
            raise ValueError(
                f"{where}: strategy 'probabilistic_conditions' requires "
                f"'selection_method' — declare one of "
                f"{sorted(_CONDITION_SELECTION_METHODS)} (no implicit default)."
            )
        if method not in _CONDITION_SELECTION_METHODS:
            raise ValueError(
                f"{where}: unknown selection_method '{method}' "
                f"(known: {sorted(_CONDITION_SELECTION_METHODS)})."
            )

    is_logic_strategy = strategy_type in _LOGIC_STRATEGIES
    for i, entry in enumerate(config.get('logic') or []):
        entry = entry or {}
        unknown = set(entry) - _LOGIC_ENTRY_KEYS
        if unknown:
            raise ValueError(
                f"{where}.logic[{i}]: unknown key(s) {sorted(unknown)} — "
                f"allowed: {sorted(_LOGIC_ENTRY_KEYS)}"
            )
        then = entry.get('then')
        if isinstance(then, dict):
            unknown = set(then) - _THEN_BLOCK_KEYS
            if unknown:
                raise ValueError(
                    f"{where}.logic[{i}].then: unknown key(s) {sorted(unknown)} — "
                    f"allowed: {sorted(_THEN_BLOCK_KEYS)}"
                )
        if is_logic_strategy:
            try:
                _classify_when(entry.get('when'))
                _classify_then(then)
            except ValueError as exc:
                raise ValueError(f"{where}.logic[{i}]: {exc}") from exc