mandos.entry.misc_commands.MiscCommands.concat()

Complexity

Conditions

8

Size

Total Lines	56
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""
    @staticmethod
    def concat(

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

This function exceeds the maximum number of variables (16/15).

        path: Path = Arg.in_dir(

Wrong hanging indentation before block (add 4 spaces).

            rf"""
            Directory containing results from a mandos search.

            {DfCliHelp.list_formats().get_short_text()}
            """
        ),
        to: Optional[Path] = Ca.out_annotations_file,

Wrong hanging indentation before block (add 4 spaces).

        replace: bool = Ca.replace,

Wrong hanging indentation before block (add 4 spaces).

        log: Optional[Path] = CommonArgs.log,

Wrong hanging indentation before block (add 4 spaces).

        stderr: str = CommonArgs.stderr,

Wrong hanging indentation before block (add 4 spaces).

    ) -> None:
        r"""
        Concatenate Mandos annotation files into one.

        Note that ``:search`` automatically performs this;
        this is needed only if you want to combine results from multiple independent searches.
        """
        LOG_SETUP(log, stderr)
        files_ = []
        for file in path.iterdir():
            ff = FileFormat.from_path_or_none(file)

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

            if ff not in [None, FileFormat.json, FileFormat.toml] and not ff.name.endswith(
                ".doc.tsv"

Wrong hanging indentation before block (add 4 spaces).

            ):
                files_.append(file)
        logger.info(f"Looking under {path} (NOT recursive)")
        logger.debug(f"Found {len(files_)} potential input files: {[f.name for f in files_]}")
        files, dfs = [], []
        for file in files_:
            try:
                df: HitDf = HitDf.read_file(file, attrs=True)

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

            except InvalidDfError:
                logger.warning(f"Skipping {file} {Chars.en} not a valid hit list")
                logger.debug(f"Error reading {file}", exc_info=True)
                continue
            df = df.set_attrs({file.name: df.attrs})

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

            dfs.append(df)
            files.append(file)
        names = [CompressionFormat.strip_suffix(f).name for f in files]
        default = path / (",".join(names) + DEF_SUFFIX)
        to = EntryUtils.adjust_filename(to, default, replace)
        logger.notice(f"Concatenated {len(files)} files")
        for f_, df_ in zip(files, dfs):

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

            logger.success(f"Included: {f_.name} with {len(df_)} rows")
        df = HitDf.of(pd.concat(dfs))

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

        counts = {k: v for k, v in df.group_by("universal_id").count().to_dict() if v > 0}
        if len(counts) > 0:
            logger.error(
                f"There are {len(counts)} universal IDs with duplicates!"
                + f": {StringTools.join_kv(counts)}"
            )
        logger.notice(f"Wrote {len(df)} rows to {to}")
        df.write_file(to)