Shared utilities for AtlasXomics Latch workflows.
In a workflow Dockerfile (uses a GitHub tarball URL — no git binary required):
# pin to a release tag so workflows stay stable
RUN pip install https://github.com/atlasxomics/atx-common/archive/refs/tags/v0.1.0.tar.gzFor local development:
pip install -e /path/to/atx-common
# or with anndata helpers:
pip install -e "/path/to/atx-common[analysis]"Genomeenum (hg38, mm10, mm39, rnor6)HasSpatialDir/HasRunInfoprotocols — let your workflow define its ownRundataclass while still using shared functions
- Chromosome sizes:
mm10_chrsizes,hg38_chrsizes,rnor6_chrsizes,chrsizes(unified dict) ref_dict— genome annotation GFF3 pathspt_sizes— DBiT channel-to-point-size mappinggene_keys— mitochondrial/ribosomal gene prefixes per genomeget_genome_fasta(genome)— S3 FASTA pathsget_blacklist_path(genome)— resolve blacklist BED files
log(msg, title, type)— combined Python logging + Latch UI messageget_channels(run, default)— read numChannels from spatial metadataget_LatchFile(directory, file_name, retries, retry_delay_s)— retry-aware file lookup
sanitize_condition(condition)— normalize condition labelsget_groups(runs)— determine grouping columns for differential analysisfilter_anndata(adata, group, subgroup)— subset AnnData by obs columnrename_obs_columns(adata)— standardize ArchR column namescreate_output_directories(project_name)— create figures/tables dirsorganize_outputs(project_name, dirs)— move outputs into standard layoutcopy_peak_files(project_name, dirs)— copy ArchR peak call filesmove_files_to_directory(patterns, target_dir)— glob-and-move helper
from dataclasses import dataclass
from latch.types import LatchFile, LatchDir
from atx_common import (
Genome,
get_channels,
get_groups,
get_LatchFile,
log,
pt_sizes,
sanitize_condition,
)
@dataclass
class Run:
run_id: str
fragments_file: LatchFile
spatial_dir: LatchDir
condition: str = "None"
# All shared functions work with your custom Run because it satisfies
# the HasSpatialDir and HasRunInfo protocols.
channels = get_channels(run)
groups = get_groups(runs)
size = pt_sizes[channels]Tag releases on GitHub (v0.1.0, v0.2.0, ...) and pin to them in Dockerfiles.
Bump the tag in a workflow's Dockerfile only when you want it to pick up changes.