<commit_before>package components

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/scipipe/scipipe"
)

// FileSplitter is a process that will split a file into multiple files, each
// with LinesPerSplit number of lines per file
type FileSplitter struct {
	scipipe.BaseProcess
	LinesPerSplit int
}

// NewFileSplitter returns an initialized FileSplitter process that will split a
// file into multiple files, each with linesPerSplit number of lines per file
func NewFileSplitter(wf *scipipe.Workflow, name string, linesPerSplit int) *FileSplitter {
	p := &FileSplitter{
		BaseProcess:   scipipe.NewBaseProcess(wf, name),
		LinesPerSplit: linesPerSplit,
	}
	p.InitInPort(p, "file")
	p.InitOutPort(p, "split_file")
	wf.AddProc(p)
	return p
}

// InFile returns the port for the input file
func (p *FileSplitter) InFile() *scipipe.InPort { return p.InPort("file") }

// OutSplitFile returns the resulting split (part) files generated0
func (p *FileSplitter) OutSplitFile() *scipipe.OutPort { return p.OutPort("split_file") }

// Run runs the FileSplitter process
func (p *FileSplitter) Run() {
	defer p.CloseAllOutPorts()

	fileReader := NewFileReader(p.Workflow(), p.Name()+"_filereader_"+getRandString(7))
	pop := scipipe.NewOutParamPort(p.Name() + "_temp_filepath_feeder")
	pop.SetProcess(p)
	fileReader.InFilePath().From(pop)

	for ft := range p.InFile().Chan {
		scipipe.Audit.Println("FileSplitter                          Now processing input file ", ft.Path(), "...")

		go func() {
			defer pop.Close()
			pop.Send(ft.Path())
		}()

		pip := scipipe.NewInParamPort(p.Name() + "temp_line_reader")
		pip.SetProcess(p)
		pip.From(fileReader.OutLine())

		go fileReader.Run()

		i := 1
		splitIdx := 1
		splitFt := newSplitIPFromIndex(ft.Path(), splitIdx)
		if !splitFt.Exists() {
			splitfile := splitFt.OpenWriteTemp()
			for line := range pip.Chan {
				// If we have not yet reached the number of lines per split ...
				/// ... then just continue to write ...
				if i < splitIdx*p.LinesPerSplit {
					splitfile.Write([]byte(line))
					i++
				} else {
					splitfile.Close()
					splitFt.Atomize()
					scipipe.Audit.Println("FileSplitter      Created split file", splitFt.Path())
					p.OutSplitFile().Send(splitFt)
					splitIdx++

					splitFt = newSplitIPFromIndex(ft.Path(), splitIdx)
					splitfile = splitFt.OpenWriteTemp()
				}
			}
			splitfile.Close()
			splitFt.Atomize()
			scipipe.Audit.Println("FileSplitter      Created split file", splitFt.Path())
			p.OutSplitFile().Send(splitFt)
		} else {
			scipipe.Audit.Printf("Split file already exists: %s, so skipping.\n", splitFt.Path())
		}
	}
}

var chars = []rune("abcdefghijklmnopqrstuvwxyz")

func getRandString(n int) string {
	rand.Seed(time.Now().UnixNano())
<commit_msg>Improve variable naming in FileSplitter component<commit_after>package components

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/scipipe/scipipe"
)

// FileSplitter is a process that will split a file into multiple files, each
// with LinesPerSplit number of lines per file
type FileSplitter struct {
	scipipe.BaseProcess
	LinesPerSplit int
}

// NewFileSplitter returns an initialized FileSplitter process that will split a
// file into multiple files, each with linesPerSplit number of lines per file
func NewFileSplitter(wf *scipipe.Workflow, name string, linesPerSplit int) *FileSplitter {
	p := &FileSplitter{
		BaseProcess:   scipipe.NewBaseProcess(wf, name),
		LinesPerSplit: linesPerSplit,
	}
	p.InitInPort(p, "file")
	p.InitOutPort(p, "split_file")
	wf.AddProc(p)
	return p
}

// InFile returns the port for the input file
func (p *FileSplitter) InFile() *scipipe.InPort { return p.InPort("file") }

// OutSplitFile returns the resulting split (part) files generated0
func (p *FileSplitter) OutSplitFile() *scipipe.OutPort { return p.OutPort("split_file") }

// Run runs the FileSplitter process
func (p *FileSplitter) Run() {
	defer p.CloseAllOutPorts()

	fileReader := NewFileReader(p.Workflow(), p.Name()+"_filereader_"+getRandString(7))
	pop := scipipe.NewOutParamPort(p.Name() + "_temp_filepath_feeder")
	pop.SetProcess(p)
	fileReader.InFilePath().From(pop)

	for ft := range p.InFile().Chan {
		scipipe.Audit.Println("FileSplitter                          Now processing input file ", ft.Path(), "...")

		go func() {
			defer pop.Close()
			pop.Send(ft.Path())
		}()

		pip := scipipe.NewInParamPort(p.Name() + "temp_line_reader")
		pip.SetProcess(p)
		pip.From(fileReader.OutLine())

		go fileReader.Run()

		lineNo := 1
		splitIdx := 1
		splitIP := newSplitIPFromIndex(ft.Path(), splitIdx)
		if !splitIP.Exists() {
			splitfile := splitIP.OpenWriteTemp()
			for line := range pip.Chan {
				// If we have not yet reached the number of lines per split ...
				/// ... then just continue to write ...
				if lineNo < splitIdx*p.LinesPerSplit {
					splitfile.Write([]byte(line))
					lineNo++
				} else {
					splitfile.Close()
					splitIP.Atomize()
					scipipe.Audit.Println("FileSplitter      Created split file", splitIP.Path())
					p.OutSplitFile().Send(splitIP)
					splitIdx++

					splitIP = newSplitIPFromIndex(ft.Path(), splitIdx)
					splitfile = splitIP.OpenWriteTemp()
				}
			}
			splitfile.Close()
			splitIP.Atomize()
			scipipe.Audit.Println("FileSplitter      Created split file", splitIP.Path())
			p.OutSplitFile().Send(splitIP)
		} else {
			scipipe.Audit.Printf("Split file already exists: %s, so skipping.\n", splitIP.Path())
		}
	}
}

var chars = []rune("abcdefghijklmnopqrstuvwxyz")

func getRandString(n int) string {
	rand.Seed(time.Now().UnixNano())

{'fraction_non_alphanumeric': 0.1306930693069307, 'fraction_numerical': 0.00144014401440144, 'mean_word_length': 3.7814113597246126, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '635314', 'n_tokens_mistral': 2118, 'n_tokens_neox': 1893, 'n_words': 455}