Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_arg(self):
rng = _msprime.RandomGenerator(1)
tables = _msprime.LightweightTableCollection()
sim = _msprime.Simulator(
get_samples(5), uniform_recombination_map(num_loci=20, rate=2),
rng, tables, store_full_arg=True)
sim.run()
mutgen = _msprime.MutationGenerator(rng, 2)
mutgen.generate(tables)
t = tskit.TableCollection.fromdict(tables.asdict())
self.assertGreater(len(t.edges), 10)
self.assertGreater(len(t.mutations), 10)
return tables
def verify_block_size(self, tables):
rng = _msprime.RandomGenerator(1)
mutgen = _msprime.MutationGenerator(rng, 2)
ll_tables = _msprime.LightweightTableCollection()
ll_tables.fromdict(tables.asdict())
mutgen.generate(ll_tables, keep=True)
after_tables = tskit.TableCollection.fromdict(ll_tables.asdict())
self.assertEqual(tables, after_tables)
def verify(self, tables):
lwt = c_module.LightweightTableCollection()
lwt.fromdict(tables.asdict())
other_tables = tskit.TableCollection.fromdict(lwt.asdict())
self.assertEqual(tables, other_tables)
def test_migrations(self):
sim, ll_tables = get_example_simulator(
10, num_populations=3, store_migrations=True)
sim.run()
tables = tskit.TableCollection.fromdict(ll_tables.asdict())
self.assertGreater(sim.get_num_migrations(), 0)
self.assertEqual(sim.get_num_migrations(), len(tables.migrations))
sim_records = sim.get_migrations()
self.assertEqual(len(sim_records), len(tables.migrations))
for sim_r, ts_r in zip(sim_records, tables.migrations):
self.assertEqual(len(sim_r), 6)
# Left and right have been remapped, so might be slightly different.
self.assertAlmostEqual(sim_r[0], ts_r.left)
self.assertAlmostEqual(sim_r[1], ts_r.right)
self.assertAlmostEqual(sim_r[2], ts_r.node)
self.assertAlmostEqual(sim_r[3], ts_r.source)
self.assertAlmostEqual(sim_r[4], ts_r.dest)
self.assertAlmostEqual(sim_r[5], ts_r.time)
def get_table_collection_copy(tables, sequence_length):
"""
Returns a copy of the specified table collection with the specified
sequence length.
"""
table_dict = tables.asdict()
table_dict["sequence_length"] = sequence_length
return tskit.TableCollection.fromdict(table_dict)
def get_tree_sequence(self, mutation_generator=None, provenance_record=None):
"""
Returns a TreeSequence representing the state of the simulation.
"""
if mutation_generator is not None:
mutation_generator.generate(self.ll_tables)
tables = tskit.TableCollection.fromdict(self.ll_tables.asdict())
if provenance_record is not None:
tables.provenances.add_row(provenance_record)
if self.from_ts is None:
# Add the populations with metadata
assert len(tables.populations) == len(self.population_configurations)
tables.populations.clear()
for pop_config in self.population_configurations:
tables.populations.add_row(metadata=pop_config.encoded_metadata)
return tables.tree_sequence()
else:
end_time = float(end_time)
parameters["end_time"] = end_time
# TODO Add a JSON representation of the model to the provenance.
provenance_dict = provenance.get_provenance_dict(parameters)
if start_time > end_time:
raise ValueError("start_time must be <= end_time")
mutation_generator = _msprime.MutationGenerator(
rng, rate, alphabet=alphabet, start_time=start_time, end_time=end_time)
lwt = _msprime.LightweightTableCollection()
lwt.fromdict(tables.asdict())
mutation_generator.generate(lwt, keep=keep)
tables = tskit.TableCollection.fromdict(lwt.asdict())
tables.provenances.add_row(json.dumps(provenance_dict))
return tables.tree_sequence()