Services API Reference

from lobster.core.provenance import AnalysisStep

class ExampleService:
    """Stateless service for biological data analysis."""

    def __init__(self):
        """Initialize the service (no state stored)."""
        pass

    def analyze(
        self,
        adata: anndata.AnnData,
        **kwargs
    ) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]:
        """
        Perform analysis on AnnData object.

        Args:
            adata: Input AnnData object
            **kwargs: Analysis parameters

        Returns:
            Tuple: (processed_adata, statistics, ir) where ir is AnalysisStep for provenance
        """
        # Process data
        processed_adata = self._process_data(adata, **kwargs)

        # Calculate statistics
        statistics = self._calculate_statistics(processed_adata, adata, **kwargs)

        # Create provenance IR
        ir = AnalysisStep(
            name="analyze",
            description="Analysis performed",
            inputs=[{"adata": "input"}],
            outputs=[{"adata": "output"}],
            parameters=kwargs
        )

        return processed_adata, statistics, ir

class PreprocessingService:
    """
    Advanced preprocessing service for single-cell RNA-seq data.

    This stateless service provides methods for ambient RNA correction, quality control filtering,
    normalization, and batch correction/integration following best practices.
    """

def correct_ambient_rna(
    self,
    adata: anndata.AnnData,
    contamination_fraction: float = 0.1,
    empty_droplet_threshold: int = 100,
    method: str = "simple_decontamination"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

def filter_cells_and_genes(
    self,
    adata: anndata.AnnData,
    min_genes_per_cell: int = 200,
    min_cells_per_gene: int = 3,
    max_genes_per_cell: int = None,
    max_pct_mito: float = 20.0,
    max_pct_ribo: float = None
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

def normalize_data(
    self,
    adata: anndata.AnnData,
    target_sum: float = 1e4,
    normalization_method: str = "log1p",
    highly_variable_genes: bool = True,
    n_top_genes: int = 2000
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class QualityService:
    """Service for assessing data quality with comprehensive metrics."""

def assess_quality_comprehensive(
    self,
    adata: anndata.AnnData,
    organism: str = "human",
    include_scrublet: bool = True
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ClusteringService:
    """Stateless service for clustering single-cell RNA-seq data."""

def cluster_and_visualize(
    self,
    adata: anndata.AnnData,
    resolution: Optional[float] = None,
    use_rep: Optional[str] = None,
    batch_correction: bool = False,
    batch_key: Optional[str] = None,
    n_pcs: int = 50,
    n_neighbors: int = 15,
    umap_min_dist: float = 0.5,
    random_state: int = 42
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class EnhancedSinglecellService:
    """Enhanced service for advanced single-cell analysis workflows."""

def detect_doublets_comprehensive(
    self,
    adata: anndata.AnnData,
    expected_doublet_rate: float = 0.1,
    use_scrublet: bool = True,
    use_doubletfinder_alternative: bool = True,
    n_neighbors: int = None,
    n_pcs: int = 30
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

def find_marker_genes(
    self,
    adata: anndata.AnnData,
    groupby: str,
    method: str = "wilcoxon",
    n_genes: int = 100,
    reference: str = "rest",
    min_fold_change: float = 1.5,
    max_pval_adj: float = 0.05
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class BulkRNAseqService:
    """Service for bulk RNA-seq differential expression analysis."""

def run_deseq2_analysis(
    self,
    adata: anndata.AnnData,
    design_formula: str,
    condition_col: str,
    reference_level: str = None,
    batch_col: str = None,
    min_count: int = 10,
    alpha: float = 0.05
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class DifferentialFormulaService:
    """Service for constructing and validating R-style formulas for differential analysis."""

def construct_formula(
    self,
    adata: anndata.AnnData,
    primary_condition: str,
    covariates: List[str] = None,
    interactions: List[Tuple[str, str]] = None,
    formula_type: str = "additive"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class PseudobulkService:
    """Service for converting single-cell data to pseudobulk for differential expression."""

def create_pseudobulk(
    self,
    adata: anndata.AnnData,
    sample_col: str,
    cluster_col: str = None,
    min_cells: int = 10,
    aggregation_method: str = "sum"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ProteomicsPreprocessingService:
    """Service for preprocessing proteomics data including missing value handling."""

def handle_missing_values(
    self,
    adata: anndata.AnnData,
    missing_strategy: str = "hybrid",
    imputation_method: str = "knn",
    filter_threshold: float = 0.7,
    min_valid_values: int = 3
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

def normalize_intensities(
    self,
    adata: anndata.AnnData,
    method: str = "tmm",
    log_transform: bool = True,
    center_median: bool = True
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ProteomicsQualityService:
    """Service for assessing proteomics data quality."""

def assess_data_quality(
    self,
    adata: anndata.AnnData,
    cv_threshold: float = 0.3,
    missing_threshold: float = 0.5
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ProteomicsAnalysisService:
    """Service for proteomics statistical analysis and pathway enrichment."""

def perform_differential_analysis(
    self,
    adata: anndata.AnnData,
    group_col: str,
    reference_group: str = None,
    method: str = "limma",
    adjust_method: str = "BH"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ClinicalMetadataService:
    """Service for processing and validating clinical trial metadata."""

def __init__(self, data_manager: DataManagerV2, cycle_length_days: int = 21) -> None

def process_sample_metadata(
    self,
    metadata_df: pd.DataFrame,
    column_mapping: Optional[Dict[str, str]] = None,
    validate: bool = True
) -> Tuple[pd.DataFrame, Dict[str, Any], AnalysisStep]

from lobster_metadata.services.metadata import ClinicalMetadataService

service = ClinicalMetadataService(data_manager, cycle_length_days=21)

# Process with validation
processed_df, stats, ir = service.process_sample_metadata(
    metadata_df,
    column_mapping={'Sample_ID': 'sample_id', 'RECIST': 'response_status'}
)

print(f"Validated {stats['validation_rate']:.1f}% samples")

def create_responder_groups(
    self,
    metadata_df: pd.DataFrame,
    response_column: str = "response_status",
    sample_id_column: str = "sample_id",
    grouping_strategy: str = "orr",
) -> Tuple[Dict[str, List[str]], Dict[str, Any], AnalysisStep]

# ORR grouping (standard)
groups, stats, ir = service.create_responder_groups(
    metadata_df, grouping_strategy="orr"
)
print(f"Responders: {stats['responder_count']}")
print(f"Non-responders: {stats['non_responder_count']}")

# DCR grouping (immunotherapy trials)
groups, stats, ir = service.create_responder_groups(
    metadata_df, grouping_strategy="dcr"
)
print(f"Disease control: {stats['disease_control_count']}")
print(f"Progressive: {stats['progressive_count']}")
print(f"DCR: {stats['disease_control_rate']:.1f}%")

def get_timepoint_samples(
    self,
    metadata_df: pd.DataFrame,
    timepoint: str,
    timepoint_column: str = "timepoint",
    sample_id_column: str = "sample_id",
) -> Tuple[List[str], Dict[str, Any], AnalysisStep]

# Get baseline samples
baseline_ids, stats, ir = service.get_timepoint_samples(metadata_df, "Baseline")

# Get cycle 2 day 1 samples
c2d1_ids, stats, ir = service.get_timepoint_samples(metadata_df, "C2D1")

def filter_by_response_and_timepoint(
    self,
    metadata_df: pd.DataFrame,
    response_group: Optional[str] = None,
    timepoint: Optional[str] = None,
    response_column: str = "response_status",
    timepoint_column: str = "timepoint",
    sample_id_column: str = "sample_id",
) -> Tuple[List[str], Dict[str, Any], AnalysisStep]

# Get responders at baseline
ids, stats, ir = service.filter_by_response_and_timepoint(
    metadata_df, response_group="responder", timepoint="Baseline"
)

# 1. Load proteomics data
adata = data_manager.get_modality("biognosys_proteomics")

# 2. Process clinical metadata
clinical_service = ClinicalMetadataService(data_manager)
processed, stats, ir = clinical_service.process_sample_metadata(adata.obs)

# 3. Create responder groups (DCR for immunotherapy)
groups, stats, ir = clinical_service.create_responder_groups(
    processed, grouping_strategy="dcr"
)

# 4. Filter proteomics by response
responder_adata = adata[groups["disease_control"], :]
progressive_adata = adata[groups["progressive"], :]

class GEOService:
    """Service for fetching and processing GEO datasets."""

def fetch_metadata_only(
    self,
    geo_id: str,
    include_sample_info: bool = True
) -> Tuple[Dict[str, Any], Dict[str, Any]]

def download_and_process(
    self,
    geo_id: str,
    sample_limit: Optional[int] = None,
    concatenation_strategy: str = "guided"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ContentAccessService:
    """
    Unified publication access service with capability-based routing.

    Provides 10 core methods organized into 4 categories:
    - Discovery (3): search_literature, discover_datasets, find_linked_datasets
    - Metadata (2): extract_metadata, validate_metadata
    - Content (3): get_abstract, get_full_content, extract_methods
    - System (1): query_capabilities

    Features:
    - 5 specialized providers with automatic routing
    - Three-tier cascade: PMC XML → Webpage → PDF
    - Session caching via DataManager
    - W3C-PROV provenance tracking
    """

ContentAccessService (Coordination Layer)
    ↓
ProviderRegistry (Capability-Based Routing)
    ↓
5 Providers:
  - AbstractProvider (Priority: 10) - Fast abstracts (200-500ms)
  - PubMedProvider (Priority: 10) - Literature search (1-3s)
  - GEOProvider (Priority: 10) - Dataset discovery (2-5s)
  - PMCProvider (Priority: 10) - PMC XML (500ms-2s, 30-40% coverage)
  - WebpageProvider (Priority: 50) - Webpage/PDF fallback (2-8s)
    ↓
DataManagerV2 (Session Caching + Provenance)

def search_literature(
    self,
    query: str,
    max_results: int = 5,
    sources: Optional[list[str]] = None,
    filters: Optional[dict[str, any]] = None,
    **kwargs
) -> Tuple[str, Dict[str, Any], AnalysisStep]

service = ContentAccessService(data_manager)

# Basic literature search
results, stats, ir = service.search_literature("BRCA1 breast cancer")
print(f"Found {stats['results_count']} papers in {stats['execution_time_ms']}ms")

# With source filter
results, stats, ir = service.search_literature("p53", sources=["pubmed"])

# With year filter
results, stats, ir = service.search_literature(
    "single-cell RNA-seq",
    max_results=10,
    filters={"publication_year": "2023"}
)

def discover_datasets(
    self,
    query: str,
    dataset_type: "DatasetType",
    max_results: int = 5,
    filters: Optional[dict[str, str]] = None
) -> Tuple[str, Dict[str, Any], AnalysisStep]

from lobster.tools.providers.base_provider import DatasetType

# Direct accession (auto-detected)
results, stats, ir = service.discover_datasets("GSM6204600", DatasetType.GEO)
if stats['accession_detected']:
    print(f"Parent series: {stats.get('parent_series', 'N/A')}")

# Text search
results, stats, ir = service.discover_datasets(
    "single-cell RNA-seq breast cancer",
    DatasetType.GEO,
    max_results=10,
    filters={"organism": "human"}
)

def find_linked_datasets(
    self,
    identifier: str,
    dataset_types: Optional[list["DatasetType"]] = None,
    include_related: bool = True
) -> str

# Find all linked datasets
results = service.find_linked_datasets("PMID:35042229")

# Filter to GEO and SRA
results = service.find_linked_datasets(
    "PMID:35042229",
    dataset_types=[DatasetType.GEO, DatasetType.SRA]
)

def extract_metadata(
    self,
    identifier: str,
    source: Optional[str] = None
) -> Union["PublicationMetadata", str]

# Publication metadata
metadata = service.extract_metadata("PMID:35042229")
print(f"Title: {metadata.title}")
print(f"Authors: {metadata.authors}")
print(f"Year: {metadata.year}")

# Dataset metadata
metadata = service.extract_metadata("GSE180759", source="geo")

def validate_metadata(
    self,
    dataset_id: str,
    required_fields: Optional[List[str]] = None,
    required_values: Optional[Dict[str, List[str]]] = None,
    threshold: float = 0.8
) -> str

# Validate with specific required fields
report = service.validate_metadata(
    "GSE180759",
    required_fields=["smoking_status", "treatment_response"],
    threshold=0.8
)
print(report)
# Output:
# ✅ PROCEED - 95% completeness (19/20 samples)
# - smoking_status: 100% (20/20 samples)
# - treatment_response: 90% (18/20 samples)

# Validate with required values
report = service.validate_metadata(
    "GSE111111",
    required_values={"condition": ["control", "normal"]},
    threshold=1.0  # All samples must be controls
)

def get_abstract(
    self,
    identifier: str,
    force_refresh: bool = False
) -> dict[str, any]

abstract = service.get_abstract("PMID:35042229")
print(f"Title: {abstract['title']}")
print(f"Authors: {', '.join(abstract['authors'][:3])}")
print(f"Abstract: {abstract['abstract'][:200]}...")

def get_full_content(
    self,
    source: str,
    prefer_webpage: bool = True,
    keywords: Optional[list[str]] = None,
    max_paragraphs: int = 100,
    max_retries: int = 2
) -> dict[str, any]

# PMC available (fast path)
content = service.get_full_content("PMID:35042229")
print(f"Tier: {content['tier_used']}")  # "full_pmc_xml"
print(f"Time: {content['extraction_time']:.2f}s")  # ~1s
print(f"Methods: {content['methods_text'][:200]}...")

# Webpage extraction
content = service.get_full_content("https://www.nature.com/articles/...")
print(f"Tier: {content['tier_used']}")  # "full_webpage"

# PDF fallback
content = service.get_full_content("https://biorxiv.org/.../file.pdf")
print(f"Tier: {content['tier_used']}")  # "full_pdf"

# Check software detected
print(f"Software: {content['metadata']['software']}")
print(f"GitHub repos: {content['metadata']['github_repos']}")

def extract_methods(
    self,
    content_result: dict[str, any],
    llm: Optional[any] = None,
    include_tables: bool = True
) -> dict[str, any]

# Get full content first
content = service.get_full_content("PMID:35042229")

# Extract methods
methods = service.extract_methods(content, include_tables=True)
print(f"Software: {methods['software_used']}")
print(f"GitHub repos: {methods['github_repos']}")
print(f"Tables: {len(methods.get('tables', []))}")

def query_capabilities(self) -> str

capabilities = service.query_capabilities()
print(capabilities)

======================================================================
LOBSTER CONTENT ACCESS SERVICE - CAPABILITY MATRIX
======================================================================

📋 AVAILABLE OPERATIONS:

  Discovery & Search:
    ✅ SEARCH_LITERATURE              → PubMedProvider
    ✅ DISCOVER_DATASETS              → GEOProvider
    ✅ FIND_LINKED_DATASETS           → PubMedProvider

  Metadata & Validation:
    ✅ EXTRACT_METADATA               → PubMedProvider, GEOProvider
    ✅ VALIDATE_METADATA              → GEOProvider

  Content Retrieval:
    ✅ GET_ABSTRACT                   → AbstractProvider
    ✅ GET_FULL_CONTENT               → PMCProvider, WebpageProvider

🔧 REGISTERED PROVIDERS:

  • AbstractProvider (Priority: 10)
    Capabilities: GET_ABSTRACT

  • PubMedProvider (Priority: 10)
    Capabilities: SEARCH_LITERATURE, FIND_LINKED_DATASETS, EXTRACT_METADATA

  • GEOProvider (Priority: 10)
    Capabilities: DISCOVER_DATASETS, EXTRACT_METADATA, VALIDATE_METADATA

  • PMCProvider (Priority: 10)
    Capabilities: GET_FULL_CONTENT

  • WebpageProvider (Priority: 50)
    Capabilities: GET_FULL_CONTENT

💾 SUPPORTED DATASET TYPES:

  ✅ GEO                    → GEOProvider

⚡ PERFORMANCE TIERS:

  Tier 1 (Fast): <500ms
    - get_abstract: AbstractProvider
    - search_literature: PubMedProvider

  Tier 2 (Moderate): 500ms-2s
    - get_full_content (PMC): PMCProvider
    - extract_metadata: PubMedProvider, GEOProvider

  Tier 3 (Slow): 2-8s
    - get_full_content (Webpage): WebpageProvider
    - get_full_content (PDF): WebpageProvider + DoclingService

🔄 CASCADE LOGIC:

  Full Content Retrieval:
    1. Check DataManager cache (fastest)
    2. Try PMC XML (Priority 10, 30-40% coverage)
    3. Fallback: Webpage HTML (Priority 50)
    4. Final fallback: PDF via Docling (Priority 100)

======================================================================

class PublicationResolver:
    """Resolver for converting identifiers to accessible PDF URLs."""

    def resolve(self, identifier: str) -> PublicationResolutionResult:
        """
        Resolve DOI/PMID to accessible URL using tiered waterfall strategy.

        Auto-detects identifier type and applies appropriate resolution method.
        """

# For PMID input
identifier = "PMID:39370688"
# → Checks PMC API for open access version
# → Returns: "https://www.ncbi.nlm.nih.gov/pmc/articles/PMC12496192/pdf/"
# → Quality: Very high (government repository, reliable)

# For DOI starting with 10.1101
identifier = "10.1101/2024.08.29.610467"
# → Resolves to bioRxiv/medRxiv PDF
# → Returns: "https://www.biorxiv.org/content/10.1101/2024.08.29.610467.full.pdf"
# → Quality: High (preprints, usually accessible)

# For non-preprint DOIs
identifier = "10.1038/s41586-025-09686-5"
# → Uses CrossRef API to find publisher URL
# → Returns: "https://www.nature.com/articles/s41586-025-09686-5"
# → Quality: Medium (may be paywalled)

def resolve(self, identifier: str) -> PublicationResolutionResult

from lobster.tools.providers.publication_resolver import PublicationResolver

resolver = PublicationResolver()

# Resolve bioRxiv DOI
result = resolver.resolve("10.1101/2024.08.29.610467")
print(f"PDF URL: {result.pdf_url}")  # https://www.biorxiv.org/content/...
print(f"Source: {result.source}")    # 'biorxiv'
print(f"Access: {result.access_type}")  # 'preprint'

# Resolve PMID to PMC
result = resolver.resolve("PMID:39370688")
print(f"PDF URL: {result.pdf_url}")  # https://www.ncbi.nlm.nih.gov/pmc/...
print(f"Source: {result.source}")    # 'pmc'
print(f"Access: {result.access_type}")  # 'open_access'

# Handle paywalled DOI gracefully
result = resolver.resolve("10.18632/aging.204666")
print(f"Accessible: {result.is_accessible()}")  # False
print(f"Access type: {result.access_type}")     # 'paywalled'
print(f"Suggestions: {result.suggestions}")     # Alternative access methods

def batch_resolve(
    self,
    identifiers: List[str],
    max_batch: int = 10
) -> List[PublicationResolutionResult]

identifiers = [
    "10.1101/2024.08.29.610467",  # bioRxiv DOI
    "PMID:39370688",              # PMID
    "10.18632/aging.204666",      # Potentially paywalled DOI
]

results = resolver.batch_resolve(identifiers)

for i, result in enumerate(results):
    print(f"Paper {i+1}: {result.source} ({result.access_type})")
    if result.is_accessible():
        print(f"  → {result.pdf_url}")
    else:
        print(f"  → {result.suggestions}")

class ConcatenationService:
    """Service for concatenating samples with batch correction and validation."""

def concatenate_samples(
    self,
    adata_list: List[anndata.AnnData],
    batch_key: str = "batch",
    batch_correction_method: str = "harmony",
    join_method: str = "outer"
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class VisualizationService:
    """Service for creating publication-quality visualizations."""

def create_umap_plot(
    self,
    adata: anndata.AnnData,
    color_by: str = None,
    use_raw: bool = False,
    point_size: float = 1.0,
    alpha: float = 0.8,
    color_map: str = "viridis"
) -> go.Figure

def create_volcano_plot(
    self,
    results_df: pd.DataFrame,
    log2fc_col: str = "log2FoldChange",
    pvalue_col: str = "padj",
    significance_threshold: float = 0.05,
    fold_change_threshold: float = 1.0
) -> go.Figure

def create_heatmap(
    self,
    adata: anndata.AnnData,
    genes: List[str],
    groupby: str = None,
    use_raw: bool = False,
    standard_scale: str = None,
    cmap: str = "RdBu_r"
) -> go.Figure

class MLProteomicsService:
    """Alpha service for machine learning applications in proteomics."""

class MLTranscriptomicsService:
    """Alpha service for machine learning applications in transcriptomics."""

class SCVIEmbeddingService:
    """Service for scVI-based dimensionality reduction and batch correction."""

def train_scvi_model(
    self,
    adata: anndata.AnnData,
    batch_key: str = None,
    n_latent: int = 10,
    n_epochs: int = 400,
    early_stopping: bool = True
) -> Tuple[anndata.AnnData, Dict[str, Any], AnalysisStep]

class ServiceError(Exception):
    """Base exception for service operations."""
    pass

class PreprocessingError(ServiceError):
    """Exception for preprocessing operations."""
    pass

class AnalysisError(ServiceError):
    """Exception for analysis operations."""
    pass

class ValidationError(ServiceError):
    """Exception for validation operations."""
    pass

def handle_service_error(func):
    """Decorator for consistent service error handling."""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"Service error in {func.__name__}: {e}")
            raise ServiceError(f"Operation failed: {str(e)}") from e
    return wrapper

def set_progress_callback(self, callback: Callable[[int, str], None]) -> None:
    """
    Set a callback function to report progress.

    Args:
        callback: Function accepting (progress_percent, message)
    """
    self.progress_callback = callback

from lobster.tools.preprocessing_service import PreprocessingService
from lobster.tools.clustering_service import ClusteringService

# Initialize services
preprocess = PreprocessingService()
cluster = ClusteringService()

# Process data through pipeline
filtered_adata, filter_stats = preprocess.filter_cells_and_genes(adata)
normalized_adata, norm_stats = preprocess.normalize_data(filtered_adata)
clustered_adata, cluster_stats = cluster.cluster_and_visualize(normalized_adata)

def create_analysis_pipeline(services: List, params: List[Dict]) -> Callable:
    """Create a pipeline from multiple services."""
    def pipeline(adata: anndata.AnnData) -> Tuple[anndata.AnnData, Dict]:
        current_adata = adata
        all_stats = {}

        for service, param_dict in zip(services, params):
            current_adata, stats = service(**param_dict)(current_adata)
            all_stats.update(stats)

        return current_adata, all_stats

    return pipeline

def validate_input(self, adata: anndata.AnnData) -> None:
    """Validate AnnData input for service operations."""
    if adata is None:
        raise ValueError("AnnData object cannot be None")
    if adata.n_obs == 0:
        raise ValueError("No observations in AnnData object")
    if adata.n_vars == 0:
        raise ValueError("No variables in AnnData object")