diff --git a/benchmarks/600.workflows/695.protein-screen/definition.json b/benchmarks/600.workflows/695.protein-screen/definition.json new file mode 100644 index 000000000..6c112465c --- /dev/null +++ b/benchmarks/600.workflows/695.protein-screen/definition.json @@ -0,0 +1,26 @@ +{ + "root": "generate", + "states": { + "generate": { + "type": "task", + "func_name": "generate", + "next": "evaluate-map" + }, + "evaluate-map": { + "type": "map", + "root": "evaluate", + "array": "schedules", + "next": "select", + "states": { + "evaluate": { + "type": "task", + "func_name": "evaluate" + } + } + }, + "select": { + "type": "task", + "func_name": "select" + } + } +} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/config.json b/benchmarks/600.workflows/723.gpu-content-moderation/config.json new file mode 100644 index 000000000..a11a12730 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 240, + "memory": 4096, + "languages": ["python"], + "modules": [] +} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/definition.json b/benchmarks/600.workflows/723.gpu-content-moderation/definition.json new file mode 100644 index 000000000..52b7a9605 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/definition.json @@ -0,0 +1,26 @@ +{ + "root": "ingest", + "states": { + "ingest": { + "type": "task", + "func_name": "ingest", + "next": "moderate-map" + }, + "moderate-map": { + "type": "map", + "root": "moderate", + "array": "posts", + "next": "review", + "states": { + "moderate": { + "type": "task", + "func_name": "moderate" + } + } + }, + "review": { + "type": "task", + "func_name": "review" + } + } +} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/input.py b/benchmarks/600.workflows/723.gpu-content-moderation/input.py new file mode 100644 index 000000000..c018849a0 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/input.py @@ -0,0 +1,26 @@ +size_generators = { + "test": (30, 100), # 30 posts, avg 100 tokens per post + "small": (100, 200), + "large": (250, 300), +} + + +def buckets_count(): + # No object storage buckets required for this workflow. + return (0, 0) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + n_posts, avg_tokens = size_generators[size] + return { + "n_posts": n_posts, + "avg_tokens": avg_tokens, + } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py new file mode 100644 index 000000000..0d15910ae --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py @@ -0,0 +1,147 @@ +import uuid +import numpy as np + + +POST_CATEGORIES = [ + "news", + "personal", + "commercial", + "entertainment", + "educational", + "political", + "social", +] +LANGUAGES = ["en", "es", "fr", "de", "pt", "it"] + +SAFE_WORDS = [ + "cat", + "dog", + "food", + "travel", + "music", + "art", + "game", + "tech", + "book", + "movie", + "friend", + "family", + "work", + "love", + "happy", + "great", + "amazing", + "beautiful", + "fun", + "enjoy", + "today", + "yesterday", + "tomorrow", + "share", + "post", + "comment", + "like", + "follow", + "update", +] + +BORDERLINE_WORDS = [ + "debate", + "opinion", + "protest", + "controversial", + "argument", + "disagree", + "conflict", + "political", + "election", + "policy", + "criticism", + "oppose", + "challenge", +] + +UNSAFE_WORDS = [ + "hate", + "violence", + "spam", + "scam", + "fake", + "attack", + "threat", + "abuse", + "harass", + "derogatory", + "discriminate", + "offensive", + "vulgar", + "explicit", +] + + +def generate_text_content(avg_tokens: int, seed: int, violation_prob: float = 0.2) -> tuple: + rng = np.random.default_rng(seed=seed) + + has_violation = rng.random() < violation_prob + + n_tokens = max(10, int(rng.normal(avg_tokens, avg_tokens * 0.3))) + + if has_violation: + violation_type = rng.choice( + ["hate_speech", "violence", "spam", "misinformation", "harassment"], + p=[0.25, 0.2, 0.3, 0.15, 0.1], + ) + + unsafe_ratio = rng.uniform(0.15, 0.4) + n_unsafe = int(n_tokens * unsafe_ratio) + n_safe = n_tokens - n_unsafe + + words = list(rng.choice(UNSAFE_WORDS, size=n_unsafe, replace=True)) + list( + rng.choice(SAFE_WORDS, size=n_safe, replace=True) + ) + else: + borderline_ratio = rng.uniform(0, 0.2) + n_borderline = int(n_tokens * borderline_ratio) + n_safe = n_tokens - n_borderline + + words = list(rng.choice(BORDERLINE_WORDS, size=n_borderline, replace=True)) + list( + rng.choice(SAFE_WORDS, size=n_safe, replace=True) + ) + violation_type = None + + rng.shuffle(words) + text = " ".join(words) + + return text, violation_type, n_tokens + + +def handler(event): + n_posts = int(event["n_posts"]) + avg_tokens = int(event["avg_tokens"]) + batch_id = event.get("batch_id", str(uuid.uuid4())[:8]) + + posts = [] + for idx in range(n_posts): + rng = np.random.default_rng(seed=idx) + + category = rng.choice(POST_CATEGORIES) + language = rng.choice(LANGUAGES) + + text, violation_type, n_tokens = generate_text_content(avg_tokens, seed=idx) + + post = { + "post_id": f"post-{batch_id}-{idx:04d}", + "category": category, + "language": language, + "text": text, + "text_tokens": n_tokens, + "timestamp": 1700000000 + idx * 60, + "user_id": f"user_{rng.integers(1000, 9999)}", + "engagement_score": float(rng.beta(2, 5)), # Simulates likes/shares + "batch_id": batch_id, + "true_violation": violation_type, # Ground truth for evaluation + } + + posts.append(post) + + return {"posts": posts} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py new file mode 100644 index 000000000..c00815d9c --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py @@ -0,0 +1,206 @@ +import numpy as np + + +UNSAFE_KEYWORDS = [ + "hate", + "violence", + "spam", + "scam", + "fake", + "attack", + "threat", + "abuse", + "harass", +] +BORDERLINE_KEYWORDS = ["controversial", "debate", "protest", "argument"] + + +def tokenize_and_embed(text: str, seed: int) -> np.ndarray: + words = text.lower().split() + n_tokens = len(words) + + rng = np.random.default_rng(seed=seed) + embedding_dim = 768 + + embeddings = rng.normal(0, 0.1, (n_tokens, embedding_dim)).astype(np.float32) + + for i, word in enumerate(words): + if any(unsafe in word for unsafe in UNSAFE_KEYWORDS): + embeddings[i] += rng.normal(2.0, 0.5, embedding_dim).astype(np.float32) + elif any(border in word for border in BORDERLINE_KEYWORDS): + embeddings[i] += rng.normal(0.5, 0.3, embedding_dim).astype(np.float32) + + sentence_embedding = embeddings.mean(axis=0) + + return sentence_embedding + + +def transformer_classification(embedding: np.ndarray, seed: int) -> dict: + rng = np.random.default_rng(seed=seed) + + # Simulate classification head (linear layer + softmax) + embedding_norm = np.linalg.norm(embedding) + + # Higher norm suggests more extreme content (either very safe or very toxic) + base_score = 1.0 / (1.0 + np.exp(-embedding_norm / 10)) + + noise = rng.normal(0, 0.05) + toxicity_score = np.clip(base_score + noise, 0, 1) + + return { + "toxicity_score": float(toxicity_score), + "embedding_norm": float(embedding_norm), + } + + +def classify_violation_type(text: str, toxicity_score: float, seed: int) -> list: + rng = np.random.default_rng(seed=seed) + violations = [] + + words = text.lower() + + if "hate" in words or "discriminate" in words or "derogatory" in words: + confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append( + { + "type": "hate_speech", + "confidence": float(confidence), + } + ) + + if "violence" in words or "attack" in words or "threat" in words: + confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append( + { + "type": "violence", + "confidence": float(confidence), + } + ) + + if "spam" in words or "scam" in words: + confidence = min(0.9, toxicity_score + rng.uniform(0.05, 0.15)) + violations.append( + { + "type": "spam", + "confidence": float(confidence), + } + ) + + if "fake" in words: + confidence = min(0.85, toxicity_score + rng.uniform(0.05, 0.15)) + violations.append( + { + "type": "misinformation", + "confidence": float(confidence), + } + ) + + if "harass" in words or "abuse" in words: + confidence = min(0.9, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append( + { + "type": "harassment", + "confidence": float(confidence), + } + ) + + return violations + + +def sentiment_analysis(text: str, seed: int) -> dict: + rng = np.random.default_rng(seed=seed) + + positive_words = ["happy", "love", "great", "amazing", "beautiful", "enjoy", "fun"] + negative_words = ["hate", "violence", "abuse", "offensive", "threat"] + + words = text.lower().split() + pos_count = sum(1 for w in words if any(p in w for p in positive_words)) + neg_count = sum(1 for w in words if any(n in w for n in negative_words)) + + total = len(words) + if total > 0: + sentiment_score = (pos_count - neg_count) / total + else: + sentiment_score = 0.0 + + # Add noise + sentiment_score += rng.normal(0, 0.1) + sentiment_score = np.clip(sentiment_score, -1, 1) + + return { + "sentiment_score": float(sentiment_score), + "polarity": "positive" + if sentiment_score > 0.1 + else "negative" + if sentiment_score < -0.1 + else "neutral", + } + + +def calculate_spam_score(post: dict) -> float: + spam_score = 0.0 + + text = post.get("text", "") + words = text.lower().split() + + unique_words = len(set(words)) + if unique_words < len(words) * 0.3: + spam_score += 0.4 + + if "spam" in text or "scam" in text: + spam_score += 0.5 + + engagement = post.get("engagement_score", 0.5) + if engagement < 0.1: + spam_score += 0.2 + + return min(spam_score, 1.0) + + +def handler(post): + post_id = post["post_id"] + text = post.get("text", "") + category = post.get("category", "unknown") + + seed = hash(post_id) % 100000 + + embedding = tokenize_and_embed(text, seed) + + toxicity_result = transformer_classification(embedding, seed + 1) + toxicity_score = toxicity_result["toxicity_score"] + + violations = classify_violation_type(text, toxicity_score, seed + 2) + + sentiment = sentiment_analysis(text, seed + 3) + + spam_score = calculate_spam_score(post) + + overall_score = ( + 0.5 * toxicity_score + + 0.3 * spam_score + + 0.2 * (1.0 if sentiment["sentiment_score"] < -0.3 else 0.0) + ) + + if overall_score > 0.75: + action = "REMOVE" + elif overall_score > 0.5: + action = "REVIEW" + elif overall_score > 0.3: + action = "FLAG" + else: + action = "APPROVE" + + if spam_score > 0.7: + action = "REMOVE" + + return { + "post_id": post_id, + "category": category, + "toxicity_score": float(toxicity_score), + "spam_score": float(spam_score), + "sentiment": sentiment, + "violations": violations, + "overall_score": float(overall_score), + "action": action, + "true_violation": post.get("true_violation"), + } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py new file mode 100644 index 000000000..113f0f3b4 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py @@ -0,0 +1,123 @@ +def calculate_accuracy(posts: list) -> dict: + true_violations = [p for p in posts if p.get("true_violation") is not None] + + detected_violations = [p for p in posts if p.get("action") in ["REMOVE", "REVIEW"]] + + true_positives = len( + [ + p + for p in posts + if p.get("true_violation") is not None and p.get("action") in ["REMOVE", "REVIEW"] + ] + ) + + false_positives = len( + [ + p + for p in posts + if p.get("true_violation") is None and p.get("action") in ["REMOVE", "REVIEW"] + ] + ) + + false_negatives = len( + [ + p + for p in posts + if p.get("true_violation") is not None and p.get("action") not in ["REMOVE", "REVIEW"] + ] + ) + + precision = true_positives / len(detected_violations) if detected_violations else 0.0 + recall = true_positives / len(true_violations) if true_violations else 0.0 + f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0 + + return { + "precision": round(precision, 4), + "recall": round(recall, 4), + "f1_score": round(f1, 4), + "true_positives": true_positives, + "false_positives": false_positives, + "false_negatives": false_negatives, + "total_actual_violations": len(true_violations), + } + + +def handler(event): + posts = event.get("posts", []) + + action_counts = {} + for post in posts: + action = post.get("action", "UNKNOWN") + action_counts[action] = action_counts.get(action, 0) + 1 + + violation_counts = {} + for post in posts: + for violation in post.get("violations", []): + vtype = violation["type"] + violation_counts[vtype] = violation_counts.get(vtype, 0) + 1 + + category_stats = {} + for post in posts: + category = post.get("category", "unknown") + toxicity = post.get("toxicity_score", 0) + + if category not in category_stats: + category_stats[category] = { + "count": 0, + "avg_toxicity": 0.0, + "removed": 0, + } + + category_stats[category]["count"] += 1 + category_stats[category]["avg_toxicity"] += toxicity + if post.get("action") == "REMOVE": + category_stats[category]["removed"] += 1 + + for category in category_stats: + count = category_stats[category]["count"] + category_stats[category]["avg_toxicity"] = round( + category_stats[category]["avg_toxicity"] / count, 4 + ) + + needs_review = [ + { + "post_id": p["post_id"], + "category": p.get("category", "unknown"), + "toxicity_score": p.get("toxicity_score", 0), + "spam_score": p.get("spam_score", 0), + "violations": [v["type"] for v in p.get("violations", [])], + "sentiment": p.get("sentiment", {}).get("polarity", "neutral"), + } + for p in posts + if p.get("action") == "REVIEW" + ] + + needs_review.sort(key=lambda p: p["toxicity_score"], reverse=True) + + accuracy = calculate_accuracy(posts) + + avg_toxicity = sum(p.get("toxicity_score", 0) for p in posts) / len(posts) + avg_spam = sum(p.get("spam_score", 0) for p in posts) / len(posts) + + removal_rate = action_counts.get("REMOVE", 0) / len(posts) if posts else 0 + quality_level = "POOR" if removal_rate > 0.3 else "MODERATE" if removal_rate > 0.15 else "GOOD" + + sentiment_dist = {"positive": 0, "negative": 0, "neutral": 0} + for post in posts: + polarity = post.get("sentiment", {}).get("polarity", "neutral") + sentiment_dist[polarity] = sentiment_dist.get(polarity, 0) + 1 + + return { + "summary": f"Moderated {len(posts)} text posts", + "total_posts": len(posts), + "action_summary": action_counts, + "violation_summary": violation_counts, + "category_statistics": category_stats, + "sentiment_distribution": sentiment_dist, + "quality_level": quality_level, + "removal_rate": round(removal_rate, 4), + "avg_toxicity_score": round(avg_toxicity, 4), + "avg_spam_score": round(avg_spam, 4), + "posts_needing_review": needs_review[:20], # Top 20 + "accuracy_metrics": accuracy, + } diff --git a/sebs/sonataflow/__init__.py b/sebs/sonataflow/__init__.py new file mode 100644 index 000000000..44a4707a8 --- /dev/null +++ b/sebs/sonataflow/__init__.py @@ -0,0 +1,2 @@ +from .workflow import SonataFlowWorkflow # noqa: F401 +from .translator import definition_to_sonataflow, sonataflow_to_yaml # noqa: F401 diff --git a/sebs/sonataflow/translator.py b/sebs/sonataflow/translator.py new file mode 100644 index 000000000..35def83a1 --- /dev/null +++ b/sebs/sonataflow/translator.py @@ -0,0 +1,131 @@ +import json +from typing import Dict, List, Set, Tuple + +try: + import yaml +except ImportError: + yaml = None + + +class UnsupportedWorkflowError(Exception): + pass + + +def _collect_functions(funcs: Set[str], func_name: str): + if func_name: + funcs.add(func_name) + + +def _task_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + func_name = state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "operation", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _map_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + nested_states: Dict = state["states"] + root = state["root"] + if root not in nested_states: + raise UnsupportedWorkflowError(f"Map state {name} missing root state {root}") + root_state = nested_states[root] + func_name = root_state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "foreach", + "inputCollection": f"$.{state['array']}", + "iterationParam": "item", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _loop_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + func_name = state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "operation", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _linearize(definition: Dict) -> List[Tuple[str, Dict]]: + states = definition["states"] + cursor = definition["root"] + ordered: List[Tuple[str, Dict]] = [] + visited = set() + while cursor: + if cursor in visited: + raise UnsupportedWorkflowError(f"Cycle detected at state {cursor}") + visited.add(cursor) + state = states[cursor] + ordered.append((cursor, state)) + cursor = state.get("next") + return ordered + + +def definition_to_sonataflow(definition: Dict, name: str = "workflow") -> Tuple[Dict, Set[str]]: + """ + Convert SEBS workflow definition.json into a minimal SonataFlow (CNCF Serverless Workflow) + spec dictionary. Supports task/map/loop constructs used by the SEBS workflows. + Returns (spec, functions_used). + """ + if "root" not in definition or "states" not in definition: + raise UnsupportedWorkflowError("Invalid workflow definition") + + ordered_states = _linearize(definition) + funcs: Set[str] = set() + sonata_states: List[Dict] = [] + + builders = {"task": _task_state, "map": _map_state, "loop": _loop_state} + + for state_name, state in ordered_states: + stype = state["type"] + if stype not in builders: + raise UnsupportedWorkflowError(f"Unsupported state type {stype}") + sonata_states.append(builders[stype](state_name, state, funcs)) + + functions_block = [ + {"name": fn, "type": "custom", "operation": f"fn://{fn}"} for fn in sorted(funcs) + ] + spec = { + "id": name, + "name": name, + "version": "0.1", + "specVersion": "0.8", + "functions": functions_block, + "states": sonata_states, + } + return spec, funcs + + +def sonataflow_to_yaml(spec: Dict) -> str: + if yaml is None: + raise ImportError("PyYAML is required for YAML export") + return yaml.safe_dump(spec, sort_keys=False) + + +def definition_to_sonataflow_yaml(definition: Dict, name: str = "workflow") -> str: + spec, _ = definition_to_sonataflow(definition, name=name) + if yaml: + return sonataflow_to_yaml(spec) + return json.dumps(spec, indent=2) diff --git a/sebs/sonataflow/workflow.py b/sebs/sonataflow/workflow.py new file mode 100644 index 000000000..65d7fb916 --- /dev/null +++ b/sebs/sonataflow/workflow.py @@ -0,0 +1,47 @@ +from typing import Dict, List, Optional, Set + +from sebs.faas.function import FunctionConfig, Workflow + + +class SonataFlowWorkflow(Workflow): + """ + Lightweight wrapper that keeps track of the generated SonataFlow (CNCF Serverless Workflow) + spec for a SEBS benchmark workflow. Mirrors the structure of other providers' Workflow + classes so it can be cached/serialized the same way. + """ + + def __init__( + self, + name: str, + benchmark: str, + code_package_hash: str, + cfg: FunctionConfig, + sonataflow_spec: Dict, + functions: Optional[Set[str]] = None, + ): + super().__init__(benchmark, name, code_package_hash, cfg) + self.sonataflow_spec = sonataflow_spec + self.functions = list(functions or []) + + @staticmethod + def typename() -> str: + return "SonataFlow.Workflow" + + def serialize(self) -> dict: + return { + **super().serialize(), + "sonataflow_spec": self.sonataflow_spec, + "functions": self.functions, + } + + @staticmethod + def deserialize(cached_config: dict) -> "SonataFlowWorkflow": + cfg = FunctionConfig.deserialize(cached_config["config"]) + return SonataFlowWorkflow( + name=cached_config["name"], + benchmark=cached_config["benchmark"], + code_package_hash=cached_config["hash"], + cfg=cfg, + sonataflow_spec=cached_config["sonataflow_spec"], + functions=set(cached_config.get("functions", [])), + ) diff --git a/tools/sonataflow_builder.py b/tools/sonataflow_builder.py new file mode 100644 index 000000000..d33fa18e3 --- /dev/null +++ b/tools/sonataflow_builder.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Generate SonataFlow (CNCF Serverless Workflow) YAML along with Knative Service +manifests for SEBS workflows. This is a thin helper so users can go from +`definition.json` + function image mapping to runnable manifests under a +Quarkus SonataFlow operator on Knative. + +Usage: + python tools/sonataflow_builder.py \ + --definition benchmarks/600.workflows/6xx.OCR-pipeline/definition.json \ + --name ocr-pipeline \ + --namespace sebs \ + --image split=ghcr.io/example/split:latest \ + --image detect=ghcr.io/example/detect:latest \ + --image recognize=ghcr.io/example/recognize:latest \ + --image merge=ghcr.io/example/merge:latest \ + --out manifests.yaml +""" +import argparse +import json +import sys +from pathlib import Path +from typing import Dict, List + +from sebs.sonataflow.translator import definition_to_sonataflow_yaml, definition_to_sonataflow + + +KNATIVE_TEMPLATE = """\ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + name: {name} + namespace: {namespace} +spec: + template: + spec: + containers: + - image: {image} + env: + - name: FUNCTION_NAME + value: {name} +""" + +SONATAFLOW_CR_TEMPLATE = """\ +apiVersion: sonataflow.org/v1alpha08 +kind: SonataFlow +metadata: + name: {name} + namespace: {namespace} +spec: + flow: | +{workflow_yaml} +""" + + +def parse_images(image_args: List[str]) -> Dict[str, str]: + mapping: Dict[str, str] = {} + for item in image_args: + if "=" not in item: + raise ValueError(f"Image argument must be key=value, got: {item}") + fn, img = item.split("=", 1) + mapping[fn.strip()] = img.strip() + return mapping + + +def build_knative_services(namespace: str, images: Dict[str, str], functions: List[str]) -> List[str]: + yamls = [] + for fn in functions: + image = images.get(fn) + if not image: + raise ValueError(f"Missing image mapping for function: {fn}") + yamls.append(KNATIVE_TEMPLATE.format(name=fn, namespace=namespace, image=image)) + return yamls + + +def main(): + parser = argparse.ArgumentParser(description="Generate SonataFlow + Knative YAML for SEBS workflow.") + parser.add_argument("--definition", required=True, help="Path to SEBS workflow definition.json") + parser.add_argument("--name", required=True, help="Workflow name/id") + parser.add_argument("--namespace", default="default", help="Kubernetes namespace for resources") + parser.add_argument( + "--image", + action="append", + default=[], + help="Mapping of function=image (repeat for each function)", + ) + parser.add_argument("--out", default="-", help="Output file path or '-' for stdout") + + args = parser.parse_args() + + definition = json.loads(Path(args.definition).read_text()) + spec, functions = definition_to_sonataflow(definition, name=args.name) + images = parse_images(args.image) + + workflow_yaml = definition_to_sonataflow_yaml(definition, name=args.name) + knative_yamls = build_knative_services(args.namespace, images, sorted(functions)) + + pieces = [] + pieces.extend(knative_yamls) + pieces.append(SONATAFLOW_CR_TEMPLATE.format(name=args.name, namespace=args.namespace, workflow_yaml="\n".join(" " + line for line in workflow_yaml.splitlines()))) + output = "\n---\n".join(pieces) + + if args.out == "-" or args.out == "/dev/stdout": + sys.stdout.write(output) + else: + Path(args.out).write_text(output) + print(f"Wrote manifests to {args.out}") + + +if __name__ == "__main__": + main()