diff --git a/bedboss/bbuploader/__init__.py b/bedboss/bbuploader/__init__.py index 30f889ae..56009817 100644 --- a/bedboss/bbuploader/__init__.py +++ b/bedboss/bbuploader/__init__.py @@ -1,13 +1 @@ """Package-level data""" - -import coloredlogs -import logmuse - -from bedboss.bbuploader.constants import PKG_NAME - -_LOGGER = logmuse.init_logger(PKG_NAME) -coloredlogs.install( - logger=_LOGGER, - datefmt="%H:%M:%S", - fmt="[%(levelname)s] [%(asctime)s] [BBUPLOADER] %(message)s", -) diff --git a/bedboss/bbuploader/main.py b/bedboss/bbuploader/main.py index ee626d0c..a8ee11c7 100644 --- a/bedboss/bbuploader/main.py +++ b/bedboss/bbuploader/main.py @@ -753,7 +753,6 @@ def _upload_gse( bedbase_config=bedbase_config, narrowpeak=required_metadata.narrowpeak, other_metadata=required_metadata.pep.model_dump(), - upload_pephub=True, upload_s3=True, upload_qdrant=True, force_overwrite=overwrite, @@ -831,7 +830,6 @@ def _upload_gse( name=gse, description=project.description, heavy=False, # TODO: set to False because can't handle bedset > 10 files - upload_pephub=True, upload_s3=True, no_fail=True, force_overwrite=overwrite_bedset, diff --git a/bedboss/bedboss.py b/bedboss/bedboss.py index 212abd33..0491b9d9 100644 --- a/bedboss/bedboss.py +++ b/bedboss/bedboss.py @@ -79,7 +79,6 @@ def run_all( update: bool = False, upload_qdrant: bool = False, upload_s3: bool = False, - upload_pephub: bool = False, lite: bool = False, # Universes universe: bool = False, @@ -90,7 +89,7 @@ def run_all( reference_genome_validator: ReferenceValidator = None, ) -> str: """ - Run bedboss: bedmaker -> bedqc -> bedclassifier -> bedstat -> upload to s3, qdrant, pephub, and bedbase. + Run bedboss: bedmaker -> bedqc -> bedclassifier -> bedstat -> upload to s3, qdrant, and bedbase. Args: input_file: Input file path. @@ -113,7 +112,6 @@ def run_all( update: Whether to update the record in the database. If True, overwrites 'force_overwrite'. Default: False. upload_qdrant: Whether to upload to qdrant. Default: False. upload_s3: Whether to upload to s3. - upload_pephub: Whether to push bedfiles and metadata to pephub. Default: False. lite: Whether to run lite version of the pipeline. Default: False. universe: Whether to add the sample as the universe. Default: False. universe_method: Method used to create the universe. @@ -280,7 +278,6 @@ def run_all( ref_validation=ref_valid_stats, license_id=license_id, upload_qdrant=upload_qdrant and not lite, - upload_pephub=upload_pephub, upload_s3=upload_s3, local_path=outfolder, overwrite=True, @@ -298,7 +295,6 @@ def run_all( ref_validation=ref_valid_stats, license_id=license_id, upload_qdrant=upload_qdrant and not lite, - upload_pephub=upload_pephub, upload_s3=upload_s3, local_path=outfolder, overwrite=force_overwrite, @@ -337,7 +333,6 @@ def insert_pep( force_overwrite: bool = False, update: bool = False, upload_s3: bool = False, - upload_pephub: bool = False, upload_qdrant: bool = False, no_fail: bool = False, standardize_pep: bool = False, @@ -366,7 +361,6 @@ def insert_pep( force_overwrite: Whether to overwrite the existing record. update: Whether to update the record in the database. Overwrites force_overwrite. Default: False. upload_s3: Whether to upload to s3. - upload_pephub: Whether to push bedfiles and metadata to pephub. Default: False. upload_qdrant: Whether to execute qdrant indexing. no_fail: Whether to raise an error if bedset was not added to the database. standardize_pep: Whether to standardize the pep file before processing by using bedms. Default: False. @@ -458,7 +452,6 @@ def insert_pep( update=update, upload_qdrant=upload_qdrant, upload_s3=upload_s3, - upload_pephub=upload_pephub, universe=pep_sample.get("universe"), universe_method=pep_sample.get("universe_method"), universe_bedset=pep_sample.get("universe_bedset"), @@ -486,7 +479,6 @@ def insert_pep( output_folder=output_folder, description=pep.description, heavy=bedset_heavy, - upload_pephub=upload_pephub, upload_s3=upload_s3, no_fail=no_fail, force_overwrite=force_overwrite, @@ -608,7 +600,6 @@ def reprocess_all( update=True, upload_qdrant=True, upload_s3=True, - upload_pephub=True, lite=False, universe=False, universe_method=None, @@ -714,7 +705,6 @@ def reprocess_one( update=True, upload_qdrant=True, upload_s3=True, - upload_pephub=True, lite=False, universe=False, universe_method=None, @@ -764,7 +754,6 @@ def reprocess_bedset( output_folder=output_folder, description=bedset_annot.description, heavy=heavy, - upload_pephub=False, upload_s3=heavy, no_fail=no_fail, force_overwrite=True, diff --git a/bedboss/bedboss_hpc.py b/bedboss/bedboss_hpc.py index b305b6fa..2e090b81 100644 --- a/bedboss/bedboss_hpc.py +++ b/bedboss/bedboss_hpc.py @@ -90,7 +90,6 @@ class BoolFlagSpec(BaseModel): "update": BoolFlagSpec(on="--update"), "upload_qdrant": BoolFlagSpec(on="--upload-qdrant", off="--no-upload-qdrant"), "upload_s3": BoolFlagSpec(on="--upload-s3", off="--no-upload-s3"), - "upload_pephub": BoolFlagSpec(on="--upload-pephub", off="--no-upload-pephub"), "no_fail": BoolFlagSpec(on="--no-fail"), "standardize_pep": BoolFlagSpec(on="--standardize-pep"), "lite": BoolFlagSpec(on="--lite"), @@ -160,7 +159,6 @@ class RunPepArgs(BaseModel): update: bool = False upload_qdrant: bool = True upload_s3: bool = True - upload_pephub: bool = True no_fail: bool = False license_id: Optional[str] = None standardize_pep: bool = False diff --git a/bedboss/bedbuncher/bedbuncher.py b/bedboss/bedbuncher/bedbuncher.py index ebcbc43a..8460b52a 100644 --- a/bedboss/bedbuncher/bedbuncher.py +++ b/bedboss/bedbuncher/bedbuncher.py @@ -99,7 +99,6 @@ def run_bedbuncher( description: str = None, annotation: dict = None, heavy: bool = False, - upload_pephub: bool = False, upload_s3: bool = False, no_fail: bool = False, force_overwrite: bool = False, @@ -118,7 +117,6 @@ def run_bedbuncher( annotation: Bedset annotation (author, source, summary, etc.). heavy: Whether to use heavy processing (add all columns to the database). If False, R-script won't be executed, only basic statistics will be calculated. - upload_pephub: Whether to create a view in pephub. upload_s3: Whether to upload files to s3. no_fail: Whether to raise an error if bedset was not added to the database. force_overwrite: Whether to overwrite the record in the database. @@ -158,7 +156,6 @@ def run_bedbuncher( bedid_list=bed_set, statistics=True, description=description, - upload_pephub=upload_pephub, upload_s3=upload_s3, plots=plots.model_dump(exclude_none=True, exclude_unset=True) if plots else {}, local_path=output_folder, @@ -175,7 +172,6 @@ def run_bedbuncher_form_pep( output_folder: str, bedset_name: str = None, heavy: bool = False, - upload_pephub: bool = False, upload_s3: bool = False, no_fail: bool = False, force_overwrite: bool = False, @@ -190,7 +186,6 @@ def run_bedbuncher_form_pep( bedset_name: Name of the bedset. heavy: Whether to use heavy processing (add all columns to the database). If False, R-script won't be executed, only basic statistics will be calculated. - upload_pephub: Whether to create a view in pephub. upload_s3: Whether to upload files to s3. no_fail: Whether to raise an error if bedset was not added to the database. force_overwrite: Whether to overwrite the record in the database. @@ -223,7 +218,6 @@ def run_bedbuncher_form_pep( name=bedset_name or pep_of_bed.name, description=pep_of_bed.description, heavy=heavy, - upload_pephub=upload_pephub, upload_s3=upload_s3, no_fail=no_fail, force_overwrite=force_overwrite, @@ -244,7 +238,6 @@ def run_bedbuncher_form_pep( # description="This is a description", # upload_s3=True, # no_fail=True, -# upload_pephub=True, # heavy=True, # ) @@ -252,5 +245,4 @@ def run_bedbuncher_form_pep( # "/home/bnt4me/virginia/repos/bbuploader/config_db_local.yaml", # bedset_name="pephub_test", # bedset_pep="khoroshevskyi/bedbunch:default", -# upload_pephub=True, # ) diff --git a/bedboss/bedclassifier/bedclassifier.py b/bedboss/bedclassifier/bedclassifier.py index f2e93c2d..45aad686 100644 --- a/bedboss/bedclassifier/bedclassifier.py +++ b/bedboss/bedclassifier/bedclassifier.py @@ -108,7 +108,7 @@ def _read_bed_file(filepath: str, skiprows: int = 0) -> pd.DataFrame | None: else: raise BedTypeException(reason="Input is not a string or dataframe.") - df = df.dropna(axis=1) + df = df.dropna(axis=1, how="all") df.columns = range(len(df.columns)) num_cols = len(df.columns) compliant_columns = 0 diff --git a/bedboss/cli.py b/bedboss/cli.py index f3a8c4b2..ef4507ce 100644 --- a/bedboss/cli.py +++ b/bedboss/cli.py @@ -1,6 +1,7 @@ import os from importlib.metadata import version as _pkg_version +import logmuse import typer __version__ = _pkg_version("bedboss") @@ -8,6 +9,8 @@ from bedboss.bbuploader.cli import app_bbuploader from bedboss.qdrant_index.qdrant_cli import qdrant_app +from bedboss.scripts.analysis_files import files_app +from bedboss.scripts.snapshot import snapshot_app # commented and made new const here, because it speeds up help function, # from bbconf.const import DEFAULT_LICENSE @@ -100,7 +103,6 @@ def run_all( ), upload_qdrant: bool = typer.Option(False, help="Upload to Qdrant"), upload_s3: bool = typer.Option(False, help="Upload to S3"), - upload_pephub: bool = typer.Option(False, help="Upload to PEPHub"), # Universes universe: bool = typer.Option(False, help="Create a universe"), universe_method: str = typer.Option( @@ -144,7 +146,6 @@ def run_all( update=update, upload_qdrant=upload_qdrant, upload_s3=upload_s3, - upload_pephub=upload_pephub, universe=universe, universe_method=universe_method, universe_bedset=universe_bedset, @@ -181,7 +182,6 @@ def run_pep( ), upload_qdrant: bool = typer.Option(True, help="Upload to Qdrant"), upload_s3: bool = typer.Option(True, help="Upload to S3"), - upload_pephub: bool = typer.Option(True, help="Upload to PEPHub"), no_fail: bool = typer.Option(False, help="Do not fail on error"), license_id: str = typer.Option(DEFAULT_LICENSE, help="License ID"), standardize_pep: bool = typer.Option(False, help="Standardize the PEP using bedMS"), @@ -222,7 +222,6 @@ def run_pep( update=update, license_id=license_id, upload_s3=upload_s3, - upload_pephub=upload_pephub, upload_qdrant=upload_qdrant, no_fail=no_fail, standardize_pep=standardize_pep, @@ -267,7 +266,6 @@ def run_pep_hpc( update: bool = typer.Option(False, help="Update existing records"), upload_qdrant: bool = typer.Option(True, help="Upload to Qdrant"), upload_s3: bool = typer.Option(True, help="Upload to S3"), - upload_pephub: bool = typer.Option(True, help="Upload to PEPHub"), no_fail: bool = typer.Option(False, help="Do not fail on error"), license_id: str = typer.Option(DEFAULT_LICENSE, help="License ID"), standardize_pep: bool = typer.Option(False, help="Standardize the PEP using bedMS"), @@ -307,7 +305,6 @@ def run_pep_hpc( update=update, upload_qdrant=upload_qdrant, upload_s3=upload_s3, - upload_pephub=upload_pephub, no_fail=no_fail, license_id=license_id, standardize_pep=standardize_pep, @@ -421,7 +418,7 @@ def reprocess_bedset( ) -@app.command(help=f"Create a bed files form a [{', '.join(options_list)}] file") +@app.command(help=f"Create a bed files from a [{', '.join(options_list)}] file") def make_bed( input_file: str = typer.Option( ..., @@ -584,7 +581,6 @@ def make_bedset( False, help="Force overwrite the output files" ), upload_s3: bool = typer.Option(False, help="Upload to S3"), - upload_pephub: bool = typer.Option(False, help="Upload to PEPHub"), no_fail: bool = typer.Option(False, help="Do not fail on error"), ): from bedboss.bedbuncher.bedbuncher import run_bedbuncher_form_pep @@ -595,7 +591,6 @@ def make_bedset( output_folder=outfolder, bedset_name=bedset_name, heavy=heavy, - upload_pephub=upload_pephub, upload_s3=upload_s3, no_fail=no_fail, force_overwrite=force_overwrite, @@ -900,9 +895,25 @@ def common( version: bool = typer.Option( None, "--version", "-v", callback=version_callback, help="App version" ), + verbosity: int = typer.Option( + None, "--verbosity", help="Set logging level: 1 (CRITICAL) to 5 (DEBUG)" + ), + logdev: bool = typer.Option(False, "--logdev", help="Use developer logging format"), + silent: bool = typer.Option(False, "--silent", help="Silence logging"), ): - pass + # This callback runs before any command, so it is where logging gets + # configured. Don't move this into `__init__.py`. + # + # Configure the root logger so that bedboss and its dependencies (pipestat, + # geniml, bbconf, pephubclient, refgenconf, bbuploader) all log through one + # handler. Each line is tagged with its logger name, so there is no need to + # attach a per-package handler to identify the source. + logmuse.init_logger( + "", make_root=True, verbosity=verbosity, devmode=logdev, silent=silent + ) app.add_typer(app_bbuploader, name="geo") app.add_typer(qdrant_app, name="qdrant") +app.add_typer(snapshot_app, name="snapshot") +app.add_typer(files_app, name="files") diff --git a/bedboss/refgenome_validator/main.py b/bedboss/refgenome_validator/main.py index 8b82cfef..472a6ed0 100644 --- a/bedboss/refgenome_validator/main.py +++ b/bedboss/refgenome_validator/main.py @@ -368,9 +368,7 @@ def calculate_rating(self, compat_stats: CompatibilityStats) -> RatingModel: if sequence_fit < 0.90: points_rating += 1 if sequence_fit < 0.60: - points_rating += 1 - if sequence_fit < 0.60: - points_rating += 1 + points_rating += 2 else: # if no chrom names were found during assessment diff --git a/bedboss/scripts/analysis_files.py b/bedboss/scripts/analysis_files.py new file mode 100644 index 00000000..7f34893c --- /dev/null +++ b/bedboss/scripts/analysis_files.py @@ -0,0 +1,147 @@ +""" +Upload and manage standalone analysis files (openSignalMatrix, models, other +analysis inputs), plus the ``bedboss files`` CLI sub-app. + +Clean bedboss/bbconf boundary, same as ``bedboss snapshot``: + +- bedboss computes the local file's ``sha256`` and size, then hands an + ``AnalysisFileArtifact`` to ``agent.analysis_files.add(...)`` — a single + bbconf call that uploads the file to S3 (under the fixed ``analysis_files/`` + prefix) *and* records it in the ``analysis_files`` table. bedboss never + touches boto3 or the ORM directly. + +Heavy imports (the bbconf agent) are deferred into the functions that need them +so importing this module for its CLI app stays cheap. +""" + +from __future__ import annotations + +import hashlib +import logging +import os + +import typer + +from bedboss.const import PKG_NAME + +_LOGGER = logging.getLogger(PKG_NAME) + + +files_app = typer.Typer( + pretty_exceptions_short=False, + pretty_exceptions_show_locals=False, + help="Upload and manage standalone analysis files " + "(openSignalMatrix, models, other analysis inputs).", +) + + +def sha256_file(path: str, chunk: int = 1 << 20) -> str: + """Streaming SHA256 of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for block in iter(lambda: f.read(chunk), b""): + h.update(block) + return h.hexdigest() + + +@files_app.command( + "upload", + help="Upload an analysis file to S3 and record it in the analysis_files table.", +) +def files_upload( + path: str = typer.Option( + ..., + help="Local path to the file to upload", + exists=True, + file_okay=True, + readable=True, + ), + name: str = typer.Option(..., help="Logical name/key, e.g. openSignalMatrix"), + config: str = typer.Option(..., help="Path to the bedbase config file"), + file_type: str = typer.Option( + None, help="Category, e.g. openSignalMatrix | reference | model" + ), + genome: str = typer.Option(None, help="Genome/assembly, e.g. hg38 (optional)"), + description: str = typer.Option(None, help="Free-text description"), + tag: list[str] = typer.Option( + None, "--tag", help="Tag to attach; repeat for multiple tags" + ), +): + from bbconf import BedBaseAgent + from bbconf.models.base_models import AnalysisFileArtifact + + checksum = sha256_file(path) + file_size = os.path.getsize(path) + + artifact = AnalysisFileArtifact( + path=path, + name=name, + file_type=file_type, + genome=genome, + description=description, + tags=list(tag) if tag else None, + file_size=file_size, + checksum=checksum, + ) + + bbagent = BedBaseAgent(config, init_ml=False) + result = bbagent.analysis_files.add(artifact) + row = result.results[0] + print( + f"Uploaded analysis file '{row.name}' (id={row.id}) -> {row.file_path} " + f"[{file_size} bytes, sha256={checksum[:12]}…]" + ) + + +@files_app.command("list", help="List analysis files, newest first.") +def files_list( + config: str = typer.Option(..., help="Path to the bedbase config file"), + file_type: str = typer.Option(None, help="Filter by file type"), + genome: str = typer.Option(None, help="Filter by genome/assembly"), + tag: str = typer.Option(None, help="Filter by a single tag"), + limit: int = typer.Option(100, help="Maximum number of rows to return"), + offset: int = typer.Option(0, help="Number of rows to skip"), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + result = bbagent.analysis_files.list( + file_type=file_type, genome=genome, tag=tag, limit=limit, offset=offset + ) + print(f"Total analysis files: {result.count}") + for row in result.results: + print( + f"id={row.id} {row.creation_date} {(row.name or ''):20} " + f"type={row.file_type} genome={row.genome} " + f"cksum={(row.checksum or '')[:12]} {row.file_path}" + ) + + +@files_app.command("delete", help="Delete an analysis file by id.") +def files_delete( + id: int = typer.Option(..., help="Analysis file row id"), + config: str = typer.Option(..., help="Path to the bedbase config file"), + remove_s3: bool = typer.Option(True, help="Also delete the underlying S3 object"), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + bbagent.analysis_files.delete(id, remove_s3=remove_s3) + print(f"Analysis file {id} deleted from the bedbase database") + + +@files_app.command("delete-by-checksum", help="Delete analysis file(s) by checksum.") +def files_delete_by_checksum( + checksum: str = typer.Option(..., help="SHA256 checksum of the analysis file"), + config: str = typer.Option(..., help="Path to the bedbase config file"), + remove_s3: bool = typer.Option( + True, help="Also delete the underlying S3 object(s)" + ), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + bbagent.analysis_files.delete_by_checksum(checksum, remove_s3=remove_s3) + print( + f"Analysis file(s) with checksum {checksum} deleted from the bedbase database" + ) diff --git a/bedboss/scripts/snapshot.py b/bedboss/scripts/snapshot.py new file mode 100644 index 00000000..b370e9fc --- /dev/null +++ b/bedboss/scripts/snapshot.py @@ -0,0 +1,515 @@ +""" +Bulk metadata snapshot of the bedbase database, plus its ``bedboss snapshot`` +CLI sub-app. + +Two-phase flow with a clean bedboss/bbconf boundary: + +- **Phase 1 (bedboss, always runs).** Connect to the database *through bbconf* + (``BedBaseAgent(...).config.db_engine.engine``) and do all the querying, + streaming, and Parquet/manifest processing here. Whole tables of interest are + selected and joined (``bed`` LEFT JOIN ``bed_metadata``, plus ``bedsets`` and + ``bedfile_bedset_relation``); the export column set and the Parquet schema are + derived from the bbconf ORM models' columns, so they cannot drift. Rows are + streamed to zstd Parquet through server-side cursors and never fully + materialized in memory. Each file is gated against a pre-scan ``count`` taken + in the same ``REPEATABLE READ`` snapshot. + +- **Phase 2 (bbconf, only with ``publish=True``).** After phase 1 succeeds, the + built artifacts are handed to ``agent.snapshot.add(...)`` as + ``BedSnapshotArtifact`` models — a single bbconf call that uploads every file + to S3 (under the fixed ``snapshot/`` prefix) *and* records it in the + ``bed_snapshots`` table. bedboss never touches boto3 or the ORM directly. + +Heavy imports (pyarrow, sqlalchemy, the bbconf agent) are deferred into the +functions that need them so importing this module for its CLI app stays cheap. +""" + +from __future__ import annotations + +import datetime +import hashlib +import json +import logging +from pathlib import Path + +import typer + +from bedboss.const import PKG_NAME + +_LOGGER = logging.getLogger(PKG_NAME) + +# Bump when the exported column set or types change in a breaking way. +SCHEMA_VERSION = 1 + +# Rows fetched per server-side cursor round trip / written per Parquet batch. +DEFAULT_BATCH_SIZE = 20_000 + + +# --------------------------------------------------------------------------- # +# CLI sub-app: ``bedboss snapshot ...`` (registered in bedboss/cli.py). +# --------------------------------------------------------------------------- # + +snapshot_app = typer.Typer( + pretty_exceptions_short=False, + pretty_exceptions_show_locals=False, + help="Create and manage bulk metadata snapshots of the bedbase database.", +) + + +@snapshot_app.command( + "new", + help="Build a new bulk metadata snapshot (Parquet exports + manifest), " + "optionally publishing it to S3 and the database.", +) +def snapshot_new( + bedbase_config: str = typer.Option( + ..., + help="Path to the bedbase config file", + exists=True, + file_okay=True, + readable=True, + ), + outfolder: str = typer.Option( + "exports", help="Local directory for the built artifacts" + ), + batch_size: int = typer.Option( + DEFAULT_BATCH_SIZE, help="Rows per server-side fetch / Parquet write batch" + ), + fail_threshold: float = typer.Option( + 0.01, help="Abort without publishing if this fraction of rows is missing" + ), + publish: bool = typer.Option( + False, help="Upload artifacts to S3 and record rows in bed_snapshots" + ), +): + run_snapshot( + bedbase_config=bedbase_config, + output_dir=outfolder, + batch_size=batch_size, + fail_threshold=fail_threshold, + publish=publish, + ) + + +@snapshot_app.command("list", help="List bulk metadata snapshots, newest first.") +def snapshot_list( + config: str = typer.Option(..., help="Path to the bedbase config file"), + file_type: str = typer.Option(None, help="Filter by file type"), + limit: int = typer.Option(100, help="Maximum number of rows to return"), + offset: int = typer.Option(0, help="Number of rows to skip"), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + result = bbagent.snapshot.list(file_type=file_type, limit=limit, offset=offset) + print(f"Total snapshots: {result.count}") + for row in result.results: + print( + f"{row.creation_date} {row.file_type:18} " + f"rows={row.record_count} cksum={(row.checksum or '')[:12]} " + f"{row.file_path}" + ) + + +@snapshot_app.command("delete", help="Delete a snapshot index row by id.") +def snapshot_delete( + id: int = typer.Option(..., help="Snapshot row id"), + config: str = typer.Option(..., help="Path to the bedbase config file"), + remove_s3: bool = typer.Option( + True, help="Also delete the underlying S3 object" + ), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + bbagent.snapshot.delete(id, remove_s3=remove_s3) + print(f"Snapshot {id} deleted from the bedbase database") + + +@snapshot_app.command( + "delete-by-checksum", help="Delete snapshot index row(s) by checksum." +) +def snapshot_delete_by_checksum( + checksum: str = typer.Option(..., help="SHA256 checksum of the snapshot file"), + config: str = typer.Option(..., help="Path to the bedbase config file"), + remove_s3: bool = typer.Option( + True, help="Also delete the underlying S3 object(s)" + ), +): + from bbconf import BedBaseAgent + + bbagent = BedBaseAgent(config, init_ml=False) + bbagent.snapshot.delete_by_checksum(checksum, remove_s3=remove_s3) + print(f"Snapshot(s) with checksum {checksum} deleted from the bedbase database") + + +# --------------------------------------------------------------------------- # +# Column plans derived from the bbconf ORM models (no hand-listed columns). +# --------------------------------------------------------------------------- # + + +class ColumnPlan: + """ + An ordered set of SQLAlchemy columns plus the matching Parquet schema. + + Built from whole ORM tables so the SELECT projection and the Parquet schema + are derived from one source and stay tied to the bbconf models. + """ + + def __init__(self, columns: list, from_clause=None): + import pyarrow as pa + + self.columns = columns + self.from_clause = from_clause + self.names = [c.name for c in columns] + self._pa_types = [_pa_type(c) for c in columns] + # Indices of JSON columns, whose values are serialized to a string. + self.json_idx = { + i for i, c in enumerate(columns) if _python_type(c) is dict + } + self.schema = pa.schema(list(zip(self.names, self._pa_types))) + + def select(self): + """A ``SELECT`` of exactly these columns (over the joined FROM, if any).""" + from sqlalchemy import select + + stmt = select(*self.columns) + if self.from_clause is not None: + stmt = stmt.select_from(self.from_clause) + return stmt + + def count_from(self): + """The table these columns' first entry belongs to, for ``count(*)``.""" + return self.columns[0].table + + +def _python_type(column): + try: + return column.type.python_type + except (NotImplementedError, AttributeError): + return str + + +def _pa_type(column): + """Map a SQLAlchemy column's Python type onto a pyarrow type.""" + import pyarrow as pa + + pytype = _python_type(column) + if pytype is bool: + return pa.bool_() + if pytype is int: + return pa.int64() + if pytype is float: + return pa.float64() + if pytype is datetime.datetime: + return pa.timestamp("us", tz="UTC") + if pytype is list: + return pa.list_(pa.string()) + if pytype is dict: + # JSON/JSONB -> serialized string. + return pa.string() + return pa.string() + + +def metadata_plan() -> ColumnPlan: + """``bed`` LEFT JOIN ``bed_metadata`` — all columns of both tables. + + The duplicate join key ``bed_metadata.id`` (equal to ``bed.id``) is dropped. + """ + from bbconf.db_utils import Bed, BedMetadata + + bed_cols = list(Bed.__table__.columns) + meta_cols = [c for c in BedMetadata.__table__.columns if c.name != "id"] + from_clause = Bed.__table__.outerjoin( + BedMetadata.__table__, Bed.__table__.c.id == BedMetadata.__table__.c.id + ) + return ColumnPlan(bed_cols + meta_cols, from_clause=from_clause) + + +def bedsets_plan() -> ColumnPlan: + from bbconf.db_utils import BedSets + + return ColumnPlan(list(BedSets.__table__.columns)) + + +def membership_plan() -> ColumnPlan: + from bbconf.db_utils import BedFileBedSetRelation + + return ColumnPlan(list(BedFileBedSetRelation.__table__.columns)) + + +# --------------------------------------------------------------------------- # +# Pure helpers (no DB, no network). +# --------------------------------------------------------------------------- # + + +def _coerce_row(row, json_idx: set) -> tuple: + """Turn a result row into a value tuple, serializing JSON columns.""" + values = list(row) + for i in json_idx: + if values[i] is not None: + values[i] = json.dumps(values[i]) + return tuple(values) + + +def rows_to_table(rows, schema): + """Convert a batch of row tuples (in schema column order) to a pyarrow Table.""" + import pyarrow as pa + + if rows: + columns = list(zip(*rows)) + else: + columns = [[] for _ in range(len(schema))] + arrays = [ + pa.array(list(col), type=schema.field(i).type) + for i, col in enumerate(columns) + ] + return pa.table(arrays, schema=schema) + + +def check_completeness(rows_written: int, expected_count: int, threshold: float): + """ + Refuse to publish a partial artifact. + + Raises if the fraction of rows missing versus the pre-scan ``count`` exceeds + ``threshold``. Extra rows (concurrent ingest) never trip it. + """ + if expected_count <= 0: + return + shortfall = (expected_count - rows_written) / expected_count + if shortfall > threshold: + raise RuntimeError( + f"Wrote {rows_written} rows but expected ~{expected_count} " + f"(short by {shortfall:.2%} > {threshold:.2%}). " + f"Refusing to publish a partial artifact." + ) + + +def sha256_file(path, chunk: int = 1 << 20) -> str: + """Streaming SHA256 of a file.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for block in iter(lambda: f.read(chunk), b""): + h.update(block) + return h.hexdigest() + + +def file_entry(path) -> dict: + """Manifest entry describing a written file: name, bytes, sha256.""" + path = Path(path) + return { + "name": path.name, + "bytes": path.stat().st_size, + "sha256": sha256_file(path), + } + + +def write_parquet_batches(path, schema, batch_iter) -> int: + """ + Stream batches (lists of row tuples) into a zstd Parquet file. + + Returns the number of rows actually written (the value the manifest and the + integrity gate must use). + """ + import pyarrow.parquet as pq + + rows_written = 0 + writer = pq.ParquetWriter(str(path), schema, compression="zstd") + try: + for batch in batch_iter: + if not batch: + continue + writer.write_table(rows_to_table(batch, schema)) + rows_written += len(batch) + finally: + writer.close() + return rows_written + + +# --------------------------------------------------------------------------- # +# Database streaming (server-side cursors; sequential, unordered scans). +# --------------------------------------------------------------------------- # + + +def stream_plan(conn, plan: ColumnPlan, batch: int): + """Yield lists of coerced row tuples from a streamed, unordered scan.""" + result = conn.execute(plan.select().execution_options(yield_per=batch)) + for partition in result.partitions(): + yield [_coerce_row(row, plan.json_idx) for row in partition] + + +def count_rows(conn, plan: ColumnPlan) -> int: + from sqlalchemy import func, select + + return conn.execute( + select(func.count()).select_from(plan.count_from()) + ).scalar_one() + + +def build_exports( + conn, + out_dir: Path, + date_str: str, + batch_size: int, + fail_threshold: float, +) -> list[dict]: + """ + Write the three Parquet files, gating each against its pre-scan count. + + Returns a list of per-file records: path, name, file_type, record_count, + file_size, checksum. All reads happen in the caller's REPEATABLE READ + transaction. + """ + records = [] + + exports = [ + ("metadata", metadata_plan()), + ("bedsets", bedsets_plan()), + ("bedset_membership", membership_plan()), + ] + + for file_type, plan in exports: + expected = count_rows(conn, plan) + fname = f"bedbase_{file_type}_{date_str}.parquet" + path = out_dir / fname + _LOGGER.info(f"Exporting {file_type}: ~{expected} rows -> {fname}") + + rows_written = write_parquet_batches( + path, plan.schema, stream_plan(conn, plan, batch_size) + ) + check_completeness(rows_written, expected, fail_threshold) + + entry = file_entry(path) + record = { + "path": str(path), + "name": entry["name"], + "file_type": file_type, + "record_count": rows_written, + "file_size": entry["bytes"], + "checksum": entry["sha256"], + } + records.append(record) + _LOGGER.info( + f" wrote {rows_written} rows, {entry['bytes'] / 1e6:.1f} MB, " + f"sha256={entry['sha256'][:12]}…" + ) + + return records + + +def write_manifest( + out_dir: Path, + date_str: str, + records: list[dict], + started: str, + ended: str, + source_db: str, +) -> dict: + manifest = { + "schema_version": SCHEMA_VERSION, + "build_started": started, + "build_ended": ended, + "source_database": source_db, + "files": [ + { + "name": r["name"], + "file_type": r["file_type"], + "rows": r["record_count"], + "bytes": r["file_size"], + "sha256": r["checksum"], + } + for r in records + ], + } + path = out_dir / f"manifest_{date_str}.json" + path.write_text(json.dumps(manifest, indent=2)) + + entry = file_entry(path) + record = { + "path": str(path), + "name": entry["name"], + "file_type": "manifest", + "record_count": None, + "file_size": entry["bytes"], + "checksum": entry["sha256"], + } + return record + + +# --------------------------------------------------------------------------- # +# Orchestration. +# --------------------------------------------------------------------------- # + + +def run_snapshot( + bedbase_config: str, + output_dir: str | Path = "exports", + batch_size: int = DEFAULT_BATCH_SIZE, + fail_threshold: float = 0.01, + publish: bool = False, +) -> None: + """ + Build a bulk metadata snapshot, optionally publishing it. + + Args: + bedbase_config: Path to the bedbase config file. + output_dir: Local directory for the built artifacts. + batch_size: Rows per server-side fetch / Parquet write batch. + fail_threshold: Abort without publishing if this fraction of rows is + missing versus the pre-scan count. + publish: If True, upload the artifacts to S3 and record them in the + ``bed_snapshots`` table (both done inside bbconf). + """ + from bbconf.bbagent import BedBaseAgent + from bbconf.models.base_models import BedSnapshotArtifact + + out_dir = Path(output_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + now = datetime.datetime.now(datetime.timezone.utc) + date_str = now.strftime("%Y_%m_%d") + started = now.isoformat() + + # Connect through bbconf; querying/processing stays here in bedboss. + agent = BedBaseAgent(bedbase_config, init_ml=False) + source_db = agent.config.config.database.database + + # One REPEATABLE READ transaction: counts and the full scans share a + # consistent snapshot. + with agent.config.db_engine.engine.connect().execution_options( + isolation_level="REPEATABLE READ" + ) as conn: + records = build_exports(conn, out_dir, date_str, batch_size, fail_threshold) + + ended = datetime.datetime.now(datetime.timezone.utc).isoformat() + manifest_record = write_manifest( + out_dir, date_str, records, started, ended, source_db + ) + _LOGGER.info(f"Manifest: {manifest_record['path']}") + + if not publish: + _LOGGER.info( + "publish=False; artifacts left on disk, nothing uploaded or recorded." + ) + return + + # Phase 2: hand off to bbconf, which does the S3 upload AND the SQL insert. + artifacts = [ + BedSnapshotArtifact( + path=r["path"], + file_type=r["file_type"], + record_count=r["record_count"], + file_size=r["file_size"], + checksum=r["checksum"], + schema_version=SCHEMA_VERSION, + ) + for r in records + [manifest_record] + ] + result = agent.snapshot.add(artifacts, creation_date=now) + + http_prefix = None + if agent.config.config.access_methods and agent.config.config.access_methods.http: + http_prefix = agent.config.config.access_methods.http.prefix + for row in result.results: + url = f"{http_prefix}{row.file_path}" if http_prefix else row.file_path + _LOGGER.info(f"Published {url}") diff --git a/bedboss/utils.py b/bedboss/utils.py index d1004654..48d79521 100644 --- a/bedboss/utils.py +++ b/bedboss/utils.py @@ -138,7 +138,6 @@ def example_bedbase_config(): "aws_secret_access_key": None, "bucket": "bedbase", }, - "phc": {"namespace": "bedbase", "name": "bedbase", "tag": "latest"}, } diff --git a/docs/changelog.md b/docs/changelog.md index 64e1afe4..2376ab52 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -2,6 +2,14 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html) and [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) format. +# [0.12.0] - 2026-08-18 +## Changes: +- All references to `upload_pephub` have been removed from function signatures, docstrings, and internal logic across the codebase, including in `bedboss.py`, `bedbuncher.py`, `bbuploader/main.py`, `bedboss_hpc.py`, and the CLI (`cli.py`). This means users can no longer upload to PEPHub using any part of the pipeline. +- Removed unused logging setup from `bedboss/bbuploader/__init__.py` and added `logmuse` import to `cli.py` for consistent logging across the CLI. +- Fixed a bug in `_read_bed_file` in `bedclassifier.py` to only drop columns that are entirely NA, instead of any column with any NA. +- Added import for `snapshot_app` in `cli.py` from `bedboss.scripts.snapshot`. +- Added Analysis files functionality. + # [0.11.1] - 2026-04-22 ## Changed: - Migrated to new yacman version 1.0.0 diff --git a/pyproject.toml b/pyproject.toml index bdde93f1..1f1810a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "bedboss" -version = "0.11.1" +version = "0.12.0" description = "Pipelines for genomic region file to produce bed files, and its statistics" readme = "README.md" license = "BSD-2-Clause" @@ -24,7 +24,7 @@ classifiers = [ "Topic :: Scientific/Engineering :: Bio-Informatics", ] dependencies = [ - "logmuse>=0.2.7", + "logmuse>=0.3.1", "coloredlogs>=15.0.1", "eido>=0.2.5", "peppy>=0.40.7", @@ -32,9 +32,10 @@ dependencies = [ "yacman>=1.0.0", "requests>=2.28.2", "piper>=0.14.3", - "bbconf>=0.14.12", + "bbconf>=0.15.0", "refgenconf>=0.13.1", "pandas>=2.0.0", + "pyarrow>=15.0.0", "ubiquerg>=0.6.2", "bedms>=0.1.0", "pephubclient>=0.5.0", diff --git a/scripts/bedclassifier_tuning/incorrect_file/a0169b30e33ceaaf1614a83f56094727.bed.gz b/scripts/bedclassifier_tuning/incorrect_file/a0169b30e33ceaaf1614a83f56094727.bed.gz new file mode 100644 index 00000000..d6a9c676 Binary files /dev/null and b/scripts/bedclassifier_tuning/incorrect_file/a0169b30e33ceaaf1614a83f56094727.bed.gz differ diff --git a/scripts/metadata_update/update_meta.py b/scripts/metadata_update/update_meta.py index 44b6e5e0..c249031a 100644 --- a/scripts/metadata_update/update_meta.py +++ b/scripts/metadata_update/update_meta.py @@ -17,7 +17,6 @@ def update_description(bed_id: str, new_description: str): bbconf.bed.update( identifier=bed_id, metadata={"description": new_description}, - upload_pephub=False, upload_s3=False, upload_qdrant=False, ) @@ -28,7 +27,6 @@ def update_assay(bed_id: str, new_assay: str): bbconf.bed.update( identifier=bed_id, metadata={"assay": new_assay}, - upload_pephub=False, upload_s3=False, upload_qdrant=False, ) diff --git a/scripts/profiling/prof.py b/scripts/profiling/prof.py index a756b5f6..833e851a 100644 --- a/scripts/profiling/prof.py +++ b/scripts/profiling/prof.py @@ -16,7 +16,6 @@ def runn(): input_file="/home/bnt4me/Downloads/test_chroms.bed.gz", input_type="bed", force_overwrite=True, - upload_pephub=True, upload_s3=True, upload_qdrant=True, name="test", diff --git a/test/test_bedclassifier.py b/test/test_bedclassifier.py index 2982e85c..ab0a936e 100644 --- a/test/test_bedclassifier.py +++ b/test/test_bedclassifier.py @@ -32,6 +32,9 @@ BED_7_03 = f"{SIMPLE_EXAMPLES_DIR}/test_bed_7_03.bed" BED_GAPPED_PEAK = f"{SIMPLE_EXAMPLES_DIR}/example.gappedPeak.gz" BED_GAPPED_PEAK_RS = f"{SIMPLE_EXAMPLES_DIR}/example_2.gappedPeak.gz" +# 5-column bed whose 4th (float) column has a single NaN value; regression test +# for the sparse-NaN column drop that previously misclassified this as bed4+0. +BED_4_PLUS_1_SPARSE_NAN = f"{SIMPLE_EXAMPLES_DIR}/test_bed_4plus1_sparse_nan.bed.gz" class TestBedClassifier: @@ -225,6 +228,15 @@ def test_get_bed_classification( non_compliant_columns=3, ), ), + ( + BED_4_PLUS_1_SPARSE_NAN, + BedClassificationOutput( + bed_compliance="bed4+1", + data_format=DATA_FORMAT.BED_LIKE, + compliant_columns=4, + non_compliant_columns=1, + ), + ), ], ) def test_get_bed_classifications(self, values):