mandos.entry.tools.fillers.CompoundIdFiller.fill()

Complexity

Conditions

13

Size

Total Lines	39
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from __future__ import annotations
    def fill(self, df: IdMatchDf) -> IdMatchDf:

Argument name "df" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Missing function or method docstring

        df = self._prep(df)
        logger.info(f"Processing {len(df)} input compounds")
        fill = []
        for i, row in enumerate(df.itertuples()):
            if i % 200 == 0 and i > 0:
                logger.notice(f"Processed {i:,} / {len(df):,}")
            elif i % 20 == 0 and i > 0:
                logger.info(f"Processed {i:,} / {len(df):,}")
            with logger.contextualize(line=i):
                proc = self._process(
                    compound_id=look(row, "compound_id"),
                    library=look(row, "library"),
                    inchi=look(row, "origin_inchi"),
                    inchikey=look(row, "origin_inchikey"),
                    pubchem_id=look(row, "origin_pubchem_id"),
                    chembl_id=look(row, "origin_chembl_id"),
                )
            fill.append(proc)
        for c in FILL_IDS:

Variable name "c" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            df[c] = [r[c] for r in fill]
        duplicate_cols = []
        for c in FILL_IDS:

Variable name "c" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            if c in df.columns and "origin_" + c in df.columns:
                if df[c].values.tolist() == df["origin_" + c].values.tolist():
                    duplicate_cols.append("origin_" + c)
        logger.notice(f"Done — filled {len(df):,} rows")
        if len(duplicate_cols) > 0:
            df = df.drop_cols(duplicate_cols)
            logger.notice(f"Dropped duplicated columns {', '.join(duplicate_cols)}")
        order = [o for o in PUT_FIRST if o in df.columns]
        order += [c for c in df.columns if c not in PUT_FIRST and c not in PUT_LAST]
        order += [o for o in PUT_LAST if o in df.columns]
        df = df.cfirst(order)
        have_chembl = len(df) - len(df[df["chembl_id"].isnull()]["chembl_id"].tolist())
        have_pubchem = len(df) - len(df[df["pubchem_id"].isnull()]["pubchem_id"].tolist())
        logger.notice(f"{have_chembl:,}/{len(df):,} have ChEMBL IDs")
        logger.notice(f"{have_pubchem:,}/{len(df):,} have PubChem IDs")
        return df