import numpy as np
import cellxgene_census
import tiledbsoma as soma
def main():
census_version = "2023-07-25"
with cellxgene_census.open_soma(census_version=census_version) as census:
summary = census["census_info"]["summary"].read().concat().to_pandas()
total_cells = int(summary["total_cell_count"].iloc[0])
print(f"Census version: {census_version}")
print(f"Total cells: {total_cells:,}")
obs = cellxgene_census.get_obs(
census,
"homo_sapiens",
value_filter="tissue_general == 'brain' and is_primary_data == True",
column_names=["cell_type", "tissue_general", "disease", "donor_id"],
)
print(f"Brain (primary) cells returned (metadata only): {len(obs):,}")
print("Top cell types:")
print(obs["cell_type"].value_counts().head(10))
adata = cellxgene_census.get_anndata(
census=census,
organism="Homo sapiens",
obs_value_filter=(
"cell_type == 'T cell' and disease == 'COVID-19' and is_primary_data == True"
),
var_value_filter="feature_name in ['CD4', 'CD8A', 'FOXP3']",
obs_column_names=["cell_type", "tissue_general", "disease", "donor_id", "sex"],
)
print(adata)
print("AnnData X shape:", adata.X.shape)
query = census["census_data"]["homo_sapiens"].axis_query(
measurement_name="RNA",
obs_query=soma.AxisQuery(
value_filter="tissue_general == 'brain' and is_primary_data == True"
),
var_query=soma.AxisQuery(
value_filter="feature_name in ['FOXP2', 'TBR1', 'SATB2']"
),
)
n = 0
s = 0.0
for batch in query.X("raw").tables():
values = batch["soma_data"].to_numpy(zero_copy_only=False)
n += values.size
s += float(values.sum())
mean_expr = s / n if n else np.nan
print(f"Out-of-core mean expression (over returned entries): {mean_expr:.6g}")
if __name__ == "__main__":
main()
