Skip to content

demographics

Demographics

Bases: Demographics

Source code in emodpy/demographics/demographics.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class Demographics(EMODAPIDemographics):

    def __init__(self, nodes: List[Node], default_node: Node = None, idref: str = None, set_defaults: bool = True):
        super().__init__(nodes=nodes, default_node=default_node, idref=idref, set_defaults=set_defaults)

    # Forces emodpy-layer Demographics instantiation to use Node object-route for default node . Cannot use
    # the old self.raw dict representation of a default node.

    @property
    def raw(self):
        raise AttributeError("raw is not a valid attribute for Demographics objects")

    @raw.setter
    def raw(self, value):
        raise AttributeError("raw is not a valid attribute for Demographics objects")

    def add_migration(self, data, migration_type: Union[MigrationType, str],
                      x_modifier: float = 1.0,
                      interpolation_type: Union[InterpolationType, str] = InterpolationType.PIECEWISE_CONSTANT,
                      migration_pattern: Union[MigrationPattern, str] = MigrationPattern.RANDOM_WALK_DIFFUSION,
                      roundtrip_duration: Optional[float] = None,
                      roundtrip_probability: Optional[float] = None,
                      roundtrip_waypoints: Optional[int] = None,
                      filename: Optional[str] = None,
                      user_notes: Optional[str] = None):
        """Assign migration data to a migration type, write the file, and set config params.

        Args:
            data: MigrationData object containing the rate data
            migration_type: MigrationType enum or string ("LOCAL", "AIR", "REGIONAL", "SEA", "FAMILY")
            x_modifier: migration rate multiplier (x_*_Migration config param). Default 1.0
            interpolation_type: InterpolationType enum or string. Default PIECEWISE_CONSTANT.
            migration_pattern: MigrationPattern enum or string. Default RANDOM_WALK_DIFFUSION.
                SINGLE_ROUND_TRIPS: uses roundtrip_duration and roundtrip_probability.
                WAYPOINTS_HOME: uses roundtrip_waypoints.
                RANDOM_WALK_DIFFUSION: no additional parameters.
            roundtrip_duration: days at destination (*_Roundtrip_Duration).
                Only used with SINGLE_ROUND_TRIPS. Default None.
            roundtrip_probability: probability of return trip (*_Roundtrip_Probability).
                Only used with SINGLE_ROUND_TRIPS. Default None.
            roundtrip_waypoints: max waypoints before returning home (Roundtrip_Waypoints).
                Only used with WAYPOINTS_HOME. Default None.
            filename: output path for the binary file. If None, auto-generates based on migration_type.
            user_notes: free-text description of this migration file — motivation, data source,
                assumptions, etc. We encourage you to record why this file was created so
                the context is preserved for future reference. Stored in the JSON metadata
                sidecar as USER_NOTES.
        """
        if not isinstance(migration_type, MigrationType):
            try:
                migration_type = MigrationType(migration_type.upper())
            except ValueError:
                raise ValueError(f"Invalid migration_type '{migration_type}'. "
                                 f"Valid options: {list(MigrationType)}")
        if not isinstance(interpolation_type, InterpolationType):
            try:
                interpolation_type = InterpolationType(interpolation_type)
            except ValueError:
                raise ValueError(f"Invalid interpolation_type '{interpolation_type}'. "
                                 f"Valid options: {list(InterpolationType)}")
        if not isinstance(migration_pattern, MigrationPattern):
            try:
                migration_pattern = MigrationPattern(migration_pattern.upper())
            except ValueError:
                raise ValueError(f"Invalid migration_pattern '{migration_pattern}'. "
                                 f"Valid options: {list(MigrationPattern)}")

        if migration_pattern != MigrationPattern.SINGLE_ROUND_TRIPS:
            if roundtrip_duration is not None or roundtrip_probability is not None:
                raise ValueError("roundtrip_duration and roundtrip_probability are only valid "
                                 "with migration_pattern=SINGLE_ROUND_TRIPS.")
        if migration_pattern != MigrationPattern.WAYPOINTS_HOME:
            if roundtrip_waypoints is not None:
                raise ValueError("roundtrip_waypoints is only valid "
                                 "with migration_pattern=WAYPOINTS_HOME.")

        valid_ids = {n.id for n in self.nodes if n.id != 0}
        data_ids = set(data.node_ids)
        unknown = data_ids - valid_ids
        if unknown:
            raise ValueError(f"Migration data contains node IDs not in demographics: {sorted(unknown)}")

        if data.idref != self.idref:
            logger.warning(f"MigrationData idref '{data.idref}' does not match demographics "
                           f"idref '{self.idref}', but the Nodes match. Updating migration idref to '{self.idref}'.")
            data._idref = self.idref

        if filename is None:
            filename = f"{str(migration_type).lower()}_migration.bin"
        path = Path(filename).absolute()

        data.to_migration_file(path, migration_type=migration_type, interpolation_type=interpolation_type,
                               user_notes=user_notes)
        self.migration_files.append(path)

        bin_filename = path.name
        self.implicits.append(partial(
            _set_migration_config,
            migration_type=migration_type,
            filename=bin_filename,
            x_modifier=x_modifier,
            migration_pattern=migration_pattern,
            roundtrip_duration=roundtrip_duration,
            roundtrip_probability=roundtrip_probability,
            roundtrip_waypoints=roundtrip_waypoints,
        ))

add_migration(data, migration_type, x_modifier=1.0, interpolation_type=InterpolationType.PIECEWISE_CONSTANT, migration_pattern=MigrationPattern.RANDOM_WALK_DIFFUSION, roundtrip_duration=None, roundtrip_probability=None, roundtrip_waypoints=None, filename=None, user_notes=None)

Assign migration data to a migration type, write the file, and set config params.

Parameters:

Name Type Description Default
data

MigrationData object containing the rate data

 required
migration_type Union[MigrationType, str]

MigrationType enum or string ("LOCAL", "AIR", "REGIONAL", "SEA", "FAMILY")

 required
x_modifier float

migration rate multiplier (x_*_Migration config param). Default 1.0

 1.0
interpolation_type Union[InterpolationType, str]

InterpolationType enum or string. Default PIECEWISE_CONSTANT.

 PIECEWISE_CONSTANT
migration_pattern Union[MigrationPattern, str]

MigrationPattern enum or string. Default RANDOM_WALK_DIFFUSION. SINGLE_ROUND_TRIPS: uses roundtrip_duration and roundtrip_probability. WAYPOINTS_HOME: uses roundtrip_waypoints. RANDOM_WALK_DIFFUSION: no additional parameters.

 RANDOM_WALK_DIFFUSION
roundtrip_duration Optional[float]

days at destination (*_Roundtrip_Duration). Only used with SINGLE_ROUND_TRIPS. Default None.

 None
roundtrip_probability Optional[float]

probability of return trip (*_Roundtrip_Probability). Only used with SINGLE_ROUND_TRIPS. Default None.

 None
roundtrip_waypoints Optional[int]

max waypoints before returning home (Roundtrip_Waypoints). Only used with WAYPOINTS_HOME. Default None.

 None
filename Optional[str]

output path for the binary file. If None, auto-generates based on migration_type.

 None
user_notes Optional[str]

free-text description of this migration file — motivation, data source, assumptions, etc. We encourage you to record why this file was created so the context is preserved for future reference. Stored in the JSON metadata sidecar as USER_NOTES.

 None
Source code in emodpy/demographics/demographics.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def add_migration(self, data, migration_type: Union[MigrationType, str],
                  x_modifier: float = 1.0,
                  interpolation_type: Union[InterpolationType, str] = InterpolationType.PIECEWISE_CONSTANT,
                  migration_pattern: Union[MigrationPattern, str] = MigrationPattern.RANDOM_WALK_DIFFUSION,
                  roundtrip_duration: Optional[float] = None,
                  roundtrip_probability: Optional[float] = None,
                  roundtrip_waypoints: Optional[int] = None,
                  filename: Optional[str] = None,
                  user_notes: Optional[str] = None):
    """Assign migration data to a migration type, write the file, and set config params.

    Args:
        data: MigrationData object containing the rate data
        migration_type: MigrationType enum or string ("LOCAL", "AIR", "REGIONAL", "SEA", "FAMILY")
        x_modifier: migration rate multiplier (x_*_Migration config param). Default 1.0
        interpolation_type: InterpolationType enum or string. Default PIECEWISE_CONSTANT.
        migration_pattern: MigrationPattern enum or string. Default RANDOM_WALK_DIFFUSION.
            SINGLE_ROUND_TRIPS: uses roundtrip_duration and roundtrip_probability.
            WAYPOINTS_HOME: uses roundtrip_waypoints.
            RANDOM_WALK_DIFFUSION: no additional parameters.
        roundtrip_duration: days at destination (*_Roundtrip_Duration).
            Only used with SINGLE_ROUND_TRIPS. Default None.
        roundtrip_probability: probability of return trip (*_Roundtrip_Probability).
            Only used with SINGLE_ROUND_TRIPS. Default None.
        roundtrip_waypoints: max waypoints before returning home (Roundtrip_Waypoints).
            Only used with WAYPOINTS_HOME. Default None.
        filename: output path for the binary file. If None, auto-generates based on migration_type.
        user_notes: free-text description of this migration file — motivation, data source,
            assumptions, etc. We encourage you to record why this file was created so
            the context is preserved for future reference. Stored in the JSON metadata
            sidecar as USER_NOTES.
    """
    if not isinstance(migration_type, MigrationType):
        try:
            migration_type = MigrationType(migration_type.upper())
        except ValueError:
            raise ValueError(f"Invalid migration_type '{migration_type}'. "
                             f"Valid options: {list(MigrationType)}")
    if not isinstance(interpolation_type, InterpolationType):
        try:
            interpolation_type = InterpolationType(interpolation_type)
        except ValueError:
            raise ValueError(f"Invalid interpolation_type '{interpolation_type}'. "
                             f"Valid options: {list(InterpolationType)}")
    if not isinstance(migration_pattern, MigrationPattern):
        try:
            migration_pattern = MigrationPattern(migration_pattern.upper())
        except ValueError:
            raise ValueError(f"Invalid migration_pattern '{migration_pattern}'. "
                             f"Valid options: {list(MigrationPattern)}")

    if migration_pattern != MigrationPattern.SINGLE_ROUND_TRIPS:
        if roundtrip_duration is not None or roundtrip_probability is not None:
            raise ValueError("roundtrip_duration and roundtrip_probability are only valid "
                             "with migration_pattern=SINGLE_ROUND_TRIPS.")
    if migration_pattern != MigrationPattern.WAYPOINTS_HOME:
        if roundtrip_waypoints is not None:
            raise ValueError("roundtrip_waypoints is only valid "
                             "with migration_pattern=WAYPOINTS_HOME.")

    valid_ids = {n.id for n in self.nodes if n.id != 0}
    data_ids = set(data.node_ids)
    unknown = data_ids - valid_ids
    if unknown:
        raise ValueError(f"Migration data contains node IDs not in demographics: {sorted(unknown)}")

    if data.idref != self.idref:
        logger.warning(f"MigrationData idref '{data.idref}' does not match demographics "
                       f"idref '{self.idref}', but the Nodes match. Updating migration idref to '{self.idref}'.")
        data._idref = self.idref

    if filename is None:
        filename = f"{str(migration_type).lower()}_migration.bin"
    path = Path(filename).absolute()

    data.to_migration_file(path, migration_type=migration_type, interpolation_type=interpolation_type,
                           user_notes=user_notes)
    self.migration_files.append(path)

    bin_filename = path.name
    self.implicits.append(partial(
        _set_migration_config,
        migration_type=migration_type,
        filename=bin_filename,
        x_modifier=x_modifier,
        migration_pattern=migration_pattern,
        roundtrip_duration=roundtrip_duration,
        roundtrip_probability=roundtrip_probability,
        roundtrip_waypoints=roundtrip_waypoints,
    ))