Source code for tools.tconversion

"""Tools to convert a CSV file into another format."""
import typing
import os
import os.path as op
import glob
from definitions import dconversion




def get_all_table_paths(indir: str, ext: str = ".csv") -> typing.List[str]:
    """This function reads all the CSV-like files that are in a folder."""
    inpaths = glob.glob(op.join(indir, "*" + ext))
    assert len(inpaths) > 0, "No CSV files were found in " + str(indir)
    return inpaths





def convert_all_table_paths(
    indir: str,
    outformat: str,
    outdir: typing.Optional[str] = None,
    outcompression: typing.Optional[str] = None,
    informat: str = "csv",
    incompression: typing.Optional[str] = None,
    keep_original: bool = True,
    verbose: bool = True,
):
    """Convert all the CSV-like files in a given directory into another format
    and/or compression.

    Args:
        indir: input directory where the files to convert are
        outformat: output format
        outdir: output directory. If not given, it is ``indir``
        outcompression: compression of the output files
        informat: format of the input files
        incompression: compression of the input files. Only use to figure out
            the extension
        keep_original: whether to keep the original files. If set to ``False``,
            the files are removed
        verbose: whether to print some information
    """
    assert not (outformat == informat and outcompression == incompression)
    if outdir is None:
        outdir = indir

    inext = dconversion.get_extension(format=informat, compression=incompression)
    outext = dconversion.get_extension(format=outformat, compression=outcompression)
    inpaths = get_all_table_paths(indir, ext=inext)

    # Functions to read and write
    read = dconversion.get_io_function(action="r", format=informat)
    write = dconversion.get_io_function(action="w", format=outformat)

    os.makedirs(outdir, exist_ok=True)

    for inpath in inpaths:
        # Output path
        basename = op.basename(inpath)
        filename = basename[:-len(inext)]
        outpath = op.join(outdir, filename + outext)

        if verbose:
            print("Convert", inpath, "into", outpath)

        # read
        table = read(inpath)
        # write
        write(table, outpath, compression=outcompression)

        if not keep_original:
            if verbose:
                print("Removing original file...")
                os.remove(inpath)