slice_data.ts

import {Configs, ProjectConfigs} from "./types";
import * as fs from 'fs'
import * as zlib from 'zlib';
import * as path from 'path';
import {glob} from 'glob'
import * as csv from 'csv';
import {SliceConstants} from "./slice_constants";

const fsPromises = fs.promises;
const readline = require('readline');
const csvToJson = require('convert-csv-to-json');
const _  = require('underscore');
export class SliceData {
    private projectConfigs: ProjectConfigs;

    public constructor(configs: Configs) {
        this.projectConfigs = SliceConstants.prepareConstants(configs);
    }

    public getProjectConfigs() {
        return this.projectConfigs;
    }

    public static async processLineByLine(projectConfigs: ProjectConfigs, file: string, list_to_include: string[]) {

        const outputFileName = SliceConstants.getOutputFileName(file, projectConfigs);
        fs.rmSync(outputFileName, {force: true});
        const fileStream = fs.createReadStream(file);
        const rl = readline.createInterface({
            input: fileStream,
            crlfDelay: Infinity
        });
        try {
            // Note: we use the crlfDelay option to recognize all instances of CR LF
            // ('\r\n') in input.txt as a single line break.
            let t = 0;
            let header = []
            for await (const line of rl) {
                if (t === 0) {
                    if (line.includes(",")) {
                        header = line.split(",")
                    } else {
                        header = line.split("\t")
                    }

                    fs.appendFileSync(outputFileName, line + "\n");
                } else {
                    // Each line in input.txt will be successively available here as `line`.
                    const found = list_to_include.filter(item => line.includes(item));
                    if (found.length > 0) {
                        fs.appendFileSync(outputFileName, line + "\n");
                    }
                }
                ++t;
            }
            //const json = csvToJson.fieldDelimiter(',').getJsonFromCsv(outputFileName);
            //const keys =  Object.keys(json);
            //console.log(header)

            //console.log(json);
        } finally {
            console.log("Finished Processing", file)
            rl.close();
            fileStream.close();
        }
    };

    public static async copySingleFile(projectConfigs: ProjectConfigs, source: string, destination: string, file: string) {
        let outputFileName = file.replace(source, destination);
        outputFileName = outputFileName.replace(projectConfigs.OLD_PROJECT_PREFIX, projectConfigs.NEW_PROJECT_PREFIX);
        fs.rmSync(outputFileName, {force: true})
        await fsPromises.copyFile(file, outputFileName)
    };

    public static async copyAllFiles(projectConfigs: ProjectConfigs, source: string, destination: string, csvFiles: string[]) {
        for await (let csvFile of csvFiles) {
            await SliceData.copySingleFile(projectConfigs, source, destination, csvFile)
        }
    };

    public static async processAllByRows(projectConfigs: ProjectConfigs, csvFiles: string[], elements_to_include: string[]) {
        for (let csvFile of csvFiles) {
            await SliceData.processLineByLine(projectConfigs, csvFile, elements_to_include);
        }
    };

    public static async processAllByColumn(projectConfigs: ProjectConfigs, csvFiles: string[]) {
        for (let csvFile of csvFiles) {
            await SliceData.processByColumn(projectConfigs, csvFile);
        }
    };

    public static async cleanUp(dir: string) {
        try {
             if (fs.existsSync(dir)) {
                fs.rmSync(dir, {recursive: true});
            }
        } catch (err) {
            console.log("error", err)
        }
    };

    public static async processByColumn(projectConfigs: ProjectConfigs, file: string) {
        let transformedPath = file.replace(projectConfigs.SOURCE_DATA_DIR, projectConfigs.DESTINATION_DATA_DIR);
        transformedPath = transformedPath.replace(projectConfigs.OLD_PROJECT_PREFIX, projectConfigs.NEW_PROJECT_PREFIX);
        fs.rmSync(transformedPath, {force: true})
        const rs = fs.createReadStream(file);
        const ws = fs.createWriteStream(transformedPath);

        rs.pipe(csv.parse({delimiter: ',', columns: true}))
            .pipe(csv.transform((input) => {
                const keys = Object.keys(input);
                for (let key of keys) {
                    if (!SliceConstants.matcher(projectConfigs.PERT_INAMES, key)) {
                        delete input[key]
                    }
                }
                return input;
            }))
            .pipe(csv.stringify({header: true}))
            .pipe(ws)
            .on('finish', () => {
                console.log('finished processing....', file);
                ws.end()
            })
            .on('error', (err) => {
                console.log('error.....', err);
                ws.end()
            });
    };

    public static async replaceProjectNames(project:string, srcFile: string, destFile: string,delimiter:string) {
        fs.rmSync(destFile, {force: true})

        return new Promise((resolve, reject) => {
            const rs = fs.createReadStream(srcFile);
            const ws = fs.createWriteStream(destFile);
            rs.pipe(csv.parse({delimiter: delimiter, columns: true}))
                .pipe(csv.transform((input) => {
                    if(input["x_project_id"]){
                        input["x_project_id"] = project
                    }
                    return input;
                }))
                .pipe(csv.stringify({header: true}))
                .pipe(ws)
                .on('finish', () => {
                    console.log('finished processing....', srcFile);
                    ws.end()
                    resolve("data")
                })
                .on('error', (err) => {
                    console.log('error.....', err);
                    ws.end()
                    reject(err)
                });
        });
    };

    public static async makeDataDirectory(project: string) {
        const destDir = "etl/" + project.toLowerCase() + "/" + project.toUpperCase() + "/";
        const promises = [];
        fs.mkdirSync(destDir + '/data/', { recursive: true });
        for (let name of SliceConstants.PROJECT_DATA_FILES) {
            console.log("name",name);
            let gl = destDir + "*/*/" + name;
            console.log(gl);
            const filesToCollate = await glob(gl);
            console.log("filesToCollate",filesToCollate)
            let outFile = destDir + '/data/' + `${project.toUpperCase()}_${name}`;
            outFile = outFile.replace("_*.csv", ".csv");
            promises.push(SliceData.concatFiles(filesToCollate,outFile, ','));
        }

        await Promise.all(promises);
    }


    public static async gzipFiles(data_dir: string) {
        const files = await glob(data_dir + "*.*", { ignore: data_dir + "*.gz" });

        const promises = [];
        for (let file of files) {
            promises.push(new Promise((resolve, reject) => {
                const gzip = zlib.createGzip();
                const inp = fs.createReadStream(file);
                const out = fs.createWriteStream(file + '.gz');
                out.on('finish', () => {
                    console.log('Successfully compressed', file);
                    resolve('Successfully compressed');
                    fs.unlink(file, (err) => {
                        if (err) {
                            console.error('Error deleting file:', err);
                            reject(err)
                        } else {
                            resolve('Deleted file')
                        }
                    });
                })
                out.on('error', (err) => {
                    console.error('Error compressing file:', err);
                    reject(err);
                });
                inp.pipe(gzip).pipe(out);
            }));
        }
        return promises;
    }

    public static async concatFiles(filePaths: string[], destFile: string,delimiter:string) {
        fs.rmSync(destFile, {force: true})
        if (filePaths.length === 0) {
            console.log('No files to concatenate');
            return;
        }

        return new Promise((resolve, reject) => {
            const parser = csv.parse({ delimiter: delimiter, columns: true });
            const stringifier = csv.stringify({ header: true });
            const outputStream = fs.createWriteStream(destFile);

            outputStream.on('finish', () => {
                console.log('All files have been concatenated successfully');
                resolve('Concatenation completed');
            });

            outputStream.on('error', (err) => {
                console.error('Error writing to the output file:', err);
                reject(err);
            });

            stringifier.on('error', (err) => {
                console.error('Error in CSV stringification:', err);
                reject(err);
            });

            stringifier.pipe(outputStream);

            let fileCount = filePaths.length;
            filePaths.forEach((filePath, index) => {
                fs.createReadStream(filePath)
                    .pipe(csv.parse({ delimiter: delimiter, columns: true }))
                    .on('data', (data) => {
                        stringifier.write(data);
                    })
                    .on('end', () => {
                        console.log(`Finished processing file: ${filePath}`);
                        if (--fileCount === 0) {
                            stringifier.end();  // Close the stringifier stream after the last file
                        }
                    })
                    .on('error', (err) => {
                        console.error('Error reading or parsing file:', err);
                        reject(err);
                    });
            });
        });

    }

    public static async replaceProjectNames_old(project:string, srcFile: string, destFile: string,delimiter:string) {
        fs.rmSync(destFile, {force: true})
        const rs = fs.createReadStream(srcFile);
        const ws = fs.createWriteStream(destFile);

        rs.pipe(csv.parse({delimiter: delimiter, columns: true}))
            .pipe(csv.transform((input) => {
                if(input["x_project_id"]){
                    input["x_project_id"] = project
                }
                //console.log(input)
                return input;
            }))
            .pipe(csv.stringify({header: true}))
            .pipe(ws)
            .on('finish', () => {
                console.log('finished processing....', srcFile);
                ws.end()
            })
            .on('error', (err) => {
                console.log('error.....', err);
                ws.end()
            });
    };
    public static async renameFileAndAddProject(projectConfigs: ProjectConfigs, file: string, delimiter:string) {
        let json = csvToJson.fieldDelimiter(delimiter).getJsonFromCsv(file);
        let col_id = 0;
        let row_id = 0
        if(file.includes("_LEVEL3_")){
            col_id = _.pluck(json,"profile_id").length;
            row_id = _.uniq(_.pluck(json,'ccle_name')).length;
            console.log("col_id",col_id)
            console.log("row_id",row_id)
        }

        const transformedPath = SliceConstants.getOutputFileName(file,projectConfigs)
        console.log(transformedPath)
        fs.rmSync(transformedPath, {force: true})
        const rs = fs.createReadStream(file);
        const ws = fs.createWriteStream(transformedPath);



        rs.pipe(csv.parse({delimiter: delimiter, columns: true}))
            .pipe(csv.transform((input) => {
                if(input["x_project_id"]){
                    input["x_project_id"] = projectConfigs.PROJECT
                }
               //console.log(input)
                return input;
            }))
            .pipe(csv.stringify({header: true}))
            .pipe(ws)
            .on('finish', () => {
                console.log('finished processing....', file);
                ws.end()
            })
            .on('error', (err) => {
                console.log('error.....', err);
                ws.end()
            });
    };

    public static async createFile(file: string, content: string) {
        return await fsPromises.writeFile(file, content);
    };

    public static async copyDirectory(source: string, destination: string) {
        // Create the destination directory if it doesn't exist
        if (!fs.existsSync(destination)) {
            fs.mkdirSync(destination, {recursive: true});
        }

        // Get a list of all files and subdirectories in the source directory
        const files = fs.readdirSync(source);

        // Iterate over each file/directory
        files.forEach((file) => {
            const sourcePath = path.join(source, file);
            const destPath = path.join(destination, file);

            // Check if the item is a directory
            if (fs.statSync(sourcePath).isDirectory()) {
                // Recursively copy the subdirectory
                SliceData.copyDirectory(sourcePath, destPath);
            } else {
                // Copy the file
                fs.copyFileSync(sourcePath, destPath);
            }
        });
    }

    public static async copyPertDirectory(projectConfigs: ProjectConfigs) {
        for (let pert_plate of projectConfigs.PERT_PLATES) {
            for (let pert_id of projectConfigs.PERT_IDS) {
                //check if the source exists
                const srcFolder = projectConfigs.SOURCE_DIR + "/" + pert_plate + "/" + pert_id;
                const destFolder = projectConfigs.DESTINATION_DIR + "/" + pert_plate + "/" + pert_id;
                if (fs.existsSync(destFolder)) {
                    fs.rmSync(destFolder, {recursive: true});
                }
                if (fs.existsSync(srcFolder)) {
                    SliceData.copyDirectory(srcFolder, destFolder);
                }
            }
        }
    };

    public static async zipFiles(projectConfigs: ProjectConfigs) {
        const zipper = require('zip-local');
        const zipFileName = projectConfigs.DESTINATION_DATA_DIR + "/" + (projectConfigs.NEW_PROJECT_PREFIX.slice(0, -1)) + ".zip"
        const filesToZipFilesGlob = [];
        for (let f of projectConfigs.FILES_TO_ZIP) {
            filesToZipFilesGlob.push(projectConfigs.DESTINATION_DATA_DIR + "/*" + f)
        }
        const zipFiles = await glob(filesToZipFilesGlob)
        const folderToZip = (projectConfigs.DESTINATION_DATA_DIR + "/" +
            projectConfigs.NEW_PROJECT_PREFIX).slice(0, -1);
        if (fs.existsSync(folderToZip)) {
            fs.rmSync(folderToZip, {recursive: true, force: true});
        }
        fs.rmSync(zipFileName, {force: true})
        fs.mkdirSync(folderToZip, {recursive: true});
        await this.copyAllFiles(projectConfigs, projectConfigs.DESTINATION_DATA_DIR, folderToZip, zipFiles)
        //create a folder and then zip it
        zipper.sync.zip(folderToZip + "/").compress().save(zipFileName);
        console.log("Zip file", zipFileName)
    };

    public static async createDataFolder(projectConfigs: ProjectConfigs) {
        //go through the destination project folder
        //glob all compound directory files



        //for each type of file vertically concatenate them and write them to the dest-folder/data/
    }


}

//sync to s3

// (async () => {
//
//     const csvColumnFilesGlob = [];
//     const csvRowFilesGlob = [];
//     const csvPlateFilesGlob = [];
//     const csvCopyFilesGlob = [];
//
//     for (let f of this.projectConfigs.COLUMN_FILES_POST_FIX) {
//         csvColumnFilesGlob.push(this.projectConfigs.SOURCE_DATA_DIR + "/*" + f)
//     }
//     for (let f of this.projectConfigs.ROW_FILES_POST_FIX) {
//         csvRowFilesGlob.push(this.projectConfigs.SOURCE_DATA_DIR + "/*" + f)
//     }
//     for (let f of this.projectConfigs.PLATE_FILES_POST_FIX) {
//         csvPlateFilesGlob.push(this.projectConfigs.SOURCE_DATA_DIR + "/*" + f)
//     }
//     for (let f of this.projectConfigs.COPY_FILES_POST_FIX) {
//         csvCopyFilesGlob.push(this.projectConfigs.SOURCE_DATA_DIR + "/*" + f)
//     }
//     //create the destination directory if it does not already exists
//     if (!fs.existsSync(this.projectConfigs.DESTINATION_DATA_DIR)) {
//         fs.mkdirSync(this.projectConfigs.DESTINATION_DATA_DIR, {recursive: true});
//     }
//
//     let promises = [
//         glob(csvRowFilesGlob),
//         glob(csvColumnFilesGlob),
//         glob(csvPlateFilesGlob),
//         glob(csvCopyFilesGlob)
//     ];
//     const ps = await Promise.all(promises);
//
//     const rowFiles = ps[0];
//     const columnFiles = ps[1];
//     const plateFiles = ps[2];
//     const copyFiles = ps[3];
//
//     promises = [
//         processAllByRows(rowFiles, this.projectConfigs.PERT_IDS),
//         processAllByRows(plateFiles, this.projectConfigs.PERT_PLATES),
//         processAllByColumn(columnFiles),
//         copyAllFiles(this.projectConfigs.SOURCE_DATA_DIR, this.projectConfigs.DESTINATION_DATA_DIR, copyFiles),
//         copyHomeFiles(),
//         copyPertDirectory()
//     ];
//     await Promise.all(promises);
//     await zipFiles();
//     await cleanUp();
//     console.log("Done");
// })();