Brothel movement model
The brothel movement model, where victims moved between brothels leave traces in bills and train trips.
- Name:
trafficking - Field:
social/organizations - Formalism: agent-based model
- References: Chesney 2021
Factory
@register_model(
field='social/organizations',
tags=('trafficking', 'agents', 'grid', 'crime'),
references=('Chesney 2021',),
defaults={'until': 90},
)
def trafficking(
width: int = 30,
height: int = 30,
*,
n_cities: int = 3,
n_houses: int = 40,
n_brothels: int = 3,
n_victims: int = 12,
n_free: int = 30,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return the brothel movement model, where victims moved between brothels leave traces in bills and train trips.
Args:
width:
The number of columns of the grid.
height:
The number of rows of the grid.
n_cities:
The number of cities, vertical bands of equal width.
n_houses:
The number of houses.
n_brothels:
The number of brothels.
n_victims:
The number of victims, spread over the brothels.
n_free:
The number of women who travel freely.
random_state:
The seed or generator of the places, the women, and the efficiencies.
Returns:
system:
The model, with the groups `premises` and `women`.
Examples:
>>> from skcomplex.models.social.organizations import trafficking
>>> [group.size for group in trafficking(n_houses=5, n_brothels=2, n_victims=4, n_free=3).elements]
[7, 7]
"""
rng = np.random.default_rng(random_state)
n_places = n_houses + n_brothels
cells = rng.permutation(width * height)[:n_places]
places = [(int(cell % width), int(cell // width)) for cell in cells]
brothel = (np.arange(n_places) >= n_houses).astype(float)
premises = GroupElement(
'premises',
element=PremisesElement,
size=n_places,
values={'brothel': brothel, 'efficiency': rng.uniform(0, 3, n_places)},
location=places,
)
homes = [places[n_houses + index % n_brothels] for index in range(n_victims)]
homes += [(int(rng.integers(width)), int(rng.integers(height))) for _ in range(n_free)]
victim = (np.arange(n_victims + n_free) < n_victims).astype(float)
women = GroupElement(
'women',
element=TravellerElement,
size=n_victims + n_free,
values={'victim': victim},
location=homes,
)
day = DailyMovementsInteraction(
'day',
women='women',
premises='premises',
n_cities=n_cities,
phase=DAILY,
victims_phase=MONTHLY,
)
bills = [
MapInteraction(
'days',
target='premises',
state='day',
function=_count_day,
mode='increment',
form='agent',
phase=BILLING,
),
MapInteraction(
'bills',
target='premises',
state='bills',
function=_heat,
sources=['women'],
mode='increment',
form='agent',
phase=BILLING,
),
MapInteraction(
'record',
target='premises',
state='last_bills',
function=_record,
form='agent',
phase=QUARTERLY,
),
MapInteraction('reset', target='premises', state='bills', function=_reset, form='agent', phase=QUARTERLY),
]
phases = [
PhaseSpec(DAILY, activation='sequential'),
PhaseSpec(MONTHLY, activation='sequential', period=MONTH),
PhaseSpec(BILLING, activation='sequential'),
PhaseSpec(QUARTERLY, activation='sequential', period=QUARTER),
]
return System(
elements=[premises, women],
interactions=[day, *bills],
space=LatticeSpace((width, height), boundary='fixed'),
time=DiscreteTime(schedule=PhasesScheduleSpec(phases)),
)
Components
class PremisesElement(Element):
"""House or brothel of the brothel movement model, whose energy bills grow with the people inside.
Every day the bills grow by `1 / (1 + exp(x0 - n))` for the `n` people in the premises' cell, and every quarter
they are recorded and reset. In `trafficking`, maps of the bills read the counts of the women by cell every day,
and maps in a phase of a period of 90 days record and reset them.
Args:
label:
The label of the agent or group.
visual:
How the agent is drawn, or `None` for `default_visual_`.
brothel:
Whether the premises are a brothel, 1, or a house, 0.
bills:
The bills of the current quarter.
last_bills:
The bills of the last full quarter.
efficiency:
The midpoint `x0` of the sigmoid of the bills.
day:
The number of days run.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.social.organizations import PremisesElement
>>> group = GroupElement('premises', element=PremisesElement, size=3, values={'brothel': 1.0})
>>> group.label, group.element.__name__, group.size
('premises', 'PremisesElement', 3)
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'brothel', 'color_range': (0.0, 1.0)}),
*,
brothel: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
bills: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
last_bills: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
efficiency: Annotated[float, StateSpec()] = 1.0,
day: Annotated[float, StateSpec(domain=NN)] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.brothel = brothel
self.bills = bills
self.last_bills = last_bills
self.efficiency = efficiency
self.day = day
@classmethod
def _get_view(cls) -> type[PremisesView]:
"""Return the class of the views of the agents, whose attributes are the states and parameters."""
return PremisesView
class TravellerElement(Element):
"""Woman of the brothel movement model: a trafficking victim, or a woman who travels freely.
Args:
label:
The label of the agent or group.
visual:
How the agent is drawn, or `None` for `default_visual_`.
victim:
Whether the woman is a trafficking victim, 1, or not, 0.
trips:
The number of train trips between cities.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.social.organizations import TravellerElement
>>> group = GroupElement('women', element=TravellerElement, size=4, values={'victim': 1.0})
>>> group.label, group.element.__name__, group.size
('women', 'TravellerElement', 4)
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'victim', 'color_range': (0.0, 1.0)}),
*,
victim: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 0.0,
trips: Annotated[float, StateSpec(domain=NN)] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.victim = victim
self.trips = trips
@classmethod
def _get_view(cls) -> type[TravellerView]:
"""Return the class of the views of the agents, whose attributes are the states and parameters."""
return TravellerView
class DailyMovementsInteraction(Interaction):
"""Movements of the women of the brothel movement model.
Every day one random free woman moves to a random cell of a random city, and in the steps of the phase
`victims_phase` the traffickers move a random number of the victims of every brothel to random brothels. A move
between cities counts as a train trip. The cities are vertical bands of equal width of the grid.
Args:
label:
The label of the interaction.
women:
The path of the group of travellers.
premises:
The path of the group of premises, whose brothels the victims move between.
n_cities:
The number of cities, vertical bands of equal width.
phase:
The phase of the step the free woman moves in, or `None` for the default phase.
victims_phase:
The phase of the step the victims move in, or `None` for the default phase.
Examples:
>>> from skcomplex.models.social.organizations import DailyMovementsInteraction
>>> DailyMovementsInteraction('day', women='women', premises='premises').n_cities
3
"""
def __init__(
self,
label: str,
*,
women: Annotated[str, RoleSpec()],
premises: Annotated[str, RoleSpec()],
n_cities: int = 3,
phase: str | None = None,
victims_phase: str | None = None,
) -> None:
super().__init__(label)
self.women = women
self.premises = premises
self.n_cities = n_cities
self.phase = phase
self.victims_phase = victims_phase
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the rules of the free woman and of the victims.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.move_free, phase=self.phase), RuleLaw(self.move_victims, phase=self.victims_phase)]
def _city(self, location: object, width: int) -> int:
"""Return the city of a cell of a grid of a given width."""
return DailyMovementsInteraction._cell(location)[0] * self.n_cities // width
@staticmethod
def _shape(model: ModelView) -> tuple[int, int]:
"""Return the width and the height of the grid of the model."""
dimensions = model.dimensions
if dimensions is None:
error_msg = 'The trafficking model runs on a grid, and its agents live in no space.'
raise SpaceError(error_msg)
width, height = dimensions
return int(width), int(height)
def _travel(self, woman: TravellerView, cell: tuple[int, int], width: int) -> None:
"""Move a woman to a cell and count a trip when the city changes."""
if self._city(cell, width) != self._city(woman.location, width):
woman.trips += 1
woman.move_to(cell)
def move_free(self, model: ModelView) -> None:
"""Move a random free woman to a random cell of a random city.
Args:
model:
The model.
"""
free = [woman for woman in model.agents('women', TravellerView) if not woman.victim]
if not free:
return
width, height = self._shape(model)
city = int(model.rng.integers(self.n_cities))
start, stop = city * width // self.n_cities, (city + 1) * width // self.n_cities
cell = (int(model.rng.integers(start, stop)), int(model.rng.integers(height)))
self._travel(free[int(model.rng.integers(len(free)))], cell, width)
def move_victims(self, model: ModelView) -> None:
"""Move a random number of the victims of every brothel to random brothels.
Args:
model:
The model.
"""
women = model.agents('women', TravellerView)
brothels = [
DailyMovementsInteraction._cell(place.location)
for place in model.agents('premises', PremisesView)
if place.brothel
]
width = self._shape(model)[0]
for brothel in brothels:
victims = [
woman for woman in women if woman.victim and DailyMovementsInteraction._cell(woman.location) == brothel
]
moved = model.rng.permutation(len(victims))[: int(model.rng.integers(len(victims) + 1))]
for index in moved:
self._travel(victims[index], brothels[int(model.rng.integers(len(brothels)))], width)
@staticmethod
def _cell(location: object) -> tuple[int, int]:
"""Return a grid location as a tuple of integers."""
column, row = (int(item) for item in np.ravel(np.asarray(location)))
return column, row
class MapInteraction(Interaction):
"""Map of a state of discrete time, whose next value is a function, or the state plus it.
In the mode `'replace'` the next value of the state is `f`, and in the mode `'increment'` it is the state plus `f`.
Under the `'synchronous'` schedule every member of a bound group updates at once, from the states at the start of
the step. Under the `'sequential'` schedule the members update one after the other, in their order, and under the
`'random-sequential'` schedule in a new random order every step, each member reading the values the members before
it set. Under the `'member'` schedule of the agent form, every member updates in its own activation, together with
the rules of its element and the other member interactions, in the order the system lists them.
In the `'symbolic'` form, `function(namespace)` returns a symbolic expression, or a vector of them for a vector
state, where `namespace.target` holds the states and parameters of the target, `namespace.sources` the namespaces
of the sources, and `namespace.t` the time, and may hold draws such as `normal('noise')`. In the `'agent'` form,
the target is an agent group and `function(agent, model)` returns the value as a number, or as an array of the
shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the model view.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state the map sets.
state:
The name of the state of the target.
function:
The function `f(namespace)` in the symbolic form, or `f(agent, model)` in the agent form.
sources:
The paths of the elements, or members of groups such as `cells[0]`, that the function reads.
mode:
`'replace'` or `'increment'`.
schedule:
`'synchronous'`, `'sequential'`, `'random-sequential'`, or `'member'` for the agent form.
form:
`'symbolic'` for update laws of the states of the target, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, the
function is not callable, or the agent form returns a value that does not match the state, raised by
`run` of the simulation.
Examples:
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import MapInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
... super().__init__(label)
... self.x = x
>>> logistic = MapInteraction(
... 'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.elements import GroupElement
>>> class WeightsElement(Element):
... def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
... super().__init__(label, visual=visual)
... self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
... 'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).check().prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
mode: str = 'replace',
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.function = function
self.sources = sources
self.mode = mode
self.schedule = schedule
self.form = form
self.phase = phase
def _check(self) -> None:
"""Raise an error for an unknown mode, schedule, or form, or a function that cannot be called.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, or the
function is not callable.
"""
super()._check()
self._check_variant('mode', self.mode, MODES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
):
if value not in variants:
error_msg = f'The {name} of map {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
if self.schedule == 'member' and self.form != 'agent':
error_msg = (
f'The map {self.label!r} updates every member in its activation, which only the agent form does.'
)
raise ValidationError(error_msg)
if not callable(self.function):
error_msg = f'The function of map {self.label!r} must be callable, but it is {self.function!r}.'
raise ValidationError(error_msg)
def _apply_mode(self, current: Value, value: Value) -> Value:
"""Return the next value of a state from its current value and the value of the function, by the mode."""
if self.mode == 'increment':
return current + value
if self.mode == 'replace':
return value
raise self._get_variant_error(self.mode, 'mode', '_apply_mode')
def _get_vector(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of a vector state of an agent."""
current = np.asarray(getattr(agent, self.state), dtype=np.float64)
value = np.asarray(self.function(agent, model), dtype=np.float64)
if value.shape != current.shape:
error_msg = (
f'The function of map {self.label!r} returns the shape {value.shape} for the state {self.state!r} of '
f'the shape {current.shape}. Return an array of the shape of the state.'
)
raise ValidationError(error_msg)
return self._apply_mode(current, value)
def _get_value(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> Value:
"""Return the next value of the state of an agent, a number or the array of a vector state."""
if np.ndim(getattr(agent, self.state)) > 0:
return self._get_vector(agent, model)
found = self.function(agent, model)
if not isinstance(found, SupportsFloat):
error_msg = f'The function of map {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
value = float(found)
return self._apply_mode(float(getattr(agent, self.state)), value)
def _set_value(self, agent: AgentViewProtocol, value: Value) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _update_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, model))
def update(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_value, self._set_value)
def _get_order(self) -> object:
"""Return the order of the members of the symbolic form: none, their own order, or a uniform draw."""
if self.schedule == 'synchronous':
return None
if self.schedule == 'sequential':
return SR(0)
return uniform(re.sub(r'\W', '_', f'order_{self.label}'))
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its next value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.sources`.
Returns:
laws:
The update law of the state, or the rule that runs the map of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._update_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.update, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
mapped = cast(Value, self.function(namespace))
next_value = self._apply_mode(value, mapped)
return [UpdateLaw(value, next_value, increment=self.mode == 'increment', order=self._get_order())]