Skip to content

steinbock.export

data

to_anndata_from_disk(x_files, *obs_file_lists)

Source code in steinbock/export/data.py 
def to_anndata_from_disk(
    x_files: Sequence[Union[str, PathLike]], *obs_file_lists
) -> Generator[Tuple[Path, AnnData], None, None]:
    for i, x_file in enumerate(x_files):
        x = io.read_data(x_file)
        merged_obs = None
        if len(obs_file_lists) > 0:
            merged_obs = io.read_data(obs_file_lists[0][i])
            for obs_files in obs_file_lists[1:]:
                merged_obs = pd.merge(
                    merged_obs,
                    io.read_data(obs_files[i]),
                    left_index=True,
                    right_index=True,
                )
            merged_obs.index = merged_obs.index.astype(str)
        yield Path(x_file), AnnData(X=x.values, obs=merged_obs)
        del x, merged_obs

to_table_from_disk(*data_file_lists)

Source code in steinbock/export/data.py 
def to_table_from_disk(*data_file_lists) -> pd.DataFrame:
    data_objs = []
    img_file_names = []
    for data_files in zip(*data_file_lists):
        data = io.read_data(data_files[0])
        for data_file in data_files[1:]:
            data = pd.merge(
                data,
                io.read_data(data_file),
                left_index=True,
                right_index=True,
            )
        img_file = io.as_path_with_suffix(data_files[0], ".tiff")
        data_objs.append(data)
        img_file_names.append(img_file.name)
    return pd.concat(data_objs, keys=img_file_names, names=["Image", "Object"])

graphs

to_networkx(graph, *data_list)

Source code in steinbock/export/graphs.py 
def to_networkx(graph: pd.DataFrame, *data_list) -> nx.Graph:
    edges = graph[["Object1", "Object2"]].astype(int).values.tolist()
    undirected_edges = [tuple(sorted(edge)) for edge in edges]
    is_directed = any([x != 2 for x in Counter(undirected_edges).values()])
    g: nx.Graph = nx.from_pandas_edgelist(
        graph,
        source="Object1",
        target="Object2",
        create_using=nx.DiGraph if is_directed else nx.Graph,
    )
    if len(data_list) > 0:
        merged_data = data_list[0]
        for data in data_list[1:]:
            merged_data = pd.merge(
                merged_data, data, left_index=True, right_index=True
            )
        node_attributes = {
            int(object_id): object_data.to_dict()
            for object_id, object_data in merged_data.iterrows()
        }
        nx.set_node_attributes(g, node_attributes)
    return g

to_networkx_from_disk(graph_files, *data_file_lists)

Source code in steinbock/export/graphs.py 
def to_networkx_from_disk(
    graph_files: Sequence[Union[str, PathLike]], *data_file_lists
) -> Generator[Tuple[Path, nx.Graph], None, None]:
    for i, graph_file in enumerate(graph_files):
        graph = io.read_graph(graph_file)
        data_list = []
        for data_files in data_file_lists:
            data_list.append(io.read_data(data_files[i]))
        g = to_networkx(graph, *data_list)
        yield Path(graph_file), g
        del graph, data_list, g
Back to top