import "github.com/scipipe/scipipe"
Package scipipe is a library for writing scientific workflows (sometimes also called "pipelines") of shell commands that depend on each other, in the Go programming languages. It was initially designed for problems in cheminformatics and bioinformatics, but should apply equally well to any domain involving complex pipelines of interdependent shell commands.
audit.go const.go ip.go log.go port.go process.go sink.go task.go utils.go workflow.go
const (
// Standard buffer size used for channels connecting processes
BUFSIZE = 16
)var (
Trace *log.Logger
Debug *log.Logger
Info *log.Logger
Audit *log.Logger
Warning *log.Logger
Error *log.Logger
LogExists bool
)ExpandParams takes a command pattern and a map of parameter names mapped to parameter values, and returns the command as a string where any parameter placeholders (on the form `{p:paramname}` are replaced with the parameter value from the provided parameter values map.
Code:
fmt.Println(ExpandParams("echo {p:msg}", map[string]string{"msg": "Hello"}))
Output:
echo Hello
func InitLog(
traceHandle io.Writer,
debugHandle io.Writer,
infoHandle io.Writer,
auditHandle io.Writer,
warningHandle io.Writer,
errorHandle io.Writer)Initiate logging
func InitLogAudit()
Initiate logging with level=AUDIT
func InitLogDebug()
Initiate logging with level=DEBUG
func InitLogError()
Initiate logging with level=ERROR
func InitLogInfo()
Initiate logging with level=INFO
func InitLogWarning()
Initiate logging with level=WARNING
type AuditInfo struct {
Command string
Params map[string]string
ExecutionTimeMilliSeconds time.Duration
UpstreamAuditInfos map[string]*AuditInfo
}ExecMode specifies which execution mode should be used for a SciProcess and its corresponding SciTasks
const (
// ExecModeLocal indicates that commands on the local computer
ExecModeLocal ExecMode = iota
// ExecModeSLURM indicates that commands should be executed on a HPC cluster
// via a SLURM resource manager
ExecModeSLURM ExecMode = iota
)type FilePort struct {
Port
Chan chan *InformationPacket // TODO: Deprecated
// contains filtered or unexported fields
}FilePort
func (pt *FilePort) AddRemoteChan(remoteChan chan *InformationPacket)
func (pt *FilePort) Recv() *InformationPacket
func (pt *FilePort) Send(ip *InformationPacket)
type IPGen struct {
Process
Out *FilePort
FilePaths []string
// contains filtered or unexported fields
}IPGen is initialized by a set of strings with file paths, and from that will return instantiated (generated) InformationPacket on its Out-port, when run.
Initialize a new IPGen component from a list of file paths
Check if the IPGen outport is connected
Execute the IPGen, returning instantiated InformationPacket
InformationPacket contains information and helper methods for a physical file on a normal disk.
func NewInformationPacket(path string) *InformationPacket
Create new InformationPacket "object"
func (ip *InformationPacket) Atomize()
Change from the temporary file name to the final file name
func (ip *InformationPacket) CreateFifo()
Create FIFO file for the InformationPacket
func (ip *InformationPacket) Exists() bool
Check if the file exists (at its final file name)
func (ip *InformationPacket) GetAuditInfo() *AuditInfo
func (ip *InformationPacket) GetFifoPath() string
Get the path to use when a FIFO file is used instead of a normal file
func (ip *InformationPacket) GetPath() string
Get the (final) path of the physical file
func (ip *InformationPacket) GetTempPath() string
Get the temporary path of the physical file
func (ip *InformationPacket) Open() *os.File
Open the file and return a file handle (*os.File)
func (ip *InformationPacket) OpenTemp() *os.File
Open the temp file and return a file handle (*os.File)
func (ip *InformationPacket) OpenWriteTemp() *os.File
Open the file for writing return a file handle (*os.File)
func (ip *InformationPacket) Read() []byte
Read the whole content of the file and return as a byte array ([]byte)
func (ip *InformationPacket) RemoveFifo()
Remove the FIFO file, if it exists
func (ip *InformationPacket) SetAuditInfo(ai *AuditInfo)
func (ip *InformationPacket) TempFileExists() bool
Check if the temp-file exists
func (ip *InformationPacket) WriteAuditLogToFile()
func (ip *InformationPacket) WriteTempFile(dat []byte)
Write a byte array ([]byte) to the file (first to its temp path, and then atomize)
ParamPort
type Process interface {
Name() string
IsConnected() bool // Sanity check, to see whether all ports are connected
Run()
}Base interface for all processes
type SciProcess struct {
Process
CommandPattern string
ExecMode ExecMode
Prepend string
Spawn bool
OutPortsDoStream map[string]bool
PathFormatters map[string]func(*SciTask) string
CustomExecute func(*SciTask)
// contains filtered or unexported fields
}func NewProc(name string, cmd string) *SciProcess
func NewSciProcess(name string, command string) *SciProcess
func ShellExpand(name string, cmd string, inPaths map[string]string, outPaths map[string]string, params map[string]string) *SciProcess
func (p *SciProcess) GetInPorts() map[string]*FilePort
func (p *SciProcess) GetOutPorts() map[string]*FilePort
func (p *SciProcess) GetParamPorts() map[string]*ParamPort
func (p *SciProcess) In(portName string) *FilePort
func (proc *SciProcess) IsConnected() (isConnected bool)
------- Sanity checks -------
func (p *SciProcess) Name() string
func (p *SciProcess) Out(portName string) *FilePort
func (p *SciProcess) PP(paramPortName string) *ParamPort
func (p *SciProcess) ParamPort(paramPortName string) *ParamPort
func (p *SciProcess) Run()
Run runs the process by instantiating and executing SciTasks for all inputs and parameter values on its in-ports. in the case when there are no inputs or parameter values on the in-ports, it will run just once before it terminates. note that the actual execution of shell commands are done inside SciTask.Execute, not here.
func (p *SciProcess) SetInPort(portName string, port *FilePort)
func (p *SciProcess) SetOutPort(portName string, port *FilePort)
func (p *SciProcess) SetParamPort(paramPortName string, paramPort *ParamPort)
func (p *SciProcess) SetPathCustom(outPortName string, pathFmtFunc func(task *SciTask) (path string))
SetPathCustom takes a function which produces a file path based on data available in *SciTask, such as concrete file paths and parameter values,
func (p *SciProcess) SetPathExtend(inPortName string, outPortName string, extension string)
SetPathExtend creates an (output) path formatter that extends the path of an input InformationPacket
SetPathReplace creates an (output) path formatter that uses an input's path but replaces parts of it.
func (p *SciProcess) SetPathStatic(outPortName string, path string)
SetPathStatic creates an (output) path formatter returning a static string file name
type SciTask struct {
Name string
Command string
ExecMode ExecMode
CustomExecute func(*SciTask)
InTargets map[string]*InformationPacket
OutTargets map[string]*InformationPacket
Params map[string]string
Done chan int
Image string
DataFolder string
}func NewSciTask(name string, cmdPat string, inTargets map[string]*InformationPacket, outPathFuncs map[string]func(*SciTask) string, outPortsDoStream map[string]bool, params map[string]string, prepend string, execMode ExecMode) *SciTask
type ShellProcess interface {
Process
In(string) *FilePort
GetInPorts() map[string]*FilePort
Out(string) *FilePort
GetOutPorts() map[string]*FilePort
SetPathStatic(outPortName string, path string)
SetPathExtend(inPortName string, outPortName string, extension string)
SetPathReplace(inPortName string, outPortName string, old string, new string)
SetPathCustom(outPortName string, pathFmtFunc func(task *SciTask) (path string))
}Sink is a simple component that just receives InformationPacket on its In-port without doing anything with them
Instantiate a Sink component
Execute the Sink component
type Workflow struct {
// contains filtered or unexported fields
}ConnectLast connects the last (most downstream) out-ports in the workflow to an implicit sink process which will be used to drive the workflow. This can be used instead of manually creating a sink, connecting it, and setting it as the driver process of the workflow.
func (wf *Workflow) NewProc(procName string, commandPattern string) *SciProcess
| Path | Synopsis |
|---|---|
| components |
Package scipipe imports 14 packages (graph) and is imported by 1 packages. Updated 2017-07-02. Refresh now. Tools for package owners.