Skip to content

steinbock.export

data

try_convert_to_anndata_from_disk(x_files, *obs_file_lists)

Source code in steinbock/export/data.py 
def try_convert_to_anndata_from_disk(
    x_files: Sequence[Union[str, PathLike]], *obs_file_lists
) -> Generator[Tuple[Path, Tuple[Path, ...], AnnData], None, None]:
    for x_file, *obs_files in zip(x_files, *obs_file_lists):
        obs_files = tuple(Path(obs_file) for obs_file in obs_files)
        try:
            x = io.read_data(x_file)
            obs = None
            if len(obs_files) > 0:
                obs = io.read_data(obs_files[0])
                for obs_file in obs_files[1:]:
                    obs = pd.merge(
                        obs,
                        io.read_data(obs_file),
                        left_index=True,
                        right_index=True,
                    )
                obs.index = obs.index.astype(str)
            anndata = AnnData(X=x.values, obs=obs)
            yield Path(x_file), obs_files, anndata
            del x, obs
        except:
            _logger.exception(f"Error converting {x_file} to AnnData")

try_convert_to_dataframe_from_disk(*data_file_lists)

Source code in steinbock/export/data.py 
def try_convert_to_dataframe_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

convert_to_networkx(neighbors, *data_list)

Source code in steinbock/export/graphs.py 
def convert_to_networkx(neighbors: pd.DataFrame, *data_list) -> nx.Graph:
    edges = neighbors[["Object", "Neighbor"]].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()])
    graph: nx.Graph = nx.from_pandas_edgelist(
        neighbors,
        source="Object",
        target="Neighbor",
        edge_attr=True,
        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(graph, node_attributes)
    return graph

try_convert_to_networkx_from_disk(neighbors_files, *data_file_lists)

Source code in steinbock/export/graphs.py 
def try_convert_to_networkx_from_disk(
    neighbors_files: Sequence[Union[str, PathLike]], *data_file_lists
) -> Generator[Tuple[Path, Tuple[Path, ...], nx.Graph], None, None]:
    for neighbors_file, *data_files in zip(neighbors_files, *data_file_lists):
        data_files = tuple(Path(data_file) for data_file in data_files)
        try:
            neighbors = io.read_neighbors(neighbors_file)
            data_list = [io.read_data(data_file) for data_file in data_files]
            graph = convert_to_networkx(neighbors, *data_list)
            yield Path(neighbors_file), data_files, graph
            del neighbors, data_list, graph
        except:
            _logger.exception(f"Error converting {neighbors_file} to networkx")
Back to top