Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
parameter.
:param stream source: The file-like object containing the text.
:param bool strict: If True, require strict tab delimiting (default). If
False, a relaxed whitespace splitting algorithm is used.
:param string encoding: Encoding used for text representation.
:param bool base64_metadata: If True, metadata is encoded using Base64
encoding; otherwise, as plain text.
:param MutationTable table: If specified, write mutations into this table.
If not, create a new :class:`.MutationTable` instance.
"""
sep = None
if strict:
sep = "\t"
if table is None:
table = tables.MutationTable()
header = source.readline().strip("\n").split(sep)
site_index = header.index("site")
node_index = header.index("node")
derived_state_index = header.index("derived_state")
parent_index = None
parent = NULL
try:
parent_index = header.index("parent")
except ValueError:
pass
metadata_index = None
try:
metadata_index = header.index("metadata")
except ValueError:
pass
for line in source:
parameter.
:param stream source: The file-like object containing the text.
:param bool strict: If True, require strict tab delimiting (default). If
False, a relaxed whitespace splitting algorithm is used.
:param string encoding: Encoding used for text representation.
:param bool base64_metadata: If True, metadata is encoded using Base64
encoding; otherwise, as plain text.
:param MutationTable table: If specified, write mutations into this table.
If not, create a new :class:`.MutationTable` instance.
"""
sep = None
if strict:
sep = "\t"
if table is None:
table = tables.MutationTable()
header = source.readline().strip("\n").split(sep)
site_index = header.index("site")
node_index = header.index("node")
derived_state_index = header.index("derived_state")
parent_index = None
parent = NULL
try:
parent_index = header.index("parent")
except ValueError:
pass
metadata_index = None
try:
metadata_index = header.index("metadata")
except ValueError:
pass
for line in source:
def _mutation_table_pickle(table):
state = {
"site": table.site,
"node": table.node,
"parent": table.parent,
"derived_state": table.derived_state,
"derived_state_offset": table.derived_state_offset,
"metadata": table.metadata,
"metadata_offset": table.metadata_offset,
}
return MutationTable, tuple(), state
individuals=_tskit.IndividualTable(),
nodes=_tskit.NodeTable(),
edges=_tskit.EdgeTable(),
migrations=_tskit.MigrationTable(),
sites=_tskit.SiteTable(),
mutations=_tskit.MutationTable(),
populations=_tskit.PopulationTable(),
provenances=_tskit.ProvenanceTable(),
sequence_length=sequence_length)
self.ll_tables = ll_tables
self.__individuals = IndividualTable(ll_table=self.ll_tables.individuals)
self.__nodes = NodeTable(ll_table=self.ll_tables.nodes)
self.__edges = EdgeTable(ll_table=self.ll_tables.edges)
self.__migrations = MigrationTable(ll_table=self.ll_tables.migrations)
self.__sites = SiteTable(ll_table=self.ll_tables.sites)
self.__mutations = MutationTable(ll_table=self.ll_tables.mutations)
self.__populations = PopulationTable(ll_table=self.ll_tables.populations)
self.__provenances = ProvenanceTable(ll_table=self.ll_tables.provenances)
def __init__(self, max_rows_increment=0, ll_table=None):
if ll_table is None:
ll_table = _tskit.MutationTable(max_rows_increment=max_rows_increment)
super(MutationTable, self).__init__(ll_table, MutationTableRow)
:param MigrationTable migrations: The MigrationTable to be simplified.
:param SiteTable sites: The SiteTable to be simplified.
:param MutationTable mutations: The MutationTable to be simplified.
:param bool filter_zero_mutation_sites: Whether to remove sites that have no
mutations from the output (default: True).
:param float sequence_length: The length of the sequence.
:return: A numpy array mapping node IDs in the input tables to their
corresponding node IDs in the output tables.
:rtype: numpy array (dtype=np.int32).
"""
if migrations is None:
migrations = MigrationTable()
if sites is None:
sites = SiteTable()
if mutations is None:
mutations = MutationTable()
if sequence_length == 0 and len(edges) > 0:
sequence_length = edges.right.max()
# To make this work with the old semantics we need to create a populations
max_pop = np.max(nodes.population)
populations = _tskit.PopulationTable()
if len(nodes) > 0:
for _ in range(max_pop + 1):
populations.add_row()
max_ind = np.max(nodes.individual)
if max_ind != tskit.NULL_INDIVIDUAL:
raise ValueError("Individuals not supported in this deprecated function")
try:
ll_tables = _tskit.TableCollection(
individuals=_tskit.IndividualTable(),
nodes=nodes.ll_table,
edges=edges.ll_table,
required if ``sites`` is provided).
:param ProvenanceTable provenances: Ignored. This argument is provided to
support calling the function like ``sort_tables(**tables.asdict())``.
:param PopulationTable populations: Ignored. This argument is provided to
support calling the function like ``sort_tables(**tables.asdict())``.
:param IndividualTable individuals: Ignored. This argument is provided to
support calling the function like ``sort_tables(**tables.asdict())``.
:param int edge_start: The index in the edge table where sorting starts
(default=0; must be <= len(edges)).
"""
if migrations is None:
migrations = MigrationTable()
if sites is None:
sites = SiteTable()
if mutations is None:
mutations = MutationTable()
sequence_length = 1
if len(edges) > 0:
sequence_length = edges.right.max()
# To make this work with the old semantics we need to create a populations
populations = _tskit.PopulationTable()
if len(nodes) > 0:
max_pop = np.max(nodes.population)
for _ in range(max_pop + 1):
populations.add_row()
max_ind = np.max(nodes.individual)
if max_ind != tskit.NULL_INDIVIDUAL:
raise ValueError("Individuals not supported in this deprecated function")
try:
ll_tables = _tskit.TableCollection(
individuals=_tskit.IndividualTable(),
nodes=nodes.ll_table,
"mutations": tables.mutations,
"provenances": tables.provenances,
}
return TableCollection, tuple(), state
# Pickle support for the various tables. We are forced to use copyreg.pickle
# here to support Python 2. For Python 3, we can just use the __setstate__.
# It would be cleaner to attach the pickle_*_table functions to the classes
# themselves, but this causes issues with Mocking on readthedocs. Sigh.
copyreg.pickle(IndividualTable, _pickle_individual_table)
copyreg.pickle(NodeTable, _pickle_node_table)
copyreg.pickle(EdgeTable, _edge_table_pickle)
copyreg.pickle(MigrationTable, _migration_table_pickle)
copyreg.pickle(SiteTable, _site_table_pickle)
copyreg.pickle(MutationTable, _mutation_table_pickle)
copyreg.pickle(PopulationTable, _population_table_pickle)
copyreg.pickle(ProvenanceTable, _provenance_table_pickle)
copyreg.pickle(TableCollection, _table_collection_pickle)
#############################################
# Table functions.
#############################################
def sort_tables(
nodes, edges, migrations=None, sites=None, mutations=None,
provenances=None, individuals=None, populations=None, edge_start=0):
"""
**This function is deprecated. Please use TableCollection.sort() instead**