Creación de un archivo lógico Python para aplicaciones de proceso

Especificación del punto de entrada

El punto de entrada o la función de entrada sirve de entrada al proceso de transformación de datos. Sin un punto de entrada definido en el archivo lógico, la transformación de datos falla incluso antes de empezar.

Este es el punto de entrada en la plantilla del archivo lógico de python:

def execute(context):

Especificación de variables configurables

Las variables configurables calculan el proceso de transformación de los datos. Incluyen variables dinámicas como token key, server o systemURL, username, y password, y otros campos de entrada dinámicos.

Si especifica las variables, más adelante el usuario podrá configurar los datos del registro de eventos cuando utilice su aplicación de procesos. Para más información, consulte Definición de entradas de usuario.

A partir de Process Mining 1.14.3, ya no es posible obtener variables configurables como variables de entorno.

Puede especificar variables configurables en una aplicación de proceso personalizada utilizando un objeto configurable de los objetos de contexto. Las variables de configuración están disponibles en un diccionario Python que puede recuperar utilizando la sintaxis context.config . En el siguiente ejemplo, se define la función execute como parámetro context :

def execute(context): 
config = context[“config”] 
# example retrieving an input property named ‘URL’ 
url = config[“URL”]

Si la fuente de datos brutos que el usuario de su aplicación de procesos carga está en formato .CSV , puede configurar el archivo lógico Python para transformar los datos para que sean la fuente de su proyecto de minería de procesos. Antes de cargar los datos, el usuario debe transformar el archivo del formato .CSV al formato .zip . A continuación, el archivo lógico Python accede al archivo .zip para realizar la transformación de datos necesaria en la misma ubicación del directorio.

El archivo lógico también puede recuperar el nombre del archivo .zip cargado desde el contexto JSON que se pasa a la función execute del archivo lógico. El nombre de la clave para el nombre del archivo lógico cargado es fileUploadName y el valor es el nombre del archivo cargado .zip . Por ejemplo, si el nombre del archivo .zip cargado es "Input_sample.zip", el archivo lógico puede recuperar el nombre del archivo .zip mediante el siguiente comando:

def execute(context): 
# Example retrieving the name of an uploaded zip file 
myFileUploadName = context[“fileUploadName”] 

Opcional: Establecer un calendario para las opciones de actualización de datos

Como usuario de la aplicación de procesos personalizados, dispone de dos opciones de programación para gestionar las actualizaciones de los datos del registro de eventos de su proceso:

• El método de sustitución, en el que se sustituye la fuente de datos existente. No se tiene en cuenta la fuente de datos del proyecto anterior. La nueva fuente de datos se procesa en función de sus necesidades: puede elegir la frecuencia con la que se sustituyen los datos o utilizar un calendario para establecer el momento preferido para la sustitución de los datos.
  

  Si sólo desea utilizar el método de sustitución, no tiene que realizar ningún cambio adicional en su archivo lógico Python.

• El método de actualización incremental, en el que los nuevos datos se añaden sobre la fuente de datos del proyecto existente. Tanto los datos anteriores como los nuevos se utilizan en el proyecto.
  

  Si desea utilizar el método de actualización incremental, debe configurar el archivo lógico asociado Python para que genere una programación para obtener nuevos datos de una fuente externa. Para ello, especifique dos variables clave, lastScheduleExecutionDate y lastScheduleExecutionTimezone, que se proporcionan en el archivo lógico Python.

  Las variables lastScheduleExecutionDate y lastScheduleExecutionTimezone son esenciales para las operaciones lógicas dentro del script Python, permitiendo la obtención incremental de nuevos datos de registro de eventos basados en la última fecha y zona horaria de ejecución exitosa del programa.

  Puede acceder a las variables lastScheduleExecutionDate y lastScheduleExecutionTimezone a través del objeto context. Estas variables son accesibles dentro del archivo lógico Python utilizando el diccionario de contexto. Esto se pasa a la lógica Python a través de la función de ejecución:

  last_schedule_execution_date = context["lastSuccessfulExecution"]["lastScheduleExecutionDate"]
  last_schedule_execution_timezone = context["lastSuccessfulExecution"]["lastScheduleExecutionTimezone"]
  …
  # Subsequent code using last_schedule_execution_date and last_schedule_execution_timezone
  if last_schedule_execution_date and last_schedule_execution_timezone:
     # Logic using the last scheduled execution date and timezone
  

from datetime import datetime
import zoneinfo

...

timezone = zoneinfo.ZoneInfo(last_schedule_execution_timezone )

# Parse the ISO 8601 formatted date string
date = datetime.fromisoformat(last_schedule_execution_date ).astimezone(timezone)

LAST_SCHEDULE_EXECUTION_DATE = None
LAST_SCHEDULE_EXECUTION_TIMEZONE = None

...

def execute(context):
   LAST_SCHEDULE_EXECUTION_DATE = context["lastSuccessfulExecution"]["lastScheduleExecutionDate"]
   LAST_SCHEDULE_EXECUTION_TIMEZONE = context["lastSuccessfulExecution"]["lastScheduleExecutionTimezone"]

if LAST_SCHEDULE_EXECUTION_DATE and LAST_SCHEDULE_EXECUTION_TIMEZONE:
   timezone = zoneinfo.ZoneInfo(last_schedule_execution_timezone )
   date = datetime.fromisoformat(last_schedule_execution_date ).astimezone(timezone)  

Especificación del formato de los datos de salida

El punto de entrada, que se define en el archivo lógico, devuelve la salida del proceso de transformación de datos. El formato de salida debe ser un objeto Python DataFrame , que es una estructura de datos para manipular y almacenar datos en Python. La salida del proceso de transformación de datos en el archivo lógico debe ser pandas o polars, que son los dos módulos DataFrame . En el siguiente ejemplo, se configura el objeto DataFrame :

   def output(event_list): 
      return pd.DataFrame(event_list) 
   def execute(context): 
      extract_and_transform(context) 
      return output(event_list)

En el ejemplo, debe cambiar las funciones según sus requisitos para el archivo lógico.

Tratamiento de excepciones en el código Python

Para informar al usuario de la aplicación de procesos sobre credenciales no válidas o pérdida de conexión con un sistema externo, puede configurar el código Python para que envíe una excepción dedicada denominada ProcessAppException en la interfaz de usuario de la aplicación de procesos. El siguiente fragmento de código ofrece un ejemplo de configuración de ProcessAppException :

   from process_app import ProcessAppException 
   def execute(context): 
      # Example raising a ProcessAppException 
      raise ProcessAppException("cannot connect to the system")

Paso siguiente

Creación de aplicaciones de proceso personalizadas