Source code for community
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This file is part of the Cortix toolkit environment.
# https://cortix.org
import logging
import numpy as np
import scipy.constants as const
from scipy.integrate import odeint
from cortix import Module
from cortix import Phase
from cortix import Quantity
[docs]class Community(Module):
'''
Community Cortix module used to model criminal group population in a community system.
Community here is the system at large with all possible adult individuals included
in a society.
Notes
-----
These are the `port` names available in this module to connect to respective
modules: `probation`, `adjudication`, `jail`, `prison`, `arrested`, and `parole`.
See instance attribute `port_names_expected`.
'''
[docs] def __init__(self, n_groups=1, non_offender_adult_population=100,
offender_pool_size=0.0, free_offender_pool_size=0.0):
'''
Parameters
----------
n_groups: int
Number of groups in the population.
non_offender_adult_population: float
Pool of individuals reaching the adult age (SI) unit. Default: 100.
offender_pool_size: float
Upperbound on the range of the existing population groups. A random value
from 0 to the upperbound value will be assigned to each group. This is
typically a small number, say a fraction of a percent.
'''
super().__init__()
self.port_names_expected = ['probation','adjudication','jail','prison',
'arrested', 'parole']
quantities = list()
self.ode_params = dict()
self.initial_time = 0.0 * const.day
self.end_time = 100 * const.day
self.time_step = 0.5 * const.day
self.show_time = (False,10*const.day)
self.log = logging.getLogger('cortix')
# Population groups
self.n_groups = n_groups
# Community offender population groups removed from circulation
f0g_0 = np.random.random(self.n_groups) * offender_pool_size
f0g = Quantity(name='f0g', formalName='offender-pop-grps',
unit='individual', value=f0g_0)
quantities.append(f0g)
# Community free-offender population groups in freedom
f0g_free_0 = np.random.random(self.n_groups) * free_offender_pool_size
f0g_free = Quantity(name='f0g_free', formalName='free-offender-pop-grps',
unit='individual', value=f0g_free_0)
quantities.append(f0g_free)
# Model parameters: commitment coefficients and their modifiers
# Community non-offerders to offenders
c00g_0 = np.random.random(self.n_groups) / (5*const.year)
c00g = Quantity(name='c00g', formalName='commit-arrested-coeff-grps',
unit='individual', value=c00g_0)
self.ode_params['general-commit-to-arrested-coeff-grps'] = c00g_0
quantities.append(c00g)
m00g_0 = np.random.random(self.n_groups)
m00g = Quantity(name='m00g', formalName='commit-arrested-coeff-mod-grps',
unit='individual', value=m00g_0)
self.ode_params['general-commit-to-arrested-coeff-mod-grps'] = m00g_0
quantities.append(m00g)
# Community offenders to arrested (recidivism)
c0rg_0 = np.random.random(self.n_groups) / (180*const.day)
c0rg = Quantity(name='c0rg', formalName='commit-arrested-coeff-grps',
unit='individual', value=c0rg_0)
self.ode_params['commit-to-arrested-coeff-grps'] = c0rg_0
quantities.append(c0rg)
m0rg_0 = np.random.random(self.n_groups)
m0rg = Quantity(name='m0rg', formalName='commit-arrested-coeff-mod-grps',
unit='individual', value=m0rg_0)
self.ode_params['commit-to-arrested-coeff-mod-grps'] = m0rg_0
quantities.append(m0rg)
# Non-offender adult population
self.ode_params['non-offender-adult-population'] = np.ones(self.n_groups) * \
non_offender_adult_population
# Death term
self.ode_params['death-rates-coeff'] = 1.0 * np.random.random(self.n_groups) / \
const.year
# Phase state
self.population_phase = Phase(self.initial_time, time_unit='s',
quantities=quantities)
self.population_phase.SetValue('f0g', f0g_0, self.initial_time)
# Initialize inflows to zero
self.ode_params['prison-inflow-rates'] = np.zeros(self.n_groups)
self.ode_params['parole-inflow-rates'] = np.zeros(self.n_groups)
self.ode_params['arrested-inflow-rates'] = np.zeros(self.n_groups)
self.ode_params['jail-inflow-rates'] = np.zeros(self.n_groups)
self.ode_params['adjudication-inflow-rates'] = np.zeros(self.n_groups)
self.ode_params['probation-inflow-rates'] = np.zeros(self.n_groups)
return
[docs] def run(self, *args):
self.__zero_ode_parameters()
time = self.initial_time
while time < self.end_time:
if self.show_time[0] and abs(time%self.show_time[1]-0.0)<=1.e-1:
self.log.info('Community::time[d] = '+str(round(time/const.day,1)))
self.__call_ports(time)
# Evolve offenders group population to the next time stamp
#---------------------------------------------------------
time = self.__step( time )
def __call_ports(self, time):
# Interactions in the jail port
#--------------------------------
# one way "from" jail
self.send( time, 'jail' )
(check_time, inflow_rates) = self.recv('jail')
assert abs(check_time-time) <= 1e-6
self.ode_params['jail-inflow-rates'] = inflow_rates
# Interactions in the adjudication port
#--------------------------------------
# one way "from" adjudication
self.send( time, 'adjudication' )
(check_time, inflow_rates) = self.recv('adjudication')
assert abs(check_time-time) <= 1e-6
self.ode_params['adjudication-inflow-rates'] = inflow_rates
# Interactions in the probation port
#--------------------------------
# one way "from" probation
self.send( time, 'probation' )
(check_time, inflow_rates) = self.recv('probation')
assert abs(check_time-time) <= 1e-6
self.ode_params['probation-inflow-rates'] = inflow_rates
# Interactions in the prison port
#--------------------------------
# one way "from" prison
self.send( time, 'prison' )
(check_time, inflow_rates) = self.recv('prison')
assert abs(check_time-time) <= 1e-6
self.ode_params['prison-inflow-rates'] = inflow_rates
# Interactions in the parole port
#--------------------------------
# one way "from" parole
self.send( time, 'parole' )
(check_time, inflow_rates) = self.recv('parole')
assert abs(check_time-time) <= 1e-6
self.ode_params['parole-inflow-rates'] = inflow_rates
# Interactions in the arrested port
#--------------------------------
# two way "to" and "from" arrested
# to
message_time = self.recv('arrested')
outflow_rates = self.__compute_outflow_rates( message_time, 'arrested' )
self.send( (message_time, outflow_rates), 'arrested' )
# from
self.send( time, 'arrested' )
(check_time, inflow_rates) = self.recv('arrested')
assert abs(check_time-time) <= 1e-6
self.ode_params['arrested-inflow-rates'] = inflow_rates
def __step(self, time=0.0):
r'''
ODE IVP problem:
Given the initial data at :math:`t=0`,
:math:`u = (u_1(0),u_2(0),\ldots)`
solve :math:`\frac{\text{d}u}{\text{d}t} = f(u)` in the interval
:math:`0\le t \le t_f`.
Parameters
----------
time: float
Time in SI unit.
Returns
-------
None
'''
# Get state values
u_0 = self.population_phase.GetValue('f0g', time)
t_interval_sec = np.linspace(0.0, self.time_step, num=2)
(u_vec_hist, info_dict) = odeint( self.__rhs_fn,
u_0, t_interval_sec,
args=( self.ode_params, ),
rtol=1e-4, atol=1e-8, mxstep=200,
full_output=True )
assert info_dict['message'] =='Integration successful.', info_dict['message']
u_vec = u_vec_hist[1,:] # solution vector at final time step
time += self.time_step
# Update state variables
values = self.population_phase.GetRow() # values existing values
self.population_phase.AddRow(time, values) # copy on new time for convenience
self.population_phase.SetValue('f0g', u_vec, time) # insert new values
# Update the population of free offenders returning to community
inflow_rates = self.ode_params['total-inflow-rates']
f0g_free = inflow_rates * self.time_step
self.population_phase.SetValue('f0g_free',f0g_free,time)
return time
def __rhs_fn(self, u_vec, t, params):
f0g = u_vec # offender population groups (removed from community)
prison_inflow_rates = params['prison-inflow-rates']
parole_inflow_rates = params['parole-inflow-rates']
arrested_inflow_rates = params['arrested-inflow-rates']
jail_inflow_rates = params['jail-inflow-rates']
adjudication_inflow_rates = params['adjudication-inflow-rates']
probation_inflow_rates = params['probation-inflow-rates']
inflow_rates = prison_inflow_rates + parole_inflow_rates +\
arrested_inflow_rates + jail_inflow_rates +\
adjudication_inflow_rates + probation_inflow_rates
params['total-inflow-rates'] = inflow_rates
assert np.all(inflow_rates>=0.0), 'values: %r'%inflow_rates
c0rg = params['commit-to-arrested-coeff-grps']
m0rg = params['commit-to-arrested-coeff-mod-grps']
c00g = params['general-commit-to-arrested-coeff-grps']
m00g = params['general-commit-to-arrested-coeff-mod-grps']
non_offender_adult_population = params['non-offender-adult-population']
# Recidivism + new offenders
outflow_rates = c0rg * m0rg * np.abs(f0g) + \
c00g * m00g * non_offender_adult_population
assert np.all(outflow_rates>=0.0), 'values: %r'%outflow_rates
death_rates = params['death-rates-coeff'] * np.abs(f0g)
assert np.all(death_rates>=0.0), 'values: %r'%death_rates
dt_f0g = inflow_rates - outflow_rates - death_rates
return dt_f0g
def __compute_outflow_rates(self, time, name):
f0g = self.population_phase.GetValue('f0g',time)
if name == 'arrested':
c0rg = self.ode_params['commit-to-arrested-coeff-grps']
m0rg = self.ode_params['commit-to-arrested-coeff-mod-grps']
c00g = self.ode_params['general-commit-to-arrested-coeff-grps']
m00g = self.ode_params['general-commit-to-arrested-coeff-mod-grps']
f0 = self.ode_params['non-offender-adult-population']
# Recidivism
outflow_rates = c0rg * m0rg * np.abs(f0g) + c00g * m00g * f0
return outflow_rates
def __zero_ode_parameters(self):
'''
If ports are not connected the corresponding outflows must be zero.
'''
zeros = np.zeros(self.n_groups)
p_names = [p.name for p in self.ports]
if 'arrested' not in p_names:
self.ode_params['commit-to-arrested-coeff-grps'] = zeros
self.ode_params['commit-to-arrested-coeff-mod-grps'] = zeros
self.ode_params['general-commit-to-arrested-coeff-grps'] = zeros
self.ode_params['general-commit-to-arrested-coeff-mod-grps'] = zeros
return
