Raw counts support, backed-mode differential expression, and auto-detection of gene symbols (#67)

* Update version to 0.18.0 and enhance raw counts handling in save_features_matrix

- Bump package version to 0.18.0.
- Introduce _resolve_raw_counts method in CyteType to improve raw counts extraction from AnnData.
- Add _is_integer_valued utility function to check if matrices contain integer values.
- Update save_features_matrix to handle raw counts and include them in the output HDF5 file.
- Enhance tests to cover new raw counts functionality and integer value checks.

* Add artifact paths for vars.h5 and obs.duckdb, enhance artifact building and uploading

- Introduced vars_h5_path and obs_duckdb_path parameters in CyteType for customizable artifact paths.
- Implemented caching of raw counts and improved error handling during artifact creation.
- Updated _upload_artifacts method to handle pre-built artifacts and log errors appropriately.
- Modified integration tests to accommodate new parameters and ensure proper artifact cleanup.

* Refactor artifact cleanup in CyteType and update tests

- Replaced the static method _cleanup_artifact_files with an instance method cleanup to manage artifact file deletion after run completion.
- Removed the cleanup_artifacts parameter from run method, simplifying the interface.
- Updated integration tests to verify that cleanup correctly deletes artifact files and clears associated paths.

* Add rank_genes_groups_backed function and update exports

- Introduced rank_genes_groups_backed in marker_detection.py for memory-efficient gene ranking on backed AnnData objects.
- Updated __init__.py files to include rank_genes_groups_backed in the public API of cytetype and preprocessing modules.
- Refactored code for improved readability in main.py, enhancing the formatting of artifact cleanup logic.

* Enhance gene symbol handling in CyteType

- Introduced resolve_gene_symbols_column function to auto-detect gene symbols in AnnData, improving flexibility in gene symbol management.
- Updated gene_symbols_column type to accept None, allowing for better handling of cases where gene symbols are not explicitly provided.
- Refactored aggregate_expression_percentages and extract_marker_genes functions to accommodate the new gene symbol resolution logic.
- Enhanced validation in _validate_gene_symbols_column to provide clearer warnings about potential gene ID misclassifications.

* Update batch size for expression percentage calculations and refactor aggregation logic

- Increased the default batch size for calculating expression percentages from 2000 to 5000 to optimize memory usage.
- Refactored the aggregate_expression_percentages function to utilize a single-pass row-batched accumulation method for improved performance.
- Introduced a new _accumulate_group_stats function to streamline the computation of per-group statistics, enhancing efficiency for large datasets.
- Updated related documentation to reflect changes in parameters and processing logic.

* Refactor logging and enhance progress reporting in CyteType

- Removed unnecessary logging statements for calculating expression percentages and extracting visualization coordinates to streamline output.
- Updated logging message for saving obs.duckdb artifact for clarity.
- Integrated progress reporting using tqdm for batch processing in save_features_matrix and extract_visualization_coordinates functions.
- Improved handling of warnings during batch processing to suppress FutureWarnings from tqdm.
- Adjusted progress descriptions for better user feedback during long-running operations.

* Add WRITE_MEM_BUDGET constant and enhance logging in CyteType

- Introduced WRITE_MEM_BUDGET constant in config.py to define memory budget for writing artifacts.
- Updated logging messages in main.py for clarity during artifact saving processes.
- Enhanced progress reporting in artifact writing functions to improve user feedback.
- Refactored warning handling to suppress FutureWarnings from tqdm during batch processing.
- Added new functions in artifacts.py for improved handling of sparse matrix writing and progress tracking.

* Enhance file upload functionality and error handling in CyteType

- Increased maximum upload size for vars_h5 from 10GB to 50GB to accommodate larger datasets.
- Introduced a new ClientDisconnectedError exception to handle client disconnection scenarios.
- Improved progress reporting during file uploads by integrating tqdm for better user feedback.
- Refactored upload logic to ensure consistent progress updates and error handling across different upload scenarios.

* Add subsampling functionality to preprocessing module

- Introduced a new `subsample_by_group` function in `subsampling.py` to limit the number of cells per group in an AnnData object.
- Updated `__init__.py` to include `subsample_by_group` in the public API of the preprocessing module.
- Enhanced error handling to check for the existence of the specified group key in the AnnData object.
- Added logging to report the results of the subsampling process.

* Refactor subsampling functionality and improve logging in preprocessing module

- Enhanced the `subsample_by_group` function to optimize performance and memory usage during subsampling.
- Improved logging to provide clearer insights into the subsampling process and results.
- Updated error handling to ensure robustness when dealing with edge cases in AnnData objects.
- Refactored related tests to validate the new subsampling logic and logging enhancements.

* formatted

* Update subsampling logic to merge subsets by taking the first occurrence in the preprocessing module

- Modified the `subsample_by_group` function to use `merge="first"` when concatenating subsampled subsets, ensuring that the first occurrence of each observation is retained.
- This change enhances the subsampling process by providing a more consistent output when merging groups.

* Enhance gene name processing in preprocessing module

- Added `clean_gene_names` function to extract gene symbols from composite gene names, improving the handling of gene identifiers.
- Updated `extract_marker_genes` to utilize `clean_gene_names` for better gene name management.
- Integrated `clean_gene_names` into the `CyteType` class for consistent gene name processing across the module.
- Enhanced logging to provide insights when composite gene values are cleaned.

* Optimize group statistics accumulation for sparse matrices in marker detection

- Enhanced the `_accumulate_group_stats` function to handle both sparse and dense matrix inputs efficiently.
- Implemented conditional logic to process sparse matrices using CSR format, improving memory usage and performance.
- Maintained existing functionality for dense matrices, ensuring compatibility with previous implementations.

* Increase default timeout for file uploads in CyteType

- Updated the timeout settings in both `main.py` and `client.py` from 30 seconds to 60 seconds to allow for longer upload durations, improving reliability for larger files.

* fomatted

* Refactor subsampling logic in `_is_integer_valued` function to improve row selection

- Updated the logic to select rows for sampling based on the number of rows in the input matrix.
- Implemented random sampling when the number of rows exceeds the specified sample size, ensuring a more representative subset.
- Maintained functionality for cases where the number of rows is less than or equal to the sample size.

* Update public API in `__init__.py` to include new plotting and subsampling functions

- Added `marker_dotplot` and `subsample_by_group` to the `__all__` list, making them accessible for import.
- This change enhances the module's functionality by exposing additional features for users.

Author: parashardhapola

cytetype/__init__.py

@@ -1,11 +1,14 @@
-__version__ = "0.17.0"
+__version__ = "0.18.0"
 
 import requests
 
 from .config import logger
 from .main import CyteType
+from .plotting import marker_dotplot
+from .preprocessing.marker_detection import rank_genes_groups_backed
+from .preprocessing.subsampling import subsample_by_group
 
-__all__ = ["CyteType"]
+__all__ = ["CyteType", "marker_dotplot", "rank_genes_groups_backed", "subsample_by_group"]
 
 _PYPI_JSON_URL = "https://pypi.org/pypi/cytetype/json"
 

cytetype/api/client.py

@@ -14,19 +14,32 @@
 
 MAX_UPLOAD_BYTES: dict[UploadFileKind, int] = {
     "obs_duckdb": 100 * 1024 * 1024,  # 100MB
-    "vars_h5": 10 * 1024 * 1024 * 1024,  # 10GB
+    "vars_h5": 50 * 1024 * 1024 * 1024,  # 10GB
 }
 
 _CHUNK_RETRY_DELAYS = (1, 5, 20)
 _RETRYABLE_API_ERROR_CODES = frozenset({"INTERNAL_ERROR", "HTTP_ERROR"})
 
 
+def _try_import_tqdm() -> type | None:
+    try:
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            from tqdm.auto import tqdm
+
+        return tqdm  # type: ignore[no-any-return]
+    except ImportError:
+        return None
+
+
 def _upload_file(
     base_url: str,
     auth_token: str | None,
     file_kind: UploadFileKind,
     file_path: str,
-    timeout: float | tuple[float, float] = (30.0, 3600.0),
+    timeout: float | tuple[float, float] = (60.0, 3600.0),
     max_workers: int = 4,
 ) -> UploadResponse:
     path_obj = Path(file_path)
@@ -62,6 +75,12 @@ def _upload_file(
     # Memory is bounded to ~max_workers × chunk_size because each thread
     # reads its chunk on demand via seek+read.
     _tls = threading.local()
+    tqdm_cls = _try_import_tqdm()
+    pbar = (
+        tqdm_cls(total=n_chunks, desc="Uploading", unit="chunk")
+        if tqdm_cls is not None and n_chunks > 0
+        else None
+    )
     _progress_lock = threading.Lock()
     _chunks_done = [0]
 
@@ -82,15 +101,18 @@ def _upload_chunk(chunk_idx: int) -> None:
                     data=chunk_data,
                     timeout=timeout,
                 )
-                with _progress_lock:
-                    _chunks_done[0] += 1
-                    done = _chunks_done[0]
-                pct = 100 * done / n_chunks
-                print(
-                    f"\r  Uploading: {done}/{n_chunks} chunks ({pct:.0f}%)",
-                    end="",
-                    flush=True,
-                )
+                if pbar is not None:
+                    pbar.update(1)
+                else:
+                    with _progress_lock:
+                        _chunks_done[0] += 1
+                        done = _chunks_done[0]
+                    pct = 100 * done / n_chunks
+                    print(
+                        f"\r  Uploading: {done}/{n_chunks} chunks ({pct:.0f}%)",
+                        end="",
+                        flush=True,
+                    )
                 return
             except (NetworkError, TimeoutError) as exc:
                 last_exc = exc
@@ -103,13 +125,9 @@ def _upload_chunk(chunk_idx: int) -> None:
             if attempt < len(_CHUNK_RETRY_DELAYS):
                 delay = _CHUNK_RETRY_DELAYS[attempt]
                 logger.warning(
-                    "Chunk %d/%d upload failed (attempt %d/%d), retrying in %ds: %s",
-                    chunk_idx + 1,
-                    n_chunks,
-                    attempt + 1,
-                    1 + len(_CHUNK_RETRY_DELAYS),
-                    delay,
-                    last_exc,
+                    f"Chunk {chunk_idx + 1}/{n_chunks} upload failed "
+                    f"(attempt {attempt + 1}/{1 + len(_CHUNK_RETRY_DELAYS)}), "
+                    f"retrying in {delay}s: {last_exc}"
                 )
                 time.sleep(delay)
 
@@ -120,9 +138,17 @@ def _upload_chunk(chunk_idx: int) -> None:
         try:
             with ThreadPoolExecutor(max_workers=effective_workers) as pool:
                 list(pool.map(_upload_chunk, range(n_chunks)))
-            print(f"\r  \033[92m✓\033[0m Uploaded {n_chunks}/{n_chunks} chunks (100%)")
+            if pbar is not None:
+                pbar.close()
+            else:
+                print(
+                    f"\r  \033[92m✓\033[0m Uploaded {n_chunks}/{n_chunks} chunks (100%)"
+                )
         except BaseException:
-            print()  # ensure newline on failure
+            if pbar is not None:
+                pbar.close()
+            else:
+                print()
             raise
 
     # Step 3 – Complete upload (returns same UploadResponse shape as before)
@@ -136,7 +162,7 @@ def upload_obs_duckdb(
     base_url: str,
     auth_token: str | None,
     file_path: str,
-    timeout: float | tuple[float, float] = (30.0, 3600.0),
+    timeout: float | tuple[float, float] = (60.0, 3600.0),
     max_workers: int = 4,
 ) -> UploadResponse:
     return _upload_file(
@@ -153,7 +179,7 @@ def upload_vars_h5(
     base_url: str,
     auth_token: str | None,
     file_path: str,
-    timeout: float | tuple[float, float] = (30.0, 3600.0),
+    timeout: float | tuple[float, float] = (60.0, 3600.0),
     max_workers: int = 4,
 ) -> UploadResponse:
     return _upload_file(

cytetype/api/exceptions.py

@@ -56,6 +56,12 @@ class LLMValidationError(APIError):
     pass
 
 
+class ClientDisconnectedError(APIError):
+    """Server detected client disconnection mid-request - CLIENT_DISCONNECTED (HTTP 499)."""
+
+    pass
+
+
 # Client-side errors with default messages
 class TimeoutError(CyteTypeError):
     """Client-side timeout waiting for results."""
@@ -87,6 +93,7 @@ def __init__(
     "JOB_NOT_FOUND": JobNotFoundError,
     "JOB_FAILED": JobFailedError,
     "LLM_VALIDATION_FAILED": LLMValidationError,
+    "CLIENT_DISCONNECTED": ClientDisconnectedError,
     "JOB_PROCESSING": APIError,  # Generic - expected during polling
     "JOB_NOT_COMPLETED": APIError,  # Generic
     "HTTP_ERROR": APIError,  # Generic

cytetype/config.py

@@ -24,3 +24,5 @@ def _log_format(record: Record) -> str:
     level="INFO",
     format=_log_format,
 )
+
+WRITE_MEM_BUDGET: int = 4 * 1024 * 1024 * 1024  # 4 GB