XES to CSV and back conversion

Author: iharsuvorau

.gitignore

@@ -127,3 +127,5 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+.idea/

README.md

@@ -1 +1,10 @@
-# pm4py-wrapper
+# pm4py-wrapper
+
+Wrapper for pm4py. 
+
+## Usage
+
+```shell
+$ pm4py_wrapper -p log.xes xes_to_csv
+$ pm4py_wrapper -p log.csv xes_to_xes
+```

setup.py

@@ -0,0 +1,21 @@
+from setuptools import setup, find_packages
+
+setup(
+    name='pm4py-wrapper',
+    version='1.0.0',
+    packages=find_packages(where='src'),
+    package_dir={"": "src"},
+    include_package_data=True,
+    install_requires=[
+        'click',
+        'pm4py',
+        'pandas',
+        'pytest',
+        'setuptools',
+    ],
+    entry_points={
+        'console_scripts': [
+            'pm4py_wrapper = pm4py_wrapper.cli:main',
+        ]
+    }
+)

src/pm4py_wrapper/cli.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+import click
+from click import BaseCommand
+
+from pm4py_wrapper.wrapper import convert_xes_to_csv, convert_csv_to_xes
+
+
+@click.group()
+@click.option('-i', '--input_log', default=None, required=True, type=Path, help='Path to the input event log.')
+@click.option('-o', '--output_dir', default=Path('.'), type=Path, help='Path to the output directory.')
+def main(input_log, output_dir):
+    # This is a main group which includes other commands specified below.
+    pass
+
+
+@main.command()
+@click.pass_context
+def xes_to_csv(ctx):
+    log_path: Path = ctx.parent.params['input_log']
+    output_dir: Path = ctx.parent.params['output_dir']
+    output_path = output_dir / log_path.with_suffix('.csv').name
+    convert_xes_to_csv(log_path, output_path)
+
+
+@main.command()
+@click.pass_context
+def csv_to_xes(ctx):
+    log_path: Path = ctx.parent.params['input_log']
+    output_dir: Path = ctx.parent.params['output_dir']
+    output_path = output_dir / log_path.with_suffix('.xes').name
+    convert_csv_to_xes(log_path, output_path)
+
+
+if __name__ == '__main__':
+    main()

src/pm4py_wrapper/wrapper.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+
+import pandas as pd
+import pm4py
+from pm4py.objects.conversion.log import converter as log_converter
+from pm4py.objects.log.util import interval_lifecycle
+from pm4py.objects.log.exporter.xes import exporter as xes_exporter
+
+
+def convert_xes_to_csv(xes_path: Path, output_path: Path):
+    log = pm4py.read_xes(str(xes_path))
+    log_interval = interval_lifecycle.to_interval(log)
+    df = log_converter.apply(log_interval, variant=log_converter.Variants.TO_DATA_FRAME)
+    df.to_csv(output_path, index=False)
+
+
+def convert_csv_to_xes(csv_path: Path, output_path: Path):
+    df = pd.read_csv(csv_path)
+    log = log_converter.apply(df, variant=log_converter.Variants.TO_EVENT_LOG)
+    xes_exporter.apply(log, str(output_path))