Merge pull request #87 from hal-lab-u-tokyo/refactor_analyze_json

Refactor execution scripts for modularity and enhanced validation

Author: takasehideki

README.md

@@ -31,10 +31,10 @@ Generated `host*_run.sh` and `local_run.sh` scripts support the runtime options
 ./host1_exec.sh
 
 # Override eval-time
-./host1_run.sh --eval-time 120
+./host1_run.sh --eval-time 60
 
 # Short options
-./host1_run.sh -t 120
+./host1_run.sh -t 60
 ```
 
 `--eval-time` is applied to all nodes launched through `*_run.sh` or `local_run.sh`. `payload_size` and `period_ms` are read from each Publisher/Intermediate entry in the topology JSON.
@@ -185,7 +185,7 @@ python3 performance_test/performance_test.py
 # Switch to native execution
 python3 performance_test/performance_test.py --exec-policy native
 # Override eval-time explicitly
-python3 performance_test/performance_test.py --eval-time 120
+python3 performance_test/performance_test.py --eval-time 60
 ```
 
 Main arguments:

manager_scripts/generate_exec/__init__.py

@@ -0,0 +1 @@
+"""Submodules for topology validation and execution-script generation."""

manager_scripts/generate_exec/metadata.py

@@ -0,0 +1,165 @@
+"""Metadata generation for created execution script sets."""
+
+import json
+import os
+import shlex
+import sys
+from datetime import datetime
+
+from .validation import normalize_intermediate_entries, require_positive_int
+
+
+def collect_metadata_node_names(json_content):
+    """Collect host names, node names, and topic counts for metadata.txt."""
+    host_names = []
+    publisher_names = []
+    subscriber_names = []
+    intermediate_names = []
+    topic_names = set()
+
+    for host_dict in json_content["hosts"]:
+        host_names.append(host_dict["host_name"])
+        for node in host_dict.get("nodes", []):
+            node_name = node["node_name"]
+            if node.get("publisher"):
+                publisher_names.append(node_name)
+                for publisher in node["publisher"]:
+                    topic_names.add(publisher["topic_name"])
+            if node.get("subscriber"):
+                subscriber_names.append(node_name)
+                for subscriber in node["subscriber"]:
+                    topic_names.add(subscriber["topic_name"])
+            if "intermediate" in node:
+                intermediate_names.append(node_name)
+                intermediate_entries = normalize_intermediate_entries(
+                    node["intermediate"], node_name
+                )
+                for intermediate_entry in intermediate_entries:
+                    for publisher in intermediate_entry.get("publisher", []):
+                        topic_names.add(publisher["topic_name"])
+                    for subscriber in intermediate_entry.get("subscriber", []):
+                        topic_names.add(subscriber["topic_name"])
+
+    return host_names, publisher_names, subscriber_names, intermediate_names, topic_names
+
+
+def collect_topic_runtime_config(json_content):
+    """Collect topic -> payload/period/publisher_count from topology."""
+    topic_cfg = {}
+
+    def add_publisher_topic(entry, context):
+        topic = entry.get("topic_name")
+        if not topic:
+            raise ValueError(f"{context}: missing topic_name")
+        payload_size = require_positive_int(entry, "payload_size", context)
+        period_ms = require_positive_int(entry, "period_ms", context)
+
+        cfg = topic_cfg.setdefault(
+            topic,
+            {
+                "payload_size": payload_size,
+                "period_ms": period_ms,
+                "publisher_count": 0,
+            },
+        )
+        if cfg["payload_size"] != payload_size or cfg["period_ms"] != period_ms:
+            raise ValueError(
+                f"Inconsistent payload/period for topic '{topic}' in topology JSON"
+            )
+        cfg["publisher_count"] += 1
+
+    for host in json_content.get("hosts", []):
+        for node in host.get("nodes", []):
+            node_name = node.get("node_name", "?")
+            for pub_idx, publisher in enumerate(node.get("publisher", []) or []):
+                add_publisher_topic(
+                    publisher, f"node '{node_name}' publisher[{pub_idx}]"
+                )
+
+            if "intermediate" in node:
+                intermediate_entries = normalize_intermediate_entries(
+                    node["intermediate"], node_name
+                )
+                for entry_idx, inter in enumerate(intermediate_entries):
+                    for pub_idx, publisher in enumerate(inter.get("publisher", []) or []):
+                        add_publisher_topic(
+                            publisher,
+                            f"node '{node_name}' intermediate[{entry_idx}] publisher[{pub_idx}]",
+                        )
+
+    return topic_cfg
+
+
+def unique_in_order(items):
+    """Remove duplicates while preserving the original order."""
+    return list(dict.fromkeys(items))
+
+
+def generate_metadata_file(
+    json_content, json_path, rmw, ws_dir, project_root, scenario_dir
+):
+    """Generate <ws-dir>/latest/metadata.txt."""
+    latest_dir = os.path.join(project_root, ws_dir, "latest")
+    metadata_path = os.path.join(latest_dir, "metadata.txt")
+
+    (
+        host_names,
+        publisher_names,
+        subscriber_names,
+        intermediate_names,
+        topic_names,
+    ) = collect_metadata_node_names(json_content)
+    host_names = unique_in_order(host_names)
+    publisher_names = unique_in_order(publisher_names)
+    subscriber_names = unique_in_order(subscriber_names)
+    intermediate_names = unique_in_order(intermediate_names)
+    topic_runtime_cfg = collect_topic_runtime_config(json_content)
+
+    all_nodes = [
+        node
+        for host in json_content["hosts"]
+        for node in host.get("nodes", [])
+    ]
+    node_count = len(all_nodes)
+
+    qos = json_content.get("qos", {})
+    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+    sections = [
+        [
+            "# --- 1. general info ---",
+            f"command: {shlex.join(sys.argv)}",
+            f"timestamp: {timestamp}",
+            f"json: {os.path.basename(json_path)}",
+            f"json_path: {json_path}",
+            f"ws_dir: {ws_dir}",
+            f"scenario_dir: {scenario_dir}",
+        ],
+        [
+            "# --- 2. test config ---",
+            f"rmw: {rmw}",
+            f"qos_history: {qos.get('history', 'KEEP_LAST')}",
+            f"qos_depth: {qos.get('depth', 1)}",
+            f"qos_reliability: {qos.get('reliability', 'RELIABLE')}",
+        ],
+        [
+            "# --- 3. topology stats ---",
+            f"host_count: {len(host_names)}",
+            f"node_count: {node_count}",
+            f"publisher_count: {len(publisher_names)}",
+            f"subscriber_count: {len(subscriber_names)}",
+            f"intermediate_count: {len(intermediate_names)}",
+            f"topic_count: {len(topic_names)}",
+            f"hosts: {', '.join(host_names)}",
+            f"publishers: {', '.join(publisher_names)}",
+            f"subscribers: {', '.join(subscriber_names)}",
+            f"intermediates: {', '.join(intermediate_names)}",
+            f"topics: {', '.join(sorted(topic_names))}",
+            (
+                "topic_runtime_json: "
+                f"{json.dumps(topic_runtime_cfg, separators=(',', ':'), sort_keys=True)}"
+            ),
+        ],
+    ]
+
+    with open(metadata_path, "w") as f:
+        f.write("\n\n".join("\n".join(section) for section in sections) + "\n")

manager_scripts/generate_exec/paths.py

@@ -0,0 +1,95 @@
+"""Filesystem/path operations for output generation."""
+
+import os
+import shutil
+import sys
+
+
+def clear_directory_contents(path):
+    """Delete everything directly under the given directory."""
+    for name in os.listdir(path):
+        target = os.path.join(path, name)
+        if os.path.islink(target) or os.path.isfile(target):
+            os.remove(target)
+        elif os.path.isdir(target):
+            shutil.rmtree(target)
+
+
+def read_existing_json_path(run_dir):
+    """Return the json_path field from metadata.txt in run_dir if it exists."""
+    metadata_path = os.path.join(run_dir, "metadata.txt")
+    if not os.path.isfile(metadata_path):
+        return None
+    with open(metadata_path) as f:
+        for line in f:
+            if line.startswith("json_path:"):
+                return os.path.normpath(line.split(":", 1)[1].strip())
+    return None
+
+
+def confirm_overwrite(output_dir, force=False, existing_json_path=None, new_json_path=None):
+    """Ask whether an existing exec_scripts directory should be overwritten."""
+    if force:
+        return True
+    if not sys.stdin.isatty():
+        raise SystemExit(
+            f"Error: '{output_dir}' already exists and stdin is not a TTY. "
+            "Use --force (-f) to overwrite without confirmation."
+        )
+    msg = f"'{output_dir}' already exists."
+    if (
+        existing_json_path is not None
+        and new_json_path is not None
+    ):
+        existing_normalized = os.path.normpath(existing_json_path)
+        new_normalized = os.path.normpath(new_json_path)
+        if existing_normalized != new_normalized:
+            msg += (
+                f"\n  WARNING: The existing scripts were generated from '{existing_normalized}',"
+                f"\n           but the current input is '{new_normalized}'."
+                f"\n  Same filename, different path -- are you sure you want to overwrite?"
+            )
+    msg += " Overwrite generated files? [y/N]: "
+    while True:
+        answer = input(msg).strip().lower()
+        if answer in ("y", "yes"):
+            return True
+        if answer in ("", "n", "no"):
+            return False
+        print("Please answer yes or no.")
+
+
+def update_latest_symlink(base_dir, target_name):
+    """Update <ws-dir>/latest to point to target_name."""
+    latest_link = os.path.join(base_dir, "latest")
+    if os.path.lexists(latest_link):
+        if os.path.islink(latest_link) or os.path.isfile(latest_link):
+            os.remove(latest_link)
+        elif os.path.isdir(latest_link):
+            shutil.rmtree(latest_link)
+    os.symlink(target_name, latest_link)
+
+
+def resolve_output_paths(json_path, rmw, ws_dir, force=False):
+    """Resolve and prepare output directory paths and the latest alias."""
+    project_root = os.getcwd()
+    perf_ws_dir = os.path.join(project_root, ws_dir)
+    os.makedirs(perf_ws_dir, exist_ok=True)
+
+    json_basename = os.path.splitext(os.path.basename(json_path))[0]
+    scenario_dir = f"{json_basename}-{rmw}"
+    run_dir = os.path.join(perf_ws_dir, scenario_dir)
+    output_dir = os.path.join(run_dir, "exec_scripts")
+
+    overwrite = os.path.isdir(output_dir)
+    if overwrite:
+        existing_json_path = read_existing_json_path(run_dir)
+        if not confirm_overwrite(
+            output_dir,
+            force=force,
+            existing_json_path=existing_json_path,
+            new_json_path=json_path,
+        ):
+            raise SystemExit("Canceled by user. No files were generated.")
+
+    return project_root, output_dir, scenario_dir, overwrite