Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_rate_size_equality_with_population_merge(self):
# Test equality between two population that split with continued
# migration
N_A = 1e7
N_B = 1e6
ddb = msprime.DemographyDebugger(
population_configurations=[
msprime.PopulationConfiguration(initial_size=N_A),
msprime.PopulationConfiguration(initial_size=N_B)
],
demographic_events=[
msprime.PopulationParametersChange(time=100, initial_size=1e5,
population_id=0, growth_rate=0),
msprime.PopulationParametersChange(time=100, initial_size=1e5,
population_id=1, growth_rate=7e-4),
msprime.MassMigration(time=100, source=1, dest=0),
msprime.PopulationParametersChange(time=200, initial_size=1e6,
population_id=0, growth_rate=-1e-2),
msprime.PopulationParametersChange(time=300, initial_size=1e6,
population_id=0, growth_rate=0),
],
migration_matrix=[
def verify_debug(
self, population_configurations, migration_matrix, demographic_events):
dd = msprime.DemographyDebugger(
population_configurations=population_configurations,
migration_matrix=migration_matrix,
demographic_events=demographic_events)
# Check the reprs
s = repr(dd.epochs)
self.assertGreater(len(s), 0)
with tempfile.TemporaryFile("w+") as f:
dd.print_history(f)
f.seek(0)
debug_output = f.read()
# TODO when there is better output, write some tests to
# verify its format.
self.assertGreater(len(debug_output), 0)
def two_pop_example(self):
# Have a history with two populations, with no migration;
# check that coalescence rate of lineages from
# a single pop reflects the size history of that pop:
# - population sizes today are [100, 1000]
# - population 1 had a size change to 200 at t=100 generations ago
# - there is a mass migration event that entirely merges population 2 (source)
# into population 1 (dest) at t=200 generations ago
# - population 1 has a size change to 100 at t=300 generations ago
ddb = msprime.DemographyDebugger(
population_configurations=[
msprime.PopulationConfiguration(initial_size=1e2),
msprime.PopulationConfiguration(initial_size=1e3)
],
demographic_events=[
msprime.PopulationParametersChange(time=100, initial_size=200,
population_id=0),
msprime.MassMigration(time=200, source=1, dest=0),
msprime.PopulationParametersChange(time=300, initial_size=100)
],
migration_matrix=[
[0, 0],
[0, 0]
])
return ddb
def test_constant_sizes(self):
# With constant population sizes, results should not depend on the steps
N_A = 1e2
ddb = msprime.DemographyDebugger(
population_configurations=[
msprime.PopulationConfiguration(initial_size=N_A)
],
demographic_events=[
msprime.PopulationParametersChange(time=100, initial_size=200),
msprime.PopulationParametersChange(time=300, initial_size=100)
])
steps = np.linspace(0, 400, 21)
rates, P = ddb.coalescence_rate_trajectory(steps=steps, num_samples=[2])
steps2 = self.subdivide(steps)
rates2, P2 = ddb.coalescence_rate_trajectory(steps=steps2, num_samples=[2])
assert(np.all(steps == steps2[::2]))
self.assertLess(max(np.abs(rates - rates2[::2])), 1e-6)
self.assertLess(max(np.abs(P - P2[::2])), 1e-6)
def test_two_pop_change_growth_rates(self):
alpha = 0.33
N0 = 1000
N1 = 10
t1 = 5
t2 = 10
t3 = 15
for model in ["dtwf", "hudson"]:
dd = msprime.DemographyDebugger(
model=model,
population_configurations=[
msprime.PopulationConfiguration(initial_size=N0, growth_rate=alpha),
msprime.PopulationConfiguration(initial_size=N1, growth_rate=0)],
demographic_events=[
# p1 changes growth rate to alpha
msprime.PopulationParametersChange(
population=1, time=t1, growth_rate=alpha),
# p0 changes growth rate to -alpha
msprime.PopulationParametersChange(
population=0, time=t2, growth_rate=-alpha),
# Both change growth_rate to 0 at t3.
msprime.PopulationParametersChange(time=t3, growth_rate=0)])
self.verify_arrays(dd)
self.assertEqual(len(dd.epochs), 4)
def test_model_change_events(self):
population_configurations = [
msprime.PopulationConfiguration(sample_size=10)]
demographic_events = [
msprime.SimulationModelChange(1, "hudson")]
with self.assertRaises(ValueError):
msprime.DemographyDebugger(
population_configurations=population_configurations,
demographic_events=demographic_events)
def test_value_errors(self):
# test all user input domains which should raise ValuErrors.
ddb = msprime.DemographyDebugger(
population_configurations=[
msprime.PopulationConfiguration(initial_size=1e2),
],
demographic_events=[
],
migration_matrix=[
[0],
])
steps = np.linspace(0, 10, 11)
# Test when num_pops != len(num_samples), we throw error
with self.assertRaises(ValueError):
ddb.coalescence_rate_trajectory(steps=steps, num_samples=[2, 0])
# Test that when steps are not strictly increasing values, we throw error.
with self.assertRaises(ValueError):
ddb.coalescence_rate_trajectory(
steps=np.flip(steps, axis=0), num_samples=[2])
def test_two_pop_update_migration_rate(self):
dd = msprime.DemographyDebugger(
population_configurations=[
msprime.PopulationConfiguration(initial_size=100),
msprime.PopulationConfiguration(initial_size=100)],
demographic_events=[
msprime.MigrationRateChange(time=20, rate=1),
msprime.MigrationRateChange(time=22, rate=1.7, matrix_index=(0, 1))],
migration_matrix=[[0, 0.25], [0, 0]])
self.verify_arrays(dd)
self.assertEqual(len(dd.epochs), 3)
e = dd.epochs[0]
self.assertEqual(e.start_time, 0)
self.assertEqual(e.end_time, 20)
self.assertEqual(e.migration_matrix, [[0, 0.25], [0, 0]])
for pop in e.populations:
self.assertEqual(pop.growth_rate, 0)
self.assertEqual(pop.start_size, pop.end_size)
msprime.MigrationRateChange(
time=T_EU_AS, rate=m_AF_B, matrix_index=(0, 1)),
msprime.MigrationRateChange(
time=T_EU_AS, rate=m_AF_B, matrix_index=(1, 0)),
msprime.PopulationParametersChange(
time=T_EU_AS, initial_size=N_B, growth_rate=0, population_id=1),
# Population B merges into YRI at T_B
msprime.MassMigration(
time=T_B, source=1, destination=0, proportion=1.0),
# Size changes to N_A at T_AF
msprime.PopulationParametersChange(
time=T_AF, initial_size=N_A, population_id=0)
]
# Use the demography debugger to print out the demographic history
# that we have just described.
dp = msprime.DemographyDebugger(
Ne=N_A,
population_configurations=population_configurations,
migration_matrix=migration_matrix,
demographic_events=demographic_events)
dp.print_history()
return(population_configurations, migration_matrix, demographic_events)