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_mass_migration_dest(self):
self.assertRaises(
ValueError, msprime.MassMigration, time=0, source=0, dest=0, destination=0)
for j in range(10):
e = msprime.MassMigration(time=0.1, source=0, dest=j, proportion=0.5)
self.assertEqual(e.time, 0.1)
self.assertEqual(e.source, 0)
self.assertEqual(e.dest, j)
self.assertEqual(e.proportion, 0.5)
e = msprime.MassMigration(0.1, 0, j, 0.5)
self.assertEqual(e.time, 0.1)
self.assertEqual(e.source, 0)
self.assertEqual(e.dest, j)
self.assertEqual(e.proportion, 0.5)
e = msprime.MassMigration(time=0.1, source=0, destination=j, proportion=0.5)
self.assertEqual(e.time, 0.1)
self.assertEqual(e.source, 0)
self.assertEqual(e.dest, j)
self.assertEqual(e.proportion, 0.5)
def test_mass_migration_source_dest(self):
g = 51
for source, dest in itertools.permutations(range(4), 2):
event = msprime.MassMigration(time=g, source=source, dest=dest)
d = event.get_ll_representation(1)
dp = {
"time": g,
"type": "mass_migration",
"source": source,
"dest": dest,
"proportion": 1}
self.assertEqual(d, dp)
def test_mass_migration_time(self):
for g in [0.1, 1, 100, 1e6]:
event = msprime.MassMigration(time=g, source=0, dest=1)
d = event.get_ll_representation(1)
dp = {
"time": g,
"type": "mass_migration",
"source": 0,
"dest": 1,
"proportion": 1}
self.assertEqual(d, dp)
# 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_defaults(self):
events = [
msprime.PopulationParametersChange(0, initial_size=1),
msprime.MigrationRateChange(0, 1),
msprime.MassMigration(0, 0),
msprime.SimulationModelChange(0, msprime.StandardCoalescent(1)),
msprime.SimpleBottleneck(0),
msprime.InstantaneousBottleneck(0)]
for event in events:
self.assertGreater(len(str(event)), 0)
# 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=[
[0, 0],
[0, 0]
])
steps = np.linspace(0, 400, 401)
rates, P = ddb.coalescence_rate_trajectory(steps=steps, num_samples=[2, 0])
pop_sizes = ddb.population_size_trajectory(steps=steps)
for r, p in zip(rates[301:], pop_sizes[:, 0][301:]):
self.assertAlmostEqual(1/(2*r), p)
check_event_time(parser, t)
demographic_events.append(
(index, msprime.PopulationParametersChange(
time=t, initial_size=x, growth_rate=0)))
for index, (t, population_id, x) in args.population_size_change:
check_event_time(parser, t)
pid = convert_population_id(parser, population_id, num_populations)
demographic_events.append(
(index, msprime.PopulationParametersChange(
time=t, initial_size=x, growth_rate=0, population_id=pid)))
for index, (t, source, dest) in args.population_split:
check_event_time(parser, t)
source_id = convert_population_id(parser, source, num_populations)
dest_id = convert_population_id(parser, dest, num_populations)
demographic_events.append(
(index, msprime.MassMigration(t, source_id, dest_id, 1.0)))
# Set the migration rates for source to 0
for j in range(num_populations):
if j != source_id:
event = msprime.MigrationRateChange(t, 0.0, (j, source_id))
demographic_events.append((index, event))
# Demographic events that affect the migration matrix
if num_populations == 1:
condition = (
len(args.migration_rate_change) > 0 or
len(args.migration_matrix_entry_change) > 0 or
len(args.migration_matrix_change) > 0)
if condition:
parser.error("Cannot change migration rates for 1 population")
for index, (t, x) in args.migration_rate_change:
if len(args.admixture) != 0:
# print("new")
for t in ts_new.trees():
pass
# print(t)
# def migration_records_example():
population_configurations = [
msprime.PopulationConfiguration(1),
msprime.PopulationConfiguration(1),
msprime.PopulationConfiguration(0),
]
demographic_events = [
msprime.MassMigration(time=5, source=0, destination=2),
msprime.MassMigration(time=6, source=1, destination=2),
]
ts = msprime.simulate(
Ne=0.25, population_configurations=population_configurations,
demographic_events=demographic_events,
random_seed=1, length=10, recombination_rate=0.5,
record_migrations=True)
for mr in ts.migrations():
print(mr)
for population_id, size in args.population_size:
pid = convert_population_id(parser, population_id, num_populations)
population_configurations[pid].initial_size = size
# First we look at population split events. We do this differently
# to ms, as msprime requires a fixed number of population. Therefore,
# modify the number of populations to take into account populations
# splits. This is a messy hack, and will probably need to be changed.
for index, (t, population_id, proportion) in args.admixture:
check_event_time(parser, t)
pid = convert_population_id(parser, population_id, num_populations)
if proportion < 0 or proportion > 1:
parser.error("Proportion value must be 0 <= p <= 1.")
# In ms, the probability of staying in source is p and the probabilty
# of moving to the new population is 1 - p.
event = (index, msprime.MassMigration(
t, pid, num_populations, 1 - proportion))
demographic_events.append(event)
num_populations += 1
# We add another element to each row in the migration matrix
# along with an other row. All new entries are zero.
for row in migration_matrix:
row.append(0)
migration_matrix.append([0 for j in range(num_populations)])
# Add another PopulationConfiguration object with a sample size
# of zero.
population_configurations.append(msprime.PopulationConfiguration(0))
# Add the demographic events
for index, (t, alpha) in args.growth_rate_change:
if len(args.admixture) != 0: