first functional flow

start visualization abstraction in container

Author: wassfila

.gitignore

@@ -0,0 +1,3 @@
+__pycache__
+venv
+/cache

README.md

@@ -1,2 +1,24 @@
 # pipeline-visualization
 how to render a pipeline flow with graph and time chart
+
+# usage
+to run the pipeline through
+
+    python runner.py
+
+testing containers e.g. graphviz rendering
+
+create `cache/test/example.dot` with content such as
+```dot
+digraph G {
+    A -> B;
+    B -> C;
+    C -> A;
+}
+```
+note cache is mapped in `/data` inside the container.
+
+Then generate the graph with
+
+    run graphviz test/example.dot
+

activate.cmd

@@ -0,0 +1 @@
+venv\Scripts\activate

containers/docker-compose.yaml

@@ -0,0 +1,7 @@
+services:
+  graphviz:
+    build:
+      context: ./dockerfiles
+      dockerfile: graphviz.Dockerfile
+    volumes:
+      - ../cache:/data

containers/dockerfiles/graphviz.Dockerfile

@@ -0,0 +1,12 @@
+# Use an official Ubuntu base image
+FROM ubuntu:latest
+
+# Install Graphviz
+RUN apt-get update && apt-get install -y graphviz
+
+# Set the working directory
+WORKDIR /data
+
+# Default command uses the `dot` tool to process a graph file
+ENTRYPOINT ["dot"]
+CMD ["-Tpng", "-o", "output.png"]  # Default command if no argument is provided

containers/run.py

@@ -0,0 +1,26 @@
+import sys
+import subprocess
+import os
+
+def graphviz(filename):
+    command_line = f"docker compose run --rm graphviz -Tsvg {filename} -o {filename}.svg"
+    subprocess.run(command_line.split(" "))
+    return
+
+def run(command):
+    # Path to the directory containing docker-compose.yml
+    docker_compose_dir = os.path.join(os.path.dirname(__file__))
+    # Change the working directory
+    os.chdir(docker_compose_dir)
+
+    if(command == "graphviz"):
+        filename = sys.argv[2]
+        graphviz(filename)
+
+if __name__ == '__main__':
+    # Default command if no argument is provided
+    command = "build"
+    # Check if an argument is provided and use it as the command
+    if len(sys.argv) > 1:
+        command = sys.argv[1]
+    run(command)

manifest.yaml

@@ -0,0 +1,9 @@
+# Stage > Job
+pipeline:
+  fetch:
+    fetch-data-1: pipeline.py#fetch
+  process:
+    calculate-functions: pipeline.py#calculate
+    compute-statistics: pipeline.py#compute
+  build:
+    generate-website: pipeline.py#build

pipeline.py

@@ -0,0 +1,25 @@
+import time
+import runner as run
+
+def fetch():
+    print("- fetching file")
+    time.sleep(0.1)
+    my_data = {
+        "hello":"world"
+    }
+    run.set_artifact(my_data,"fetch/content.json")
+    return
+
+def calculate():
+    print("- hello from calc2")
+    new_data = run.get_artifact("content")
+    print(new_data)
+    return
+
+def compute():
+    print("- hellor from comp3")
+    return
+
+def build():
+    print("- hi build4")
+    return

run.cmd

@@ -0,0 +1,2 @@
+@echo off
+python containers\run.py %*

runner.py

@@ -0,0 +1,71 @@
+import utils as utl
+import importlib
+from os.path import dirname,splitext,basename,join
+import state
+from datetime import datetime
+
+class ArtifactError(Exception):
+    """Custom exception for artifact management errors."""
+    pass
+
+def log_job(stage_name, job_name,start):
+    stop = datetime.now()
+    state.pipe.append({
+        "stage":stage_name,
+        "job":job_name,
+        "start":str(start),
+        "stop":str(stop),
+        "duration": str(stop-start),
+        "duration_text": utl.duration_text(stop-start)
+    })
+    return
+
+def set_artifact(data,filepath,type="generic"):
+    id,ext = splitext(basename(filepath))
+    if(id in state.artifacts):
+        raise ArtifactError(f"Artifact with ID '{id}' already exists.")
+    path = dirname(filepath)
+    state.artifacts[id] = {
+        "path":path,
+        "ext":ext,
+        "type":type,
+        "filepath":filepath
+    }
+    abs_filepath = join("cache",filepath)
+    if(ext == ".json"):
+        utl.save_json(data,abs_filepath)
+    return
+
+def get_artifact(id):
+    if(id not in state.artifacts):
+        raise ArtifactError(f"Artifact with ID '{id}' does not exist")
+    artifact = state.artifacts[id]
+    if(artifact["ext"] == ".json"):
+        return utl.load_json(join("cache",artifact["filepath"]))
+    return None
+
+def run_stage(stage_name, jobs):
+    print(f"Running stage: {stage_name}")
+    state.stage = stage_name
+    for job_name, job in jobs.items():
+        module_name, function_name = job.split('#')
+        state.job = job_name
+        state.step = function_name
+        module = importlib.import_module(module_name.replace('.py', ''))
+        func = getattr(module, function_name)
+        print(f"    Executing job: {job_name}")
+        start = datetime.now()
+        func()
+        log_job(stage_name, job_name,start)
+
+def run_pipeline(pipeline):
+    state.value = 1
+    for stage, jobs in pipeline.items():
+        run_stage(stage, jobs)
+    utl.save_json(state.artifacts,"cache/artifacts.json")
+    utl.save_json(state.pipe,"cache/pipeline.json")
+    
+
+if __name__ == '__main__':
+    manifest = utl.load_yaml("manifest.yaml")
+    run_pipeline(manifest["pipeline"])

state.py

@@ -0,0 +1,7 @@
+
+pipe = []
+artifacts = {}
+
+stage = ""
+job = ""
+step = ""

test-manifest.yaml

@@ -0,0 +1,8 @@
+- group: fetch
+  step: fetch-data-1
+- group: process
+  step: calculate-functions
+- group: process
+  step: compute-statistics
+- group: build
+  step: generate-website

utils.py

@@ -0,0 +1,54 @@
+import json
+import yaml
+from os import makedirs
+from os.path import dirname
+from datetime import timedelta
+
+def load_yaml(fileName):
+    with open(fileName, "r") as stream:
+        try:
+            return yaml.safe_load(stream)
+        except yaml.YAMLError as e:
+            print(e)
+    return
+
+def load_json(fileName):
+    return json.load(open(fileName,encoding='utf-8'))
+
+def save_json(data,fileName):
+    path = dirname(fileName)
+    makedirs(path, exist_ok=True)
+    jfile = open(fileName, "w")
+    jfile.write(json.dumps(data, indent=4))
+    jfile.close()
+    return
+
+def duration_text(duration: timedelta):
+    # Ensure that the duration is non-negative
+    if duration < timedelta(0):
+        duration = -duration
+    # Extract days, seconds, and microseconds from the timedelta object
+    days = duration.days
+    seconds = duration.seconds
+    microseconds = duration.microseconds
+    milliseconds = microseconds // 1000  # Convert microseconds to milliseconds
+
+    # Calculate hours, minutes, and seconds
+    hours = seconds // 3600
+    minutes = (seconds % 3600) // 60
+    seconds = seconds % 60
+
+    # Build the duration text string
+    text = ""
+    if days > 0:
+        text += f"{days} d "
+    if hours > 0:
+        text += f"{hours} h "
+    if minutes > 0:
+        text += f"{minutes} mn "
+    if seconds > 0:
+        text += f"{seconds} s "
+    if milliseconds > 0:
+        text += f"{milliseconds} ms "
+
+    return text.strip()  # Remove any trailing space