Source code for AMDirT.validate.application

from typing import Iterable, AnyStr, Union
from AMDirT.validate.domain import DatasetValidator, DFError
from AMDirT.core import get_json_path
from AMDirT.core.diff import get_sample_diff
from AMDirT.core.ena import ENAPortalAPI
from rich.progress import track
from pathlib import Path
import pandas as pd
import json

[docs] class AMDirValidator(DatasetValidator): """Validator Class for AncientMetagenomeDir datasets""" def check_duplicate_dois(self) -> bool: project_dois = self.dataset.groupby("project_name")[ "publication_doi" ].unique() doi_unique = self.dataset.groupby("project_name")[ "publication_doi" ].nunique() err_cnt = 0 for project in doi_unique.index: if doi_unique[project] > 1: err_cnt += 1 self.add_error( DFError( error="Duplicated DOI Error", source=project_dois[project], column="publication_doi", row="", message=f"Duplicate DOI for {project} project. Make sure each project has a single DOI", ) ) if err_cnt > 0: return False return True
[docs] def check_multi_values( self, column_names: Iterable[str] = ["archive_accession"] ) -> bool: """Check for duplicates entries in multi values column Args: column_names (Iterable[str], optional): List of multi values columns to check for duplications. Defaults to ["archive_accession"]. """ err_cnt = 0 for column in column_names: row = 0 for archives in self.dataset[column]: archives = archives.split(",") if len(set(archives)) != len(archives): self.add_error( DFError( error="Duplicates in multi values column", source=archives, column=column, row=row, message=f"Duplicates in multi values column {column}. Make sure each value in combination is unique", ) ) err_cnt += 1 row += 1 if err_cnt > 0: return False else: return True
[docs] def check_sample_accession(self, remote: Union[AnyStr, None] = None) -> bool: """Check that sample accession are valid Args: remote (AnyStr | None, optional): Remote to check against. Defaults to None. """ if not remote: with open(get_json_path()) as f: tables = json.load(f) samples = tables["samples"] for table in samples: if self.dataset_name == Path(samples[table]).name: remote = samples[table] if remote is None: raise SystemExit( f"No remote found for {self.dataset} dataset, please provide one" ) remote_samples = DatasetValidator(schema=self.schema_path, dataset=remote) df_change = pd.concat([remote_samples.dataset, self.dataset]).drop_duplicates( keep=False ) df_change.drop_duplicates( inplace=True, keep="last", subset=list(df_change.columns)[:-1] ) is_ok = True print(df_change) if df_change.shape[0] > 0: e = ENAPortalAPI() change_dict = {} for i in df_change.index: try: supported_archive = df_change.loc[i, "archive"] in ["SRA", "ENA"] except ValueError as e: print(e) print(df_change.loc[i, :]) supported_archive = False continue if supported_archive: samples = df_change.loc[i, "archive_accession"].split(",") project = df_change.loc[i, "archive_project"] if project not in change_dict: change_dict[project] = {"index": i, "sample": samples} else: change_dict[project]["sample"].extend(samples) else: continue for project in track(change_dict, description="Checking ENA/SRA accessions..."): json_result = e.query( accession=project, result_type="read_experiment", fields=["secondary_sample_accession"], ) ena_samples = [] for i in json_result: ena_samples.append(i["secondary_sample_accession"]) for sample in change_dict[project]["sample"]: if sample not in ena_samples: row = df_change.query( f"archive_accession.str.contains('{sample}') and archive_project.str.contains('{project}')" ).index[0] self.add_error( DFError( error="Invalid sample accession", source=sample, column="archive_accession", row=row, message=f"Sample accession {sample} is not a valid ENA/SRA sample accession for the project {project}", ) ) is_ok = False return is_ok