Shutdown Python Scripts in FME

Introduction

This article is part 5 of a 5 part series to help you get started with basic Python and FME. This article is about using the shutdown Python script functionality within Workbench and includes examples of using a shutdown Python script to email translation statistics, open the output dataset with the default application for the file format, and write to a translation history log.
 

A Python script can be added to a workspace that will run when the workspace completes. For example, you may want to copy files somewhere when a workspace is finished or open the output dataset with an application that is not Data Inspector.

A shutdown script runs after all of the transformers, readers, writers, and the log file are closed in a successful translation. In the case of a failed translation, the script will still execute but may fail as only fme.status and fme.failureMessage are guaranteed to contain valid values.

As mentioned on the Help page for Startup and Shutdown scripts, it is not recommended to use any modules from the FME Objects Python API with shutdown scripts because resources required by FME Objects are no longer available so calls to the API will have unpredictable results.

However, you can access the FME parameter values in the same way as in Startup scripts, scripted parameters, and PythonCaller using the fme.macroValues[] dictionary. A shutdown Python script can also access a number of global variables which FME creates such as fme.featuresRead, fme.featuresWritten, and many more. In addition, the script can access any global Python variables set in a startup Python script. For a complete list of the global variables please see the Help page for Startup and Shutdown Python Scripts, under the topic FME_END_PYTHON.

Examples

Example 1: Email Translation Statistics

Attached workspace: ShutdownPython1.fmw

In this example, we add a shutdown Python script to a workspace that sends an email using Gmail when the workspace finishes. You can edit the script to use other mail servers if you wish. The subject of the email will say whether the workspace was successful or not and include the number of features written. A similar example can also be found on the Help page for Startup and Shutdown Python Scripts. This task can also be accomplished using the Emailer transformer.

Open the ShutdownPython1 workspace (see Files section above) and in the Navigator pane under Workspace Parameters > Scripting, you will see a parameter called Shutdown Python Script. Here you can enter or view the Python code below which will send the email using a GMail account.

Location of Shutdown Python Script parameter within the Navigator pane

The script imports the smtplib and fme modules, which gives emailing tools and access to FME dictionaries respectively. The next section retrieves the FME parameters set by the user. Then the script gets the number of features written and converts the number into text. The if/else statement determines which message text to use according to whether the translation succeeded or failed. The last section deals with using the GMail smtp server to send the email.

# The smtplib gives us the emailing tools
import smtplib, fme
 
# Get the FME parameter values set when workspace run
to =  fme.macroValues['EmailTo']
gmail_user =  fme.macroValues['GmailUser']
gmail_pwd =  fme.macroValues['Password']
Subject =  fme.macroValues['SuccessSubject']
FailSubject =  fme.macroValues['FailSubject']
Emailfrom =  fme.macroValues['EmailFrom']

#Get number of features FME wrote
FeaturesWritten = str(fme.featuresWritten)

# If FME Fails then Change Subject
# FME_Status tells if FME was successful
status = fme.status
if status == 0:
    Subject =  FailSubject
    Message = 'Workspace Failed'    
else:
    Message = f"Workspace Successful the following features were written: {FeaturesWritten}"

# Sending the email

# Using the gmail smtp server
#You can edit this line to use your own email server
smtpserver = smtplib.SMTP("smtp.gmail.com",587)

smtpserver.ehlo()
smtpserver.starttls()
smtpserver.ehlo()
smtpserver.login(gmail_user, gmail_pwd)
header = f"To:{to}\nFrom: {Emailfrom}\nSubject:{Subject}\n"
print(header)
msg = f"{header}\n{Message}\n\n"
smtpserver.sendmail(gmail_user, to, msg)
print('done!')
smtpserver.close()

There are two key points to note:

1. We are getting user parameters using the fme.macroValues[] dictionary in the same way as in examples from Startup Python Script and Scripted Parameters articles. For example, the Gmail username can be set when the workspace is run and is accessed from the published parameter like this: fme.macroValues['GmailUser'].
2. Secondly, notice we use two global variables from fme module: fme.featuresWritten which is a Python dictionary indexed by feature type, and fme.status which is a boolean and tells us if the workspace succeeded or failed.

Run the workspace using File > Run with Prompt or select the prompt and run button, then the run button. A Translation Parameters dialog box will appear. Fill in the appropriate parameters. You will need your own Gmail account username (email address)** and password and a from-email address which should be the same as the username. Also, fill in the to-email address (can be the same as the from email address) and a subject line for failure and success. In the attached sample workspace the Fail parameter is used to simulate workspace failure by sending features to the Terminator transformer. Click OK to send.

**Note: If you have two-factor authentication enabled for your Gmail account, you may need to generate an App password associated with your Gmail account. Login to your Gmail account and you will find a page to generate App passwords under My Account > Sign-in & security. Once you have generated one, enter the App password as your password when you run the workspace. If you use your normal password, the Python script may fail to execute with the message “Username and password not accepted” or Authentication Error.

Please see the following two Knowledge Center articles for more details on authentication issues when using Gmail:

IMAP Publication or Email Subscription are not Reading Emails from Gmail

Authentication issue when using Gmail as an FME Server email subscriber

Translation parameters dialog box for ShutdownPython1.fmw

You (or your recipient) should receive the email sent by the shutdown Python script shortly. It should look like this if you chose ‘No’ for the fail parameter:

Example of successful translation email sent by the shutdown script

Or if you chose ‘Yes’ for the fail parameter:

Example of failed translation email sent by the shutdown script

Example 2: Open output dataset with the default application

Attached workspace: ShutdownPython2.fmwt

With Workbench, users have the option to redirect dataset(s) written to the Data Inspector. However, what if the user wants to automatically open the destination dataset with the data’s default application? A shutdown Python script can be used to do so.

In this example, we use a shutdown Python script to open a CSV file after the translation is finished. A CSV file was chosen for this example as it is a file format that most users can open. But you should be able to open other file formats with the format’s default application using this script.

Open ShutdownPython2.fmwt. The workspace is quite simple. It reads individual tree data within the city of Vancouver from a CSV file. An AttributeManager is used to delete any attributes that are not required. An Aggregator is used to find the number of trees in a neighbourhood and to find the average height and diameter of all the trees in a neighbourhood. The output is written out as a CSV file.

Workspace for ShutdownPython2.fmwt which reads tree data from a CSV, aggregates the data by neighborhood and writes output in CSV format.

The shutdown script is used to open the output CSV file with the default application set on your machine. In the Navigator pane, go to Workspace Parameters > Scripting > Shutdown Python Script to view the script or copy it below if you are creating the workspace with your own data.

import subprocess, fme, sys

# Paths of common CSV default application on Windows 64-bit
# r"C:\Program Files (x86)\Microsoft Office\Office15\EXCEL.exe"
# r"C:\Program Files (x86)\Notepad++\notepad++.exe" r”C:\Windows\System32\notepad.exe”

# Use FME's macro values dictionary to get directory location of output
outputDir = fme.macroValues['DestDataset_CSV2']
# Use FME's macro values dictionary to get output file name
outputName = fme.macroValues['csv_file_name']
# Use FME's macro values dictionary to get output file extension
ext = fme.macroValues['CSV2_OUT_EXTENSION_CSV2']


# Append directory and file name to get file path of output file
outputPath = f'{outputDir}\\{outputName}.{ext}'

# Print functions for debugging
#print(outputPath)

# Use subprocess module to open output file with default application
try:
    # To use shell=False, you should specify the full file path of the default application and outputfile
    # Uncomment the line below, add your applicable file paths, and comment out the line containing shell=True. 
    # P = subprocess.Popen(["<DefaultApplicationFilePath>", "<OutputDatasetFilePath>"], shell=False)
    P = subprocess.Popen([outputPath], shell=True)
    print("Output CSV file successfully opened")
except Exception as e:
    print("SHUTDOWN SCRIPT FAILED. Error: ", e)

The script is relatively simple. Using the subprocess and the fme modules, the script finds and creates a variable containing the file path of the output dataset. This variable is used with the subprocess module’s Popen constructor to create a child process that opens the output dataset with the format’s default application.

The Popen constructor was used instead of the subprocess.call() because call() will use the same process as the translation engine (FME.exe), which means your translation will not finish until the user closes the application which is used to open the output dataset. As the Popen() constructor creates a child process, FME.exe can be closed after the translation and the workspace will not hang until the application opened is closed.

Important note: Using the shell=True argument for the Popen constructor is generally not recommended as this makes programs vulnerable to shell injection and can be a security hazard if combined with untrusted input. For more information, please see the official Python documentation here: Python Popen Constructor. The shell=True argument is used for simplicity as it does not require the application path to be specified. In this case, since the workspace user is responsible for the input for the command, it poses less of a security hazard. However, use it at your own risk.

If you wish to use shell=False as the argument, uncomment the following line:

P = subprocess.Popen(["<DefaultApplicationFilePath>", "<OutputDatasetFilePath>"], shell=False)

Replace <DefaultApplicationFilePath> and <OutputDatasetFilePath> with the applicable file paths. Make sure the quotes are retained. Comment out the line containing shell=True.

Ensure the Run with Prompt is enabled and run the workspace. Enter your parameters or leave as defaults. 6 features should be written out to the output CSV file. The application set as the default for CSV file formats on your machine should start automatically and you should be able to view and edit the output CSV file.

Message in Translation Log pane saying output CSV file successfully opened. If the CSV file could not be opened, an error message will appear instead.

Example 3: Create a translation history log

Attached workspace: ShutdownPython3.fmw

How can a user keep track of translations if a workspace is run multiple times or on a schedule (for example: as shown in this Batch Processing Method 1: Command Line or Batch File article)? Of course, you can look through log files if the 'Append to log' option was selected within Workbench (see here for log documentation) or if you have specified a log history to be created within your batch file. But another option is to create a translation history log for a specific workspace with a shutdown Python script.

In this example, the shutdown script will create or append to a CSV log with attributes that a user may be interested in such as translation status, translation date and time, features read or written, how long the translation took, etc. To do this, the script takes advantage of a number of the fme module’s global variables.

Open the workspace attached (ShutdownPython3.fmw - please see Files section above). The workspace is very simple containing a LANDSAT8 reader, Tester, GeoTIFF writer, and Terminator. This workspace will retrieve the latest imagery matching the reader’s criteria and write out the imagery as GeoTIFF raster(s).

A very simple workspace which demonstrates using a shutdown Python script to create a translation history log.

In the Navigator pane under Workspace Parameters > Scripting, you will see a parameter called Shutdown Python Script. Double-click to open the script editor window. If you are creating your own workspace, copy and paste the Python code below which will create a translation history log or append translation details.

# Import modules
import fme, csv, datetime
from os import path

# Get current date and time
dateToday = datetime.date.today().strftime('%Y/%m/%d')
timeNow = datetime.datetime.now().time().strftime('%H.%M.%S')

# Get workspace's directory path
# Replace backward slashes with forward slashes to avoid escape character issues
workspacePath = fme.macroValues['FME_MF_DIR']
workspacePath = workspacePath.replace('\\','/')
# Get workspace name
workspaceName = fme.macroValues['WORKSPACE_NAME']

# Create file path for master file
masterFile = f'{workspacePath}{workspaceName}.csv'

try:
    # If master file already exists, open and append to file
    if path.exists(masterFile) == True:
        # Create CSV writer
        with open(masterFile, 'a', newline='') as csvfile:
            writer = csv.writer(csvfile, delimiter=",", quotechar="'" , quoting=csv.QUOTE_MINIMAL)
            # If translation status is successful, write all translation attributes
            if fme.status == True:
                writer.writerow([dateToday, timeNow, fme.status, fme.failureMessage, fme.elapsedRunTime, fme.totalFeaturesRead, fme.totalFeaturesWritten, fme.logFileName])
            # If translation status is not successful, write status and failure message only
            else:
                writer.writerow([dateToday, timeNow, fme.status, fme.failureMessage,'','','',''])

    # If master file does not exist, open and write to file
    else:
        with open(masterFile, 'w', newline='') as csvfile:
            # Create CSV writer
            writer = csv.writer(csvfile, delimiter=",", quotechar="'" , quoting=csv.QUOTE_MINIMAL)
            # Write header row
            writer.writerow(['Date', 'Time', 'TranslationStatus', 'FailureMessage', 'TranslationTime', 'TotalFeaturesRead', 'TotalFeaturesWritten', 'LogFileName'])
            # If translation status is successful, write all translation attributes
            if fme.status == True:
                writer.writerow([dateToday, timeNow, fme.status, fme.failureMessage, fme.elapsedRunTime, fme.totalFeaturesRead, fme.totalFeaturesWritten, fme.logFileName])
            # If translation status is not successful, write status and failure message only
            else:
                writer.writerow([dateToday, timeNow, fme.status, fme.failureMessage,'','','',''])
except Exception as e:
    print("WRITE TO TRANSLATION HISTORY FAILED")
    print("Error: ", e)

For simplicity, the translation history log will be created in the same folder as the workspace and have the same name as the workspace. The script uses the fme.macroValues[] dictionary, to get the path of the folder containing the workspace and the workspace name with the macros (predefined parameters) ‘FME_MF_DIR’ and ‘WORKSPACE_NAME’ to create a file path for the translation history file.

The actual creation, writing/appending of the CSV file is controlled by nested if/else statements. The first if/else checks whether the translation history file exists or not with path.exists(). If the workspace has not been run before, there will be no translation history log (path.exists() will be False). The csv log will be opened with the built-in Python function open() using the write (‘w’) mode.

The script will then write two rows: first, the header row, as this row will not exist, and second, the row containing the translation details. Translation details written are accessed with global variables such as fme.elapsedRunTime, fme.totalFeaturesWritten, fme.totalFeaturesRead, etc. For the complete list of fme functions and variables available for Python shutdown scripts, please see FME_END_PYTHON in the Documentation section below.

If the workspace has been run previously, the translation history file will already exist so append mode (‘a’) is used instead. Append will write to the end of the file which means previous rows won’t be overwritten and lost. In addition, it would not make sense to write another header row to the file since one already exists, so only the translation detail row will need to be written.

The second if/else statement checks the translation status with the fme.status variable. Remember, if a translation fails, only fme.status and fme.failureMessage are guaranteed to contain valid values. In order to avoid potential issues with invalid values, if fme.status is not True (ie. translation failed), then only the date, time, translation status, and failure message will be written.

Run the workspace, enter 'No' for the Terminate Translation parameter, and enter where you want the output to be written. The feature passes through the Passed Tester port. After the workspace finishes, go to the folder where the workspace is saved. You should see a CSV file with the same name as the workspace (ShutdownPython3.csv for this example). Open the file. There should be two rows: header and the row containing details for the translation that just finished. Close the CSV file.

How the translation history log appears after the workspace is run successfully.

Go back to the workspace. Run the workspace again but select 'Yes' for Terminate Translation parameter. This time, the translation should fail due to the Terminator as the feature passes through the Failed Tester port.

Open the translation history file. There should be a third row that only contains the date, time, translation status, and failure message.

The translation history log's last row should contain only four values after a failed translation.

You can now run your workspace with a .bat file (or your OS equivalent), a task scheduler, or even manually through Workbench and have a record of each translation for the workspace in an easily searchable file.

For more complex shutdown Python examples, see the article:

Example Workflow using FME, Python and Oracle
 

Documentation

To access Help for Startup and Shutdown scripts within Workbench, open Help from the menu bar and enter the keyword ‘Python’ in the search bar and select the appropriate entry, or go to:

Startup and Shutdown Python Scripts

For a complete list of FME variables and functions available for Shutdown Python scripts please go to: FME_END_PYTHON

You can find complete documentation of the Python FME Objects API here: FME Objects Python API

Data Attribution

The data used here originates from open data made available by the City of Vancouver, British Columbia. It contains information licensed under the Open Government License - Vancouver.