Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def validate_encoding(self, params):
pdict = provenance.get_provenance_dict("test", **params)
encoded = pdict["parameters"]
self.assertEqual(encoded["command"], "test")
del encoded["command"]
self.assertEqual(encoded, params)
def test_no_command(self):
with self.assertRaises(ValueError):
provenance.get_provenance_dict()
flags = np.zeros_like(tables.nodes.flags)
index = tables.nodes.flags == tskit.NODE_IS_SAMPLE
flags[index] = tskit.NODE_IS_SAMPLE
index = tables.nodes.flags != tskit.NODE_IS_SAMPLE
flags[index] = np.bitwise_and(tables.nodes.flags[index], ~tskit.NODE_IS_SAMPLE)
tables.nodes.set_columns(
flags=flags,
time=tables.nodes.time,
population=tables.nodes.population,
individual=tables.nodes.individual,
metadata=tables.nodes.metadata,
metadata_offset=tables.nodes.metadata_offset,
)
record = provenance.get_provenance_dict(command="extract_ancestors")
tables.provenances.add_row(record=json.dumps(record))
return tables, node_id_map
parent=parent,
child=child,
)
self.convert_inference_mutations(tables)
logger.debug("Sorting ancestors tree sequence")
tables.sort()
# Note: it's probably possible to compute the mutation parents from the
# tsb data structures but we're not doing it for now.
tables.build_index()
tables.compute_mutation_parents()
logger.debug("Sorting ancestors tree sequence done")
for timestamp, record in self.ancestor_data.provenances():
tables.provenances.add_row(timestamp=timestamp, record=json.dumps(record))
record = provenance.get_provenance_dict(
command="match_ancestors", source={"uuid": self.ancestor_data.uuid}
)
tables.provenances.add_row(record=json.dumps(record))
logger.info(
"Built ancestors tree sequence: {} nodes ({} pc ancestors); {} edges; "
"{} sites; {} mutations".format(
len(tables.nodes),
np.sum(pc_ancestors),
len(tables.edges),
len(tables.mutations),
len(tables.sites),
)
)
return tables.tree_sequence()
right=np.full(num_intersecting, left, dtype=np.float64),
parent=i_parent[index],
child=i_child[index],
)
# Only insert valid edges (remove any entirely lost topology)
index = right < i_right
num_intersecting = np.sum(index)
tables.edges.append_columns(
left=np.full(num_intersecting, right, dtype=np.float64),
right=i_right[index],
parent=i_parent[index],
child=i_child[index],
)
tables.sort()
record = provenance.get_provenance_dict(
command="snip_centromere", left=left, right=right
)
tables.provenances.add_row(record=json.dumps(record))
return tables.tree_sequence()
def record_provenance(self, command=None, **kwargs):
"""
Records the provenance information for this file using the
tskit provenances schema.
"""
timestamp = datetime.datetime.now().isoformat()
record = provenance.get_provenance_dict(command=command, **kwargs)
self.add_provenance(timestamp, record)
tables.nodes.time = tables.nodes.time + 1
# TODO - check this works for augmented ancestors with missing data
left, right, parent, child = tsb.dump_edges()
tables.edges.set_columns(
left=self.position_map[left],
right=self.position_map[right],
parent=parent,
child=child,
)
tables.sites.clear()
tables.mutations.clear()
self.convert_inference_mutations(tables)
record = provenance.get_provenance_dict(command="augment_ancestors")
tables.provenances.add_row(record=json.dumps(record))
logger.debug("Sorting ancestors tree sequence")
tables.sort()
logger.debug("Sorting ancestors tree sequence done")
logger.info(
"Augmented ancestors tree sequence: {} nodes ({} extra pc ancestors); "
"{} edges; {} sites; {} mutations".format(
len(tables.nodes),
num_pc_ancestors,
len(tables.edges),
len(tables.mutations),
len(tables.sites),
)
)
return tables.tree_sequence()