<commit_before>use VRPipe::Base;

class VRPipe::Steps::fastq_import with VRPipe::StepRole {
    use File::Fetch;
    use Net::FTP::Robust;
    
    method options_definition {
        return { };
    }
    method inputs_definition {
        return { fastq_files => VRPipe::StepIODefinition->get(type => 'fq',
                                                              description => 'fastq files',
                                                              max_files => -1,
                                                              metadata => {remote_path => 'the complete remote location of the file',
                                                                           expected_md5 => 'the md5 checksum the file is supposed to have',
                                                                           optional => ['remote_path', 'expected_md5']},
                                                              check_existence => 0) };
    }
    method body_sub {
        return sub {
            my $self = shift;
            
            my $req = $self->new_requirements(memory => 500, time => 1);
            foreach my $fq_file (@{$self->inputs->{fastq_files}}) {
                my $ifile = $fq_file->path;
                
                my $meta = $fq_file->metadata;
                my $remote_path = $meta->{remote_path};
                if ($remote_path && $remote_path ne $ifile) {
                    # download to the path of our input file, which doesn't
                    # exist yet
                    $self->output_file(output_key => 'local_fastq_files', output_dir => $ifile->dir, basename => $ifile->basename, type => 'fq', metadata => $fq_file->metadata);
                    $self->dispatch_vrpipecode(qq[use VRPipe::Steps::fastq_import; VRPipe::Steps::fastq_import->download_fastq(source => q[$remote_path], dest => q[$ifile]);],
                                               $req);
                }
                else {
                    # symlink our existing input file to the pipeline output dir
                    # so that if this step is restarted, we won't delete our
                    # input file
                    my $ofile = $self->output_file(output_key => 'local_fastq_files', basename => $ifile->basename, type => 'fq', metadata => $fq_file->metadata);
                    $fq_file->symlink($ofile);
                }
            }
        };
    }
    method outputs_definition {
        return { local_fastq_files => VRPipe::StepIODefinition->get(type => 'fq',
                                                                    description => 'a fastq file on a local disc',
                                                                    max_files => -1) };
    }
    method post_process_sub {
        return sub { return 1; };
    }
    method description {
        return "If fastq files in the datasource are on an external ftp site, downloads them to local disc";
    }
    method max_simultaneous {
        return 0; # meaning unlimited
    }
    
    method download_fastq (ClassName|Object $self: Str :$source!, Str|File :$dest!) {
        my $fq_file = VRPipe::File->get(path => $dest);
        my $meta = $fq_file->meta;
        my $expected_md5 = $meta->{expected_md5};
        my $local_dir = $fq_file->dir;
        my $out_file = $fq_file->path;
        
        $fq_file->disconnect;
        <commit_msg>limit to 50 max_simultaneous to avoid overloading ftp server<commit_after>use VRPipe::Base;

class VRPipe::Steps::fastq_import with VRPipe::StepRole {
    use File::Fetch;
    use Net::FTP::Robust;
    
    method options_definition {
        return { };
    }
    method inputs_definition {
        return { fastq_files => VRPipe::StepIODefinition->get(type => 'fq',
                                                              description => 'fastq files',
                                                              max_files => -1,
                                                              metadata => {remote_path => 'the complete remote location of the file',
                                                                           expected_md5 => 'the md5 checksum the file is supposed to have',
                                                                           optional => ['remote_path', 'expected_md5']},
                                                              check_existence => 0) };
    }
    method body_sub {
        return sub {
            my $self = shift;
            
            my $req = $self->new_requirements(memory => 500, time => 1);
            foreach my $fq_file (@{$self->inputs->{fastq_files}}) {
                my $ifile = $fq_file->path;
                
                my $meta = $fq_file->metadata;
                my $remote_path = $meta->{remote_path};
                if ($remote_path && $remote_path ne $ifile) {
                    # download to the path of our input file, which doesn't
                    # exist yet
                    $self->output_file(output_key => 'local_fastq_files', output_dir => $ifile->dir, basename => $ifile->basename, type => 'fq', metadata => $fq_file->metadata);
                    $self->dispatch_vrpipecode(qq[use VRPipe::Steps::fastq_import; VRPipe::Steps::fastq_import->download_fastq(source => q[$remote_path], dest => q[$ifile]);],
                                               $req);
                }
                else {
                    # symlink our existing input file to the pipeline output dir
                    # so that if this step is restarted, we won't delete our
                    # input file
                    my $ofile = $self->output_file(output_key => 'local_fastq_files', basename => $ifile->basename, type => 'fq', metadata => $fq_file->metadata);
                    $fq_file->symlink($ofile);
                }
            }
        };
    }
    method outputs_definition {
        return { local_fastq_files => VRPipe::StepIODefinition->get(type => 'fq',
                                                                    description => 'a fastq file on a local disc',
                                                                    max_files => -1) };
    }
    method post_process_sub {
        return sub { return 1; };
    }
    method description {
        return "If fastq files in the datasource are on an external ftp site, downloads them to local disc";
    }
    method max_simultaneous {
        return 50;
    }
    
    method download_fastq (ClassName|Object $self: Str :$source!, Str|File :$dest!) {
        my $fq_file = VRPipe::File->get(path => $dest);
        my $meta = $fq_file->meta;
        my $expected_md5 = $meta->{expected_md5};
        my $local_dir = $fq_file->dir;
        my $out_file = $fq_file->path;
        
        $fq_file->disconnect;
        
{'fraction_non_alphanumeric': 0.10430076067875951, 'fraction_numerical': 0.004534815681685196, 'mean_word_length': 1.3082376772451048, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6523066', 'n_tokens_mistral': 1752, 'n_tokens_neox': 1699, 'n_words': 438}