This section provides an example of how you can use the Python collection and processing agents for a batch workflow. In this instance the Python collection agent reads an Excel file.
Image Added
Workflow example with the Python collection and processing agents for batch
Image Added
The Excel file being processed
Workflow Configuration
The sections below provide descriptions of agent configurations for the example workflow shown above.
Read_Excel
The Python collection agent is configured to read all the Excel files from the configured directory, process each file in a separate batch and output one UDR per row.
General Tab
Image Added
| Info |
|---|
| title | Example - Code for Read_Excel |
|---|
|
| Code Block |
|---|
import os
import pandas
from .UFL_Types import ItemCount
txn_directory = persistent()
txn_filename = persistent()
def execute():
# Read all excel files from the configured directory
txn_directory.value = dynamicFieldGet("Directory", "Name")
xls = [f for f in os.listdir(txn_directory.value) if f.endswith(".xlsx")]
# Process each excel file in a separate batch, output one UDR per row
for fn in xls:
if not isStopped():
txn_filename.value = fn
df = pandas.read_excel(os.path.join(txn_directory.value, fn))
mimSet("Filename", fn)
beginBatch()
debug("Begin batch (txnid=%s, filename=%r)" % (getTransactionId(), fn))
for idx, item, count in df.itertuples():
udr = ItemCount()
udr.item = item
udr.count = count
udrRoute(udr)
endBatch()
def commit():
# On commit, move the processed file into the "done/" subdirectory
debug("Commit (txnid=%s, filename=%r)" % (getTransactionId(), txn_filename.value))
old = os.path.join(txn_directory.value, txn_filename.value)
new = os.path.join(os.path.join(txn_directory.value, "done"), txn_filename.value)
if os.path.exists(old):
os.rename(old, new)
|
|
MIM Tab
The agent publishes Filename as a string.
Image Added
Analyse_Rows
The Python processing agent processes UDRs by defining a consume block, and uses the selected Interpreter profile that is used to configure the Python executables.
Image Added
| Info |
|---|
| title | Example - Code for Analyse_Rows |
|---|
|
| Code Block |
|---|
total_count = 0
def beginBatch():
fn = mimGet("Read_Excel", "Filename")
debug("Begin batch (txnid=%s, filename=%r)" % (getTransactionId(), fn))
global total_count
total_count = 0
def consume(input):
debug("%s=%s" % (input.item, input.count))
global total_count
total_count += input.count
def endBatch():
fn = mimGet("Read_Excel", "Filename")
debug("File %r contained %s items" % (fn, total_count))
|
|
Workflow Monitor
The output is shown step by step in the workflow monitor when you have debug enabled.
Image Added
Workflow Monitor showing the output for both agents
