This document provides comprehensive examples of different types of Python nodes that can be created in Workflows, showing various patterns and use cases.

Overview

Python nodes in Workflows can be designed for many different purposes. This document shows examples of various node types to help you understand how to create nodes for different scenarios.

File Processing Nodes

File Copy Node

A node that copies files from input to output:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(120, 180, 180)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Copy file logic here
            output.addUniversalPath(up)
    
    return True

File Rename Node

A node that renames files:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(255, 165, 0)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("newNamePattern")
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    if not self.newNamePattern:
        return self.critical("rename", "Name pattern is required")
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Rename logic here
            new_up = up.clone()
            # Apply renaming pattern
            output.addUniversalPath(new_up)
    
    return True

Metadata Nodes

Metadata Reader Node

A node that reads metadata from files:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(128, 0, 128)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("metadataFields")
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Read metadata
            metadata = up.getMetadata()
            self.info("metadata", "File: {} - Metadata: {}".format(up.getFileName(), metadata))
            output.addUniversalPath(up)
    
    return True

Metadata Writer Node

A node that writes metadata to files:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(0, 128, 0)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("customMetadata")
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Write metadata
            if self.customMetadata:
                up.setMetadata(self.customMetadata)
            output.addUniversalPath(up)
    
    return True

External Application Nodes

FFmpeg Node

A complex node that integrates with FFmpeg:

from Mistika.Qt import QColor,QRegularExpression
from Mistika.classes import Cconnector,CbaseItem,CexternalAppNodeLink
from Mistika import sgoPaths
import os
import sys
from baseItemTools import totalNumberOfUPs
import Mistika
 
def ffmpegPath():
    ext=".exe" if sys.platform == "win32" else ""
    fldr = "/MacOS" if sys.platform == "darwin" else "/bin"
    fp=sgoPaths.apps()+ fldr + "/ffmpeg" + ext
    return fp
    
def init(self):
    self.setClassName("FFmpeg") 
    self.color=QColor(92, 184, 92)
    self.addConnector("input",Cconnector.CONNECTOR_TYPE_INPUT,Cconnector.MODE_OPTIONAL)    
    self.addConnector("output",Cconnector.CONNECTOR_TYPE_OUTPUT,Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True,"input_%") 
    self.addProperty("url")    
    self.addProperty("threads",1)
    self.addProperty("dstExtension","mp4")
    self.addProperty("cmdLineParams","-i [input] -c:v [param1] -b:v [param2] -c:a [param3] [param4] [param5] [param6] [param7] [param8] [output]")
    self.addProperty("param1","h264_nvenc")
    self.addProperty("param2","5M")
    self.addProperty("param3","aac")
    self.addProperty("param4","")
    self.addProperty("param5","")
    self.addProperty("param6","")
    self.addProperty("param7","")
    self.addProperty("param8","")
    
    self.bypassSupported=True
    self.setDropToProperty("url")
    #1=Directories, 2=Files, 3=both
    self.setDropSupportedTypes(1) 
    return True
        
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True    
    fp=ffmpegPath()
    if not os.path.exists(fp):
        return self.critical("ffmpeg:isReady","ffmpeg binary not found: ".format(fp))
    return True
 
 
def process(self):      
        
    res=True
    inputs=self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT,Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    out = self.getFirstConnectorByName("output")
    out.clearUniversalPaths()
    dstUrl=self.evaluate(self.url).strip()
    self.setComplexity(totalNumberOfUPs(self)*100)
    if self.bypassEnabled:
        return True
        
    self.progressUpdated(0)
    binfp=ffmpegPath()
    apps=[]
    threads=int(self.threads)    
# First, we create a queue of threads to execute the binary
    queue=self.createExecutionQueue(threads)
    cancelled=False
    totalFrames=0
    for c in inputs:
        if cancelled:
            break
        for up in c.getUniversalPaths():
            mfid=up.getMediaFileInfoData()
            frames=mfid.getToken("Frames")
            fp=up.getFilePath()
            self.info("ffmpeg:process:up","Preparing {}".format(fp))
            dst=self.composeDstFilePath(dstUrl,up,False,0)
            ext=self.evaluate(self.dstExtension.strip())
            if ext:
                dst.setExtension(ext)     
# For each UP to process, we create an instance of the app to execute it
            app=self.createExternalApp()
# Add the new app to the apps list. this list is used at the end to delete them
            apps.append((app,up,dst))
            if frames>0:
                totalFrames+=frames
                app.setMaxProgress(frames,totalFrames)       
# Register the Hooks to parse the command output
            app.addHookRX("errorRX",QRegularExpression("^(Error .*)$"))
            app.addHookRX("progressRX",QRegularExpression("^frame=\\s*(\\d+).*"))
            app.setOutputUP(dst)
# select if we want the app to print its output or not (printing it is not needed for the books to work, it is just for log/debug proposes            
            app.setPrintOutputs(app.EA_STD_BOTH)
# define the default tokens dictionary ([input] [input-firstFile] [output])            
            defTokens=app.createDefaultDict(up)
#parse the cmdLIne tokens            
            args=app.createArgumentsList(self.evaluate(self.cmdLineParams.strip()),up,defTokens)
            args=('-y',)+args # adds -y to cmdline to avoid ffmpeg blocking waiting an overwrite answer
# add " to the elements with spaces
            quoted=app.quoteList(args)
            if self.isCancelled():
                res=False
                cancelled=True
                queue.cancelAll()
                break
            self.info("ffmpeg:process:args","\"{}\" {}".format(binfp,' '.join(quoted)))
#execute the job           
            app.setParams(binfp,args)                   
            queue.addJob(app)
#wait for the apps to finish       
    while queue.wait(1000):
        if self.isCancelled():
            res=False
            queue.cancelAll()
#build the output connector and delete all the apps and queue
    for app,up,dst in apps:   
        exitCode=app.exitCode()
        fp=up.getFilePath()
        if exitCode: 
            res=self.critical("ffmpeg:process:error","Unable to process {}. exitCode={}".format(fp,exitCode)) and res
            self.addFailedUP(up,"Unable to process {}. exitCode={}".format(fp,exitCode))
        else:
            out.addUniversalPath(dst)
        app.deleteLater()
    queue.deleteLater()
    self.progressUpdated(self.complexity())
    return res
 
# appId ExternalApp emiting the signal
# output can be EA_STD_OUT or EA_STD_ERR
# name is the hook name
# rxMatch is the matching QRegularExpressionMatch 
# line is the matching line
def onExternalAppHook(self,appId,output,name,rxMatch,line):
    # you can manage your custom hooks here
    print("Signal",name,"received from std",output,"(Ignored)")

Network Nodes

Google Drive Node

A node that interacts with Google Drive:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(30, 144, 255)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("driveFolderId")
    self.addProperty("accessToken")
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    if not self.driveFolderId:
        return self.critical("drive", "Drive folder ID is required")
    if not self.accessToken:
        return self.critical("drive", "Access token is required")
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    # Google Drive integration logic here
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Upload/download logic
            output.addUniversalPath(up)
    
    return True

Data Processing Nodes

CSV Processor Node

A node that processes CSV data:

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
import csv
 
def init(self):
    self.color = QColor(255, 215, 0)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("csvDelimiter", ",")
    self.addProperty("processHeaders", True)
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()
    
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # CSV processing logic
            with open(up.getFilePath(), 'r') as f:
                reader = csv.reader(f, delimiter=self.csvDelimiter)
                # Process CSV data
                pass
            output.addUniversalPath(up)
    
    return True

Utility Nodes

Configuration Node

A node that manages configuration:

from Mistika.Qt import QColor
 
def init(self):
    self.color = QColor(225, 225, 210)
    self.addProperty("watchers") 
    self.addProperty("deliveries")
    return True
 
def isReady(self):
    res=True
    if self.watchers=="":
        res=self.critical("config:watchers","'Watchers' folder can not be empty") and res
    if self.deliveries=="":
        res=self.critical("config:deliveries","'Deliveries' folder can not be empty") and res
    return res
 
def process(self):
    return True

Logger Node

A node that logs information:

from Mistika.Qt import QColor
 
def init(self):
    self.color = QColor(128, 128, 128)
    self.addProperty("logLevel", "INFO")
    self.addProperty("logMessage")
    return True
 
def isReady(self):
    return True
 
def process(self):
    self.info("logger", self.logMessage)
    return True

Node Structure Best Practices

Consistent Structure

All nodes should follow a consistent structure:

# Import statements
from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
# Helper functions (if needed)
def helper_function():
    pass
 
# Main functions
def init(self):
    # Setup code
    return True
 
def isReady(self):
    # Readiness checks
    return True
 
def process(self):
    # Main processing logic
    return True

Error Handling

Proper error handling should be implemented:

def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    
    # Validate properties
    if not self.requiredProperty:
        return self.critical("validation", "Required property missing")
    
    # Check external dependencies
    if not self.checkExternalDependency():
        return self.critical("dependency", "External dependency not available")
    
    return True
 
def process(self):
    try:
        # Main logic
        pass
    except Exception as e:
        self.critical("process", "Error: {}".format(str(e)))
        return False
    
    return True

Testing Examples

Manual Testing

Nodes should be tested manually in workflows:

Create workflow with the node
Connect appropriate inputs/outputs
Run workflow
Check logs for errors
Verify outputs are correct

Parameter Validation

def isReady(self):
    # Validate numeric parameters
    if self.threads < 1:
        return self.critical("threads", "Threads must be at least 1")
    
    # Validate string parameters
    if not self.url or not self.url.strip():
        return self.critical("url", "URL cannot be empty")
    
    # Validate file paths
    if self.filePath and not os.path.exists(self.filePath):
        return self.critical("file", "File path does not exist")
    
    return True

Summary

This document provides examples of various node types that can be created in Workflows:

File processing nodes
Metadata nodes
External application integration nodes
Network nodes
Data processing nodes
Utility nodes

Each example demonstrates different patterns and best practices for node development. The key is to understand the requirements and choose the appropriate patterns for your specific use case.