dataflow-analysis/trace-analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "import re\n",
    "import math\n",
    "import graphviz as gv\n",
    "from tqdm import tqdm\n",
    "from bisect import bisect_right\n",
    "from termcolor import colored\n",
    "\n",
    "import matplotlib.patches as mpatch\n",
    "from scipy import stats\n",
    "from cycler import cycler\n",
    "\n",
    "import glob\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from matplotlib import pyplot as plt\n",
    "\n",
    "from misc.utils import cached, parse_as\n",
    "\n",
    "%matplotlib inline"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "##################################################\n",
    "# User Settings\n",
    "##################################################\n",
    "# Change these to influence the execution of the\n",
    "# notebook.\n",
    "# You can override these from the command line\n",
    "# by defining environment variables with the\n",
    "# name of the constants below, prefixed with\n",
    "# \"ANA_NB_\".\n",
    "# For example, the environment variable\n",
    "# \"ANA_NB_TR_PATH\" will override the \"TR_PATH\"\n",
    "# setting below.\n",
    "##################################################\n",
    "\n",
    "# The path to the build folder of a ROS2 workspace that contains the\n",
    "# tracetools_read and tracetools_analysis folders.\n",
    "TRACING_WS_BUILD_PATH = \"~/Projects/autoware/build\"\n",
    "\n",
    "# Path to trace directory (e.g. ~/.ros/my-trace/ust) or to a converted trace file.\n",
    "# Using the path \"/ust\" at the end is optional but greatly reduces processing time\n",
    "# if kernel traces are also present.\n",
    "TR_PATH = \"data/trace-awsim-x86/ust\"\n",
    "\n",
    "# Path to the folder all artifacts from this notebook are saved to.\n",
    "# This entails plots as well as data tables.\n",
    "OUT_PATH = \"out/\"\n",
    "\n",
    "# Whether to cache the results of long computations per set of inputs\n",
    "CACHING_ENABLED = False\n",
    "\n",
    "# Whether to annotate topics/publications with bandwidth/message size\n",
    "BW_ENABLED = False\n",
    "# Path to a results folder as output by ma-hw-perf-tools/messages/record.bash\n",
    "# Used to annotate message sizes in E2E latency calculations\n",
    "BW_PATH = \"../ma-hw-perf-tools/data/results\"\n",
    "\n",
    "# Whether to use dependencies extracted by the Clang-tools to supplement\n",
    "# automatic node-internal data flow annotations.\n",
    "# If in doubt, set to False.\n",
    "CL_ENABLED = False\n",
    "# Path to the output directory of the ROS2 dependency checker.\n",
    "# Will only be used if CL_ENABLED is True.\n",
    "CL_PATH = \"~/Projects/llvm-project/clang-tools-extra/ros2-internal-dependency-checker/output\"\n",
    "\n",
    "# Whether to compute data flow graphs.\n",
    "# If you are only interested in E2E latencies, set this to False\n",
    "DFG_ENABLED = False\n",
    "# Whether to plot data flow graphs (ignored if DFG_ENABLED is False)\n",
    "DFG_PLOT = False\n",
    "\n",
    "# The maximum node namespace hierarchy level to be plotted.\n",
    "# Top-level (1): e.g. /sensing, /control, etc.\n",
    "# Level 3: e.g. /sensing/lidar/pointcloud_processor\n",
    "DFG_MAX_HIER_LEVEL = 100\n",
    "\n",
    "# RegEx pattern for nodes that shall be marked as system inputs\n",
    "# These will be plotted with a start arrow as known from automata diagrams\n",
    "DFG_INPUT_NODE_PATTERNS = [r\"^/sensing\"]\n",
    "# RegEx pattern for nodes that shall be marked as system outputs\n",
    "# These will be plotted with a double border\n",
    "DFG_OUTPUT_NODE_PATTERNS = [r\"^/awapi\", r\"^/control/external_cmd_converter\", \"emergency\"]\n",
    "# RegEx for nodes which shall not be plotted in the DFG\n",
    "DFG_EXCL_NODE_PATTERNS = [r\"^/rviz2\", r\"transform_listener_impl\"]\n",
    "\n",
    "# Whether to compute E2E latencies.\n",
    "E2E_ENABLED = True\n",
    "# Whether to plot end-to-end latency information (ignored if E2E_ENABLED is False)\n",
    "E2E_PLOT = True\n",
    "# The index of the output message that shall be used in plots that visualize a specific\n",
    "# message dependency tree. This index has to be 0 <= n < #output messages\n",
    "E2E_PLOT_TIMESTAMP = 1000\n",
    "# E2E latency threshold. Every E2E latency higher than this is discarded.\n",
    "# Set this as low as comfortably possible to speed up calculations.\n",
    "# WARNING: If you set this too low (i.e. to E2E latencies that plausibly can happen)\n",
    "# your results will be wrong)\n",
    "E2E_TIME_LIMIT_S = 2\n",
    "\n",
    "# All topics containing any of these RegEx patterns are considered output topics in E2E latency calculations\n",
    "# E.g. r\"^/control/\" will cover all control topics\n",
    "E2E_OUTPUT_TOPIC_PATTERNS = [r\"^/control/.*?/control_cmd\"]\n",
    "# All topics containing any of these RegEx patterns are considered input topics in E2E latency calculations\n",
    "# E.g. r\"^/sensing/\" will cover all sensing topics\n",
    "E2E_INPUT_TOPIC_PATTERNS = [r\"^/vehicle/status/\", r\"^/sensing/\"]\n",
    "\n",
    "# This code overrides the above constants with environment variables, do not edit.\n",
    "for env_key, env_value in os.environ.items():\n",
    "    if env_key.startswith(\"ANA_NB_\"):\n",
    "        key = env_key.removeprefix(\"ANA_NB_\")\n",
    "        if key not in globals().keys():\n",
    "            continue\n",
    "        value = parse_as(type(globals()[key]), env_value)\n",
    "        globals()[key] = value\n",
    "\n",
    "\n",
    "# Convert input paths to absolute paths\n",
    "def _expand_path(path):\n",
    "    return os.path.realpath(os.path.expandvars(os.path.expanduser(path)))\n",
    "\n",
    "\n",
    "TRACING_WS_BUILD_PATH = _expand_path(TRACING_WS_BUILD_PATH)\n",
    "TR_PATH = _expand_path(TR_PATH)\n",
    "OUT_PATH = _expand_path(OUT_PATH)\n",
    "BW_PATH = _expand_path(BW_PATH)\n",
    "CL_PATH = _expand_path(CL_PATH)\n",
    "\n",
    "os.makedirs(OUT_PATH, exist_ok=True)\n",
    "\n",
    "print(\"User Settings:\")\n",
    "for k, v in globals().copy().items():\n",
    "    if not k.isupper():\n",
    "        continue\n",
    "    print(f\"  {k:.<40s} := {v}\")"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "sys.path.append(os.path.join(TRACING_WS_BUILD_PATH, \"tracetools_read/\"))\n",
    "sys.path.append(os.path.join(TRACING_WS_BUILD_PATH, \"tracetools_analysis/\"))\n",
    "from tracetools_read.trace import *\n",
    "from tracetools_analysis.loading import load_file\n",
    "from tracetools_analysis.processor.ros2 import Ros2Handler\n",
    "\n",
    "from tracing_interop.tr_types import *\n",
    "from latency_graph.message_tree import DepTree\n",
    "from matching.subscriptions import sanitize"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Organize Trace Data"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "def _load_traces():\n",
    "    file = load_file(TR_PATH)\n",
    "    handler = Ros2Handler.process(file)\n",
    "    return TrContext(handler)\n",
    "\n",
    "\n",
    "_tracing_context = cached(\"tr_objects\", _load_traces, [TR_PATH], not CACHING_ENABLED)\n",
    "_tr_globals = [\"nodes\", \"publishers\", \"subscriptions\", \"timers\", \"timer_node_links\", \"subscription_objects\",\n",
    "               \"callback_objects\", \"callback_symbols\", \"publish_instances\", \"callback_instances\", \"topics\"]\n",
    "\n",
    "# Help the IDE recognize those identifiers\n",
    "nodes = publishers = subscriptions = timers = timer_node_links = subscription_objects = callback_objects = callback_symbols = publish_instances = callback_instances = topics = None\n",
    "\n",
    "for name in _tr_globals:\n",
    "    globals()[name] = getattr(_tracing_context, name)\n",
    "\n",
    "print(\"Done.\")"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "for topic in sorted(topics, key=lambda t: t.name):\n",
    "    topic: TrTopic\n",
    "    print(f\"{topic.name:.<120s} | {sum(map(lambda p: len(p.instances), topic.publishers))}\")"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "# E2E Latency Calculation"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "from latency_graph import latency_graph as lg\n",
    "\n",
    "\n",
    "def _make_latency_graph():\n",
    "    return lg.LatencyGraph(_tracing_context)\n",
    "\n",
    "\n",
    "lat_graph = cached(\"lat_graph\", _make_latency_graph, [TR_PATH], not CACHING_ENABLED)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false DFG_ENABLED\n",
    "%%skip_if_false DFG_PLOT\n",
    "\n",
    "#################################################\n",
    "# Plot DFG\n",
    "#################################################\n",
    "\n",
    "# Compare with: https://autowarefoundation.github.io/autoware-documentation/main/design/autoware-architecture/node-diagram/\n",
    "node_colors = {\n",
    "        \"sensing\": {\"fill\": \"#e1d5e7\", \"stroke\": \"#9673a6\"},\n",
    "        \"localization\": {\"fill\": \"#dae8fc\", \"stroke\": \"#6c8ebf\"},\n",
    "        \"perception\": {\"fill\": \"#d5e8d4\", \"stroke\": \"#82b366\"},\n",
    "        \"planning\": {\"fill\": \"#fff2cc\", \"stroke\": \"#d6b656\"},\n",
    "        \"control\": {\"fill\": \"#ffe6cc\", \"stroke\": \"#d79b00\"},\n",
    "        \"system\": {\"fill\": \"#f8cecc\", \"stroke\": \"#b85450\"},\n",
    "        \"vehicle_interface\": {\"fill\": \"#b0e3e6\", \"stroke\": \"#0e8088\"},\n",
    "        None: {\"fill\": \"#f5f5f5\", \"stroke\": \"#666666\"}\n",
    "}\n",
    "\n",
    "node_namespace_mapping = {\n",
    "        'perception': 'perception',\n",
    "        'sensing': 'sensing',\n",
    "        'plannitokenizeng': 'planning',\n",
    "        'control': 'control',\n",
    "        'awapi': None,\n",
    "        'autoware_api': None,\n",
    "        'map': None,\n",
    "        'system': 'system',\n",
    "        'localization': 'localization',\n",
    "        'robot_state_publisher': None,\n",
    "        'aggregator_node': None,\n",
    "        'pointcloud_container': 'sensing',\n",
    "}\n",
    "\n",
    "\n",
    "g = gv.Digraph('G', filename=\"latency_graph.gv\",\n",
    "               node_attr={'shape': 'plain'},\n",
    "               graph_attr={'pack': '1'})\n",
    "g.graph_attr['rankdir'] = 'LR'\n",
    "\n",
    "\n",
    "def plot_hierarchy(gv_parent, lg_node: lg.LGHierarchyLevel, **subgraph_kwargs):\n",
    "    if lg_node.name == \"[NONE]\":\n",
    "        return\n",
    "\n",
    "    print(f\"{'  ' * lg_node.full_name.count('/')}Processing {lg_node.name}: {len(lg_node.callbacks)}\")\n",
    "    with gv_parent.subgraph(name=f\"cluster_{lg_node.full_name.replace('/', '__')}\", **subgraph_kwargs) as c:\n",
    "        c.attr(label=lg_node.name)\n",
    "        for cb in lg_node.callbacks:\n",
    "            if isinstance(cb, lg.LGTrCallback):\n",
    "                tr_cb = cb.cb\n",
    "                try:\n",
    "                    sym = _tracing_context.callback_symbols.by_id.get(tr_cb.callback_object)\n",
    "                    pretty_sym = repr(sanitize(sym.symbol))\n",
    "                except KeyError:\n",
    "                    pretty_sym = cb.name\n",
    "                except TypeError:\n",
    "                    pretty_sym = cb.name\n",
    "            else:\n",
    "                pretty_sym = cb.name\n",
    "\n",
    "            pretty_sym = pretty_sym.replace(\"&\", \"&amp;\").replace(\"<\", \"&lt;\").replace(\">\", \"&gt;\")\n",
    "\n",
    "            c.node(cb.id(),\n",
    "                   f'<<table BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\"><tr><td port=\"in\"></td><td>{pretty_sym}</td><td port=\"out\"></td></tr></table>>')\n",
    "\n",
    "        for ch in lg_node.children:\n",
    "            plot_hierarchy(c, ch, **subgraph_kwargs)\n",
    "\n",
    "\n",
    "def plot_lg(graph: lg.LatencyGraph):\n",
    "    for top_level_node in graph.top_node.children:\n",
    "        colors = node_colors[node_namespace_mapping.get(top_level_node.name)]\n",
    "        plot_hierarchy(g, top_level_node, graph_attr={'bgcolor': colors[\"fill\"], 'pencolor': colors[\"stroke\"]})\n",
    "\n",
    "    for edge in graph.edges:\n",
    "        g.edge(f\"{edge.start.id()}:out\", f\"{edge.end.id()}:in\")\n",
    "\n",
    "\n",
    "plot_lg(lat_graph)\n",
    "\n",
    "g.save(\"latency_graph.gv\")\n",
    "g.render(\"latency_graph.svg\")\n",
    "\n",
    "g"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false DFG_ENABLED\n",
    "%%skip_if_false DFG_PLOT\n",
    "\n",
    "\n",
    "##################################################\n",
    "# Compute in/out topics for hierarchy level X\n",
    "##################################################\n",
    "\n",
    "def get_nodes_on_level(lat_graph: lg.LatencyGraph):\n",
    "    def _traverse_node(node: lg.LGHierarchyLevel, cur_lvl=0):\n",
    "        if cur_lvl == DFG_MAX_HIER_LEVEL:\n",
    "            return [node]\n",
    "\n",
    "        if not node.children and cur_lvl < DFG_MAX_HIER_LEVEL:\n",
    "            return [node]\n",
    "\n",
    "        collected_nodes = []\n",
    "        for ch in node.children:\n",
    "            collected_nodes += _traverse_node(ch, cur_lvl + 1)\n",
    "        return collected_nodes\n",
    "\n",
    "    return _traverse_node(lat_graph.top_node)\n",
    "\n",
    "\n",
    "lvl_nodes = get_nodes_on_level(lat_graph)\n",
    "lvl_nodes = [n for n in lvl_nodes if not any(re.search(p, n.full_name) for p in DFG_EXCL_NODE_PATTERNS)]\n",
    "\n",
    "input_nodes = [n.full_name for n in lvl_nodes if any(re.search(p, n.full_name) for p in DFG_INPUT_NODE_PATTERNS)]\n",
    "output_nodes = [n.full_name for n in lvl_nodes if any(re.search(p, n.full_name) for p in DFG_OUTPUT_NODE_PATTERNS)]\n",
    "\n",
    "print(', '.join(map(lambda n: n, input_nodes)))\n",
    "print(', '.join(map(lambda n: n, output_nodes)))\n",
    "print(', '.join(map(lambda n: n.full_name, lvl_nodes)))\n",
    "\n",
    "\n",
    "def _collect_callbacks(n: lg.LGHierarchyLevel):\n",
    "    callbacks = []\n",
    "    callbacks += n.callbacks\n",
    "    for ch in n.children:\n",
    "        callbacks += _collect_callbacks(ch)\n",
    "    return callbacks\n",
    "\n",
    "\n",
    "cb_to_node_map = {}\n",
    "for n in lvl_nodes:\n",
    "    cbs = _collect_callbacks(n)\n",
    "    for cb in cbs:\n",
    "        cb_to_node_map[cb.id()] = n\n",
    "\n",
    "edges_between_nodes = {}\n",
    "for edge in lat_graph.edges:\n",
    "    from_node = cb_to_node_map.get(edge.start.id())\n",
    "    to_node = cb_to_node_map.get(edge.end.id())\n",
    "\n",
    "    if from_node is None or to_node is None:\n",
    "        continue\n",
    "\n",
    "    if from_node.full_name == to_node.full_name:\n",
    "        continue\n",
    "\n",
    "    k = (from_node.full_name, to_node.full_name)\n",
    "\n",
    "    if k not in edges_between_nodes:\n",
    "        edges_between_nodes[k] = []\n",
    "\n",
    "    edges_between_nodes[k].append(edge)\n",
    "\n",
    "g = gv.Digraph('G', filename=\"latency_graph.gv\",\n",
    "               node_attr={'shape': 'plain'},\n",
    "               graph_attr={'pack': '1'})\n",
    "g.graph_attr['rankdir'] = 'LR'\n",
    "\n",
    "for n in lvl_nodes:\n",
    "    colors = node_colors[node_namespace_mapping.get(n.full_name.strip(\"/\").split(\"/\")[0])]\n",
    "    peripheries = \"1\" if n.full_name not in output_nodes else \"2\"\n",
    "    g.node(n.full_name, label=n.full_name, fillcolor=colors[\"fill\"], color=colors[\"stroke\"],\n",
    "           shape=\"box\", style=\"filled\", peripheries=peripheries)\n",
    "\n",
    "    if n.full_name in input_nodes:\n",
    "        helper_node_name = f\"{n.full_name}__before\"\n",
    "        g.node(helper_node_name, label=\"\", shape=\"none\", height=\"0\", width=\"0\")\n",
    "        g.edge(helper_node_name, n.full_name)\n",
    "\n",
    "\n",
    "def compute_e2e_paths(start_nodes, end_nodes, edges):\n",
    "    frontier_paths = [[n] for n in start_nodes]\n",
    "    final_paths = []\n",
    "\n",
    "    while frontier_paths:\n",
    "        frontier_paths_new = []\n",
    "\n",
    "        for path in frontier_paths:\n",
    "            head = path[-1]\n",
    "            if head in end_nodes:\n",
    "                final_paths.append(path)\n",
    "                continue\n",
    "\n",
    "            out_nodes = [n_to for n_from, n_to in edges if n_from == head if n_to not in path]\n",
    "            new_paths = [path + [n] for n in out_nodes]\n",
    "            frontier_paths_new += new_paths\n",
    "\n",
    "        frontier_paths = frontier_paths_new\n",
    "\n",
    "    final_paths = [[(n_from, n_to)\n",
    "                    for n_from, n_to in zip(path[:-1], path[1:])]\n",
    "                   for path in final_paths]\n",
    "    return final_paths\n",
    "\n",
    "\n",
    "e2e_paths = compute_e2e_paths(input_nodes, output_nodes, edges_between_nodes)\n",
    "\n",
    "for (src_name, dst_name), edges in edges_between_nodes.items():\n",
    "    print(src_name, dst_name, len(edges))\n",
    "    color = \"black\" if any((src_name, dst_name) in path for path in e2e_paths) else \"tomato\"\n",
    "    g.edge(src_name, dst_name, penwidth=str(math.log(len(edges)) * 2 + .2), color=color)\n",
    "\n",
    "g.save(\"level_graph.gv\")\n",
    "g.render(\"level_graph.svg\")\n",
    "\n",
    "g"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "\n",
    "\n",
    "def inst_get_dep_msg(inst: TrCallbackInstance):\n",
    "    if inst.callback_object not in _tracing_context.callback_objects.by_callback_object:\n",
    "        # print(\"Callback not found (2)\")\n",
    "        return None\n",
    "\n",
    "    if not isinstance(inst.callback_obj.owner, TrSubscriptionObject):\n",
    "        # print(f\"Wrong type: {type(inst.callback_obj.owner)}\")\n",
    "        return None\n",
    "\n",
    "    sub_obj: TrSubscriptionObject = inst.callback_obj.owner\n",
    "    if sub_obj and sub_obj.subscription and sub_obj.subscription.topic:\n",
    "        # print(f\"Subscription has no topic\")\n",
    "        pubs = sub_obj.subscription.topic.publishers\n",
    "    else:\n",
    "        pubs = []\n",
    "\n",
    "    def _pub_latest_msg_before(pub: TrPublisher, inst):\n",
    "        i_latest_msg = bisect_right(pub.instances, inst.timestamp, key=lambda x: x.timestamp) - 1\n",
    "        if i_latest_msg < 0 or i_latest_msg >= len(pub.instances):\n",
    "            return None\n",
    "        latest_msg = pub.instances[i_latest_msg]\n",
    "        if latest_msg.timestamp >= inst.timestamp:\n",
    "            return None\n",
    "\n",
    "        return latest_msg\n",
    "\n",
    "    msgs = [_pub_latest_msg_before(pub, inst) for pub in pubs]\n",
    "    msgs = [msg for msg in msgs if msg is not None]\n",
    "    msgs.sort(key=lambda i: i.timestamp, reverse=True)\n",
    "    if msgs:\n",
    "        msg = msgs[0]\n",
    "        return msg\n",
    "\n",
    "    # print(f\"No messages found for topic {sub_obj.subscription.topic}\")\n",
    "    return None\n",
    "\n",
    "\n",
    "def inst_get_dep_insts(inst: TrCallbackInstance):\n",
    "    if inst.callback_object not in _tracing_context.callback_objects.by_callback_object:\n",
    "        # print(\"Callback not found\")\n",
    "        return []\n",
    "    dep_cbs = get_cb_dep_cbs(inst.callback_obj)\n",
    "\n",
    "    def _cb_to_chronological_inst(cb: TrCallbackObject, inst):\n",
    "        i_inst_latest = bisect_right(cb.callback_instances, inst.timestamp, key=lambda x: x.timestamp)\n",
    "\n",
    "        for inst_before in cb.callback_instances[i_inst_latest::-1]:\n",
    "            if lg.inst_runtime_interval(inst_before)[-1] < inst.timestamp:\n",
    "                return inst_before\n",
    "\n",
    "        return None\n",
    "\n",
    "    insts = [_cb_to_chronological_inst(cb, inst) for cb in dep_cbs]\n",
    "    insts = [inst for inst in insts if inst is not None]\n",
    "    return insts\n",
    "\n",
    "\n",
    "def get_cb_dep_cbs(cb: TrCallbackObject):\n",
    "    match cb.owner:\n",
    "        case TrSubscriptionObject() as sub_obj:\n",
    "            sub_obj: TrSubscriptionObject\n",
    "            owner = sub_obj.subscription.node\n",
    "        case TrTimer() as tmr:\n",
    "            tmr: TrTimer\n",
    "            owner = tmr.node\n",
    "        case _:\n",
    "            raise RuntimeError(f\"Encountered {cb.owner} as callback owner\")\n",
    "\n",
    "    owner: TrNode\n",
    "    dep_sub_objs = {sub.subscription_object for sub in owner.subscriptions}\n",
    "    dep_cbs = {callback_objects.by_id.get(sub_obj.id) for sub_obj in dep_sub_objs if sub_obj is not None}\n",
    "    dep_cbs |= {callback_objects.by_id.get(tmr.id) for tmr in owner.timers}\n",
    "    dep_cbs.discard(cb)\n",
    "    dep_cbs.discard(None)\n",
    "\n",
    "    return dep_cbs\n",
    "\n",
    "\n",
    "def get_msg_dep_cb(msg: TrPublishInstance):\n",
    "    \"\"\"\n",
    "    For a given message instance `msg`, find the publishing callback,\n",
    "    as well as the message instances that callback depends on (transitively within its TrNode).\n",
    "    \"\"\"\n",
    "\n",
    "    # Find CB instance that published msg\n",
    "    # print(f\"PUB {msg.publisher.node.path if msg.publisher.node is not None else '??'} ==> {msg.publisher.topic_name}\")\n",
    "    pub_cbs = lat_graph.pub_cbs.get(msg.publisher)\n",
    "    if pub_cbs is None:\n",
    "        # print(\"Publisher unknown to lat graph. Skipping.\")\n",
    "        return None\n",
    "\n",
    "    # print(f\"Found {len(pub_cbs)} pub cbs\")\n",
    "    cb_inst_candidates = []\n",
    "    for cb in pub_cbs:\n",
    "        # print(f\"  > CB ({len(cb.callback_instances)} instances): {cb.callback_symbol.symbol if cb.callback_symbol else cb.id}\")\n",
    "        i_inst_after = bisect_right(cb.callback_instances, msg.timestamp, key=lambda x: x.timestamp)\n",
    "\n",
    "        for inst in cb.callback_instances[:i_inst_after]:\n",
    "            inst_start, inst_end = lg.inst_runtime_interval(inst)\n",
    "            if msg.timestamp > inst_end:\n",
    "                continue\n",
    "\n",
    "            assert inst_start <= msg.timestamp <= inst_end\n",
    "\n",
    "            cb_inst_candidates.append(inst)\n",
    "\n",
    "    if len(cb_inst_candidates) > 1:\n",
    "        # print(\"Found multiple possible callbacks\")\n",
    "        return None\n",
    "    if not cb_inst_candidates:\n",
    "        # print(\"Found no possible callbacks\")\n",
    "        return None\n",
    "\n",
    "    dep_inst = cb_inst_candidates[0]\n",
    "    return dep_inst\n",
    "\n",
    "\n",
    "def get_dep_tree(inst: TrPublishInstance | TrCallbackInstance, lvl=0, visited_topics=None, is_dep_cb=False,\n",
    "                 start_time=None):\n",
    "    if visited_topics is None:\n",
    "        visited_topics = set()\n",
    "\n",
    "    if start_time is None:\n",
    "        start_time = inst.timestamp\n",
    "\n",
    "    if inst.timestamp - start_time > E2E_TIME_LIMIT_S:\n",
    "        return None\n",
    "\n",
    "    children_are_dep_cbs = False\n",
    "\n",
    "    match inst:\n",
    "        case TrPublishInstance(publisher=pub):\n",
    "            if pub.topic_name in visited_topics:\n",
    "                return None\n",
    "\n",
    "            visited_topics.add(pub.topic_name)\n",
    "            deps = [get_msg_dep_cb(inst)]\n",
    "        case TrCallbackInstance() as cb_inst:\n",
    "            deps = [inst_get_dep_msg(cb_inst)]\n",
    "            if not is_dep_cb:\n",
    "                deps += inst_get_dep_insts(cb_inst)\n",
    "                children_are_dep_cbs = True\n",
    "        case _:\n",
    "            print(\n",
    "                f\"[WARN] Expected inst to be of type TrPublishInstance or TrCallbackInstance, got {type(inst).__name__}\")\n",
    "            return None\n",
    "    #print(\"Rec level\", lvl)\n",
    "    deps = [dep for dep in deps if dep is not None]\n",
    "    deps = [get_dep_tree(dep, lvl + 1, set(visited_topics), children_are_dep_cbs, start_time) for dep in deps]\n",
    "    deps = [dep for dep in deps if dep is not None]\n",
    "    return DepTree(inst, deps)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "\n",
    "end_topics = [t for t in _tracing_context.topics if any(re.search(f, t.name) for f in E2E_OUTPUT_TOPIC_PATTERNS)]\n",
    "\n",
    "def build_dep_trees():\n",
    "    all_trees = []\n",
    "    for end_topic in end_topics:\n",
    "        end_topic: TrTopic\n",
    "        print(f\"====={end_topic.name}\")\n",
    "\n",
    "        pubs = end_topic.publishers\n",
    "        for pub in pubs:\n",
    "            msgs = list(pub.instances)\n",
    "            for msg in tqdm(msgs, desc=\"Processing output messages\"):\n",
    "                msg: TrPublishInstance\n",
    "                tree = get_dep_tree(msg)\n",
    "                all_trees.append(tree)\n",
    "    return all_trees\n",
    "\n",
    "try:\n",
    "    trees = cached(\"trees\", build_dep_trees, [TR_PATH], not CACHING_ENABLED)\n",
    "except Exception as e:\n",
    "    import traceback\n",
    "    print(e)\n",
    "    traceback.print_exc()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "%%skip_if_false BW_ENABLED\n",
    "\n",
    "\n",
    "\n",
    "def parse_bytes(string):\n",
    "    match string[-1]:\n",
    "        case 'K':\n",
    "            exponent = 1e3\n",
    "        case 'M':\n",
    "            exponent = 1e6\n",
    "        case _:\n",
    "            exponent = 1\n",
    "\n",
    "    num = float(string.split(\" \")[0])\n",
    "    return num * exponent\n",
    "\n",
    "\n",
    "def bytes_str(bytes):\n",
    "    if bytes >= 1024 ** 2:\n",
    "        return f\"{bytes / (1024 ** 2):.2f} MiB\"\n",
    "    if bytes >= 1024:\n",
    "        return f\"{bytes / 1024:.2f} KiB\"\n",
    "    return f\"{bytes:.0f} B\"\n",
    "\n",
    "\n",
    "bw_files = glob.glob(os.path.join(BW_PATH, \"*.log\"))\n",
    "msg_sizes = {}\n",
    "for bw_file in bw_files:\n",
    "    with open(bw_file) as f:\n",
    "        lines = f.readlines()\n",
    "        topic = os.path.splitext(os.path.split(bw_file)[1])[0].replace(\"__\", \"/\")\n",
    "\n",
    "        if not lines or re.match(f\"^\\s*$\", lines[-1]):\n",
    "            #print(f\"No data for {topic}\")\n",
    "            continue\n",
    "\n",
    "        line_pattern = re.compile(\n",
    "            r\"(?P<bw>[0-9.]+ [KM]?)B/s from (?P<n_msgs>[0-9.]+) messages --- Message size mean: (?P<mean>[0-9.]+ [KM]?)B min: (?P<min>[0-9.]+ [KM]?)B max: (?P<max>[0-9.]+ [KM]?)B\\n\")\n",
    "        m = re.fullmatch(line_pattern, lines[-1])\n",
    "        if m is None:\n",
    "            print(f\"Line could not be parsed in {topic}: '{lines[-1]}'\")\n",
    "            continue\n",
    "\n",
    "        msg_sizes[topic] = {'bw': parse_bytes(m.group(\"bw\")),\n",
    "                            'min': parse_bytes(m.group(\"min\")),\n",
    "                            'mean': parse_bytes(m.group(\"mean\")),\n",
    "                            'max': parse_bytes(m.group(\"max\"))}\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "\n",
    "def leaf_topics(tree: DepTree, lvl=0):\n",
    "    ret_list = []\n",
    "    match tree.head:\n",
    "        case TrPublishInstance(publisher=pub):\n",
    "            if pub:\n",
    "                ret_list += [(lvl, pub.topic_name)]\n",
    "    ret_list += [(lvl, None)]\n",
    "\n",
    "    for dep in tree.deps:\n",
    "        ret_list += leaf_topics(dep, lvl + 1)\n",
    "    return ret_list\n",
    "\n",
    "\n",
    "def all_e2es(tree: DepTree, t_start=None):\n",
    "    if t_start is None:\n",
    "        t_start = tree.head.timestamp\n",
    "\n",
    "    if not tree.deps:\n",
    "        return [t_start - tree.head.timestamp]\n",
    "\n",
    "    ret_list = []\n",
    "    for dep in tree.deps:\n",
    "        ret_list += all_e2es(dep, t_start)\n",
    "    return ret_list\n",
    "\n",
    "\n",
    "def relevant_e2es(tree: DepTree, input_topic_patterns, t_start=None, path=None):\n",
    "    if t_start is None:\n",
    "        t_start = tree.head.timestamp\n",
    "\n",
    "    if path is None:\n",
    "        path = []\n",
    "\n",
    "    latency = t_start - tree.head.timestamp\n",
    "    if latency > E2E_TIME_LIMIT_S:\n",
    "        return []\n",
    "\n",
    "    new_path = [tree.head] + path\n",
    "\n",
    "    if not tree.deps:\n",
    "        match tree.head:\n",
    "            case TrPublishInstance(publisher=pub):\n",
    "                if pub and any(re.search(f, pub.topic_name) for f in input_topic_patterns):\n",
    "                    return [(latency, new_path)]\n",
    "\n",
    "    ret_list = []\n",
    "    for dep in tree.deps:\n",
    "        ret_list += relevant_e2es(dep, input_topic_patterns, t_start, new_path)\n",
    "    return ret_list\n",
    "\n",
    "\n",
    "e2ess = []\n",
    "e2e_pathss = []\n",
    "for tree in trees:\n",
    "    e2es, e2e_paths = zip(*relevant_e2es(tree, E2E_INPUT_TOPIC_PATTERNS))\n",
    "    e2ess.append(e2es)\n",
    "    e2e_pathss.append(e2e_paths)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "#from matplotlib.animation import FuncAnimation\n",
    "#from IPython import display\n",
    "\n",
    "#fig, ax = plt.subplots(figsize=(16, 9))\n",
    "#ax: plt.Axes\n",
    "#ax.set_xlim(0, 4)\n",
    "\n",
    "#ax.hist([], bins=200, range=(0, 4), histtype='stepfilled')\n",
    "#ax.set_title(\"Time: 0.000000s\")\n",
    "\n",
    "#def anim(frame):\n",
    "#    print(frame, end='\\r')\n",
    "#    ax.clear()\n",
    "#    ax.hist(e2es[frame], bins=200, range=(0, 4), histtype='stepfilled')\n",
    "#    ax.set_title(f\"Time: {(trees[frame].head.timestamp - trees[0].head.timestamp):.6}s\")\n",
    "\n",
    "\n",
    "#anim_created = FuncAnimation(fig, anim, min(len(trees), 10000), interval=16, repeat_delay=200)\n",
    "\n",
    "#video = anim_created.save(\"anim.mp4\", dpi=120)\n",
    "\n",
    "#for tree in trees:\n",
    "#    path = tree.critical_path(start_topic_filters)\n",
    "#    for i, inst in enumerate(path[::-1]):\n",
    "#        match inst:\n",
    "#            case TrPublishInstance(publisher=pub):\n",
    "#                print(f\"  {i:>3d}: T\", pub.topic_name)\n",
    "#            case TrCallbackInstance(callback_obj=cb):\n",
    "#                match cb.owner:\n",
    "#                    case TrSubscriptionObject(subscription=sub):\n",
    "#                        node = sub.node\n",
    "#                    case TrTimer() as tmr:\n",
    "#                        node = tmr.node\n",
    "#                    case _:\n",
    "#                        raise ValueError(f\"Callback owner type not recognized: {type(cb.owner).__name__}\")\n",
    "#\n",
    "#                print(f\"  {i:>3d}: N\", node.path)\n",
    "#    print(\"==================\")\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "%%skip_if_false E2E_PLOT\n",
    "\n",
    "\n",
    "fig, ax = plt.subplots(figsize=(18, 9), num=\"e2e_plot\")\n",
    "DS = 1\n",
    "times = [tree.head.timestamp - trees[0].head.timestamp for tree in trees[::DS]]\n",
    "\n",
    "ax2 = ax.twinx()\n",
    "ax2.plot(times, list(map(lambda paths: sum(map(len, paths)) / len(paths), e2e_pathss[::DS])), color=\"orange\")\n",
    "ax2.fill_between(times,\n",
    "                 list(map(lambda paths: min(map(len, paths)), e2e_pathss[::DS])),\n",
    "                 list(map(lambda paths: max(map(len, paths)), e2e_pathss[::DS])),\n",
    "                 alpha=.3, color=\"orange\")\n",
    "\n",
    "ax.plot(times, [np.mean(e2es) for e2es in e2ess[::DS]])\n",
    "ax.fill_between(times, [np.min(e2es) for e2es in e2ess[::DS]], [np.max(e2es) for e2es in e2ess[::DS]], alpha=.3)\n",
    "\n",
    "\n",
    "def scatter_topic(topic_name, y=0, **scatter_kwargs):\n",
    "    for pub in topics.by_name[topic_name].publishers:\n",
    "        if not pub:\n",
    "            continue\n",
    "\n",
    "        inst_timestamps = [inst.timestamp - trees[0].head.timestamp for inst in pub.instances if\n",
    "                           inst.timestamp >= trees[0].head.timestamp]\n",
    "        scatter_kwargs_default = {\"marker\": \"x\", \"color\": \"indianred\"}\n",
    "        scatter_kwargs_default.update(scatter_kwargs)\n",
    "        ax.scatter(inst_timestamps, np.full(len(inst_timestamps), fill_value=y), **scatter_kwargs_default)\n",
    "\n",
    "\n",
    "scatter_topic(\"/autoware/engage\")\n",
    "scatter_topic(\"/planning/scenario_planning/parking/trajectory\", y=-.04, color=\"cadetblue\")\n",
    "scatter_topic(\"/planning/scenario_planning/lane_driving/trajectory\", y=-.08, color=\"darkgreen\")\n",
    "\n",
    "ax.set_xlabel(\"Simulation time [s]\")\n",
    "ax.set_ylabel(\"End-to-End latency [s]\")\n",
    "ax2.set_ylabel(\"End-to-End path length\")\n",
    "None"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "\n",
    "def critical_path(self):\n",
    "    if not self.deps:\n",
    "        return [self.head]\n",
    "    return [self.head, *min(map(critical_path, self.deps), key=lambda ls: ls[-1].timestamp)]\n",
    "\n",
    "\n",
    "def e2e_lat(self):\n",
    "    return self.head.timestamp - critical_path(self)[-1].timestamp\n",
    "\n",
    "\n",
    "def get_relevant_tree(tree: DepTree, accept_leaf=lambda x: True, root=None):\n",
    "    if root is None:\n",
    "        root = tree.head\n",
    "    if not tree.deps:\n",
    "        if accept_leaf(tree.head, root):\n",
    "            return tree\n",
    "        return None\n",
    "\n",
    "    relevant_deps = [get_relevant_tree(dep, accept_leaf, root) for dep in tree.deps]\n",
    "    if not any(relevant_deps):\n",
    "        return None\n",
    "\n",
    "    return DepTree(tree.head, [dep for dep in relevant_deps if dep])\n",
    "\n",
    "\n",
    "def fanout(self):\n",
    "    if not self.deps:\n",
    "        return 1\n",
    "\n",
    "    return sum(map(fanout, self.deps))\n",
    "\n",
    "\n",
    "def sort_subtree(subtree: DepTree, sort_func=lambda t: t.head.timestamp - e2e_lat(t)):\n",
    "    subtree.deps.sort(key=sort_func)\n",
    "    for dep in subtree.deps:\n",
    "        sort_subtree(dep, sort_func)\n",
    "    return subtree\n",
    "\n",
    "\n",
    "def _leaf_filter(inst, root):\n",
    "    if root.timestamp - inst.timestamp > E2E_TIME_LIMIT_S:\n",
    "        return False\n",
    "\n",
    "    match inst:\n",
    "        case TrPublishInstance(publisher=pub):\n",
    "            return pub and any(re.search(f, pub.topic_name) for f in E2E_INPUT_TOPIC_PATTERNS)\n",
    "    return False\n",
    "\n",
    "\n",
    "relevant_trees = [get_relevant_tree(tree, _leaf_filter) for tree in trees]\n",
    "relevant_trees = [t for t in relevant_trees if t]"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "%%skip_if_false E2E_PLOT\n",
    "\n",
    "def dict_safe_append(dictionary, key, value):\n",
    "    if key not in dictionary:\n",
    "        dictionary[key] = []\n",
    "    dictionary[key].append(value)\n",
    "\n",
    "\n",
    "fig, (ax, ax_rel) = plt.subplots(2, 1, sharex=True, figsize=(60, 30), num=\"crit_plot\")\n",
    "ax.set_prop_cycle(cycler('color', [plt.cm.nipy_spectral(i / 4) for i in range(5)]))\n",
    "ax_rel.set_prop_cycle(cycler('color', [plt.cm.nipy_spectral(i / 4) for i in range(5)]))\n",
    "\n",
    "critical_paths = [critical_path(tree) for tree in relevant_trees[::DS]]\n",
    "\n",
    "time_breakdown = {}\n",
    "\n",
    "for path in critical_paths:\n",
    "    tmr_cb_calc_time = 0.0\n",
    "    sub_cb_calc_time = 0.0\n",
    "    tmr_cb_relevant_time = 0.0\n",
    "    sub_cb_relevant_time = 0.0\n",
    "    dds_time = 0.0\n",
    "    idle_time = 0.0\n",
    "\n",
    "    last_pub_time = None\n",
    "    last_cb_time = None\n",
    "    for inst in path:\n",
    "        match inst:\n",
    "            case TrPublishInstance(timestamp=t):\n",
    "                assert last_pub_time is None, \"Two publication without callback inbetween\"\n",
    "\n",
    "                if last_cb_time is not None:\n",
    "                    dds_time += last_cb_time - t\n",
    "\n",
    "                last_pub_time = t\n",
    "                last_cb_time = None\n",
    "            case TrCallbackInstance(callback_obj=cb, timestamp=t, duration=d):\n",
    "                if last_pub_time is not None:\n",
    "                    assert last_pub_time <= t + d, \"Publication out of CB instance timeframe\"\n",
    "\n",
    "                    match cb.owner:\n",
    "                        case TrTimer():\n",
    "                            tmr_cb_calc_time += d\n",
    "                            tmr_cb_relevant_time += last_pub_time - t\n",
    "                        case TrSubscriptionObject():\n",
    "                            sub_cb_calc_time += d\n",
    "                            sub_cb_relevant_time += last_pub_time - t\n",
    "                elif last_cb_time is not None:\n",
    "                    idle_time += last_cb_time - (t + d)\n",
    "                last_pub_time = None\n",
    "                last_cb_time = t\n",
    "\n",
    "    dict_safe_append(time_breakdown, \"Timer CB\", tmr_cb_relevant_time)\n",
    "    dict_safe_append(time_breakdown, \"Subscription CB\", sub_cb_relevant_time)\n",
    "    dict_safe_append(time_breakdown, \"DDS\", dds_time)\n",
    "    dict_safe_append(time_breakdown, \"Idle\", idle_time)\n",
    "\n",
    "time_breakdown = {k: np.array(v) for k, v in time_breakdown.items()}\n",
    "\n",
    "timer_cb_times = [sum(inst.duration for inst in path if\n",
    "                      isinstance(inst, TrCallbackInstance) and isinstance(inst.callback_obj, TrTimer)) for path in\n",
    "                  critical_paths]\n",
    "sub_cb_times = [sum(inst.duration for inst in path if isinstance(inst, TrCallbackInstance)) for path in critical_paths]\n",
    "\n",
    "labels, values = list(zip(*time_breakdown.items()))\n",
    "\n",
    "#ax.plot(range(len(relevant_trees[::DS])), [e2e_lat(tree) for tree in relevant_trees[::DS]], label=\"Total E2E\")\n",
    "ax.stackplot(range(len(relevant_trees[::DS])), values, labels=labels)\n",
    "ax.legend()\n",
    "ax.set_title(\"End-to-End Latency Breakdown\")\n",
    "ax.set_ylabel(\"End-to-End Latency [s]\")\n",
    "\n",
    "timestep_mags = np.array([sum(vs) for vs in zip(*values)])\n",
    "ax_rel.stackplot(range(len(relevant_trees[::DS])), [val / timestep_mags for val in values], labels=labels)\n",
    "ax_rel.set_title(\"End-to-End Latency Breakdown (relative)\")\n",
    "ax_rel.set_ylabel(\"End-to-End Latency Fraction\")\n",
    "ax_rel.set_xlabel(\"Timestep\")\n",
    "ax_rel.legend()\n",
    "\n",
    "None"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "%%skip_if_false E2E_PLOT\n",
    "\n",
    "fig, ax = plt.subplots(figsize=(60, 15), num=\"crit_pdf\")\n",
    "ax.set_prop_cycle(cycler('color', [plt.cm.nipy_spectral(i / 4) for i in range(5)]))\n",
    "\n",
    "kde = stats.gaussian_kde(timestep_mags)\n",
    "xs = np.linspace(timestep_mags.min(), timestep_mags.max(), 1000)\n",
    "ax.plot(xs, kde(xs), label=\"End-to-End Latency\")\n",
    "perc = 90\n",
    "ax.axvline(np.percentile(timestep_mags, perc), label=f\"{perc}th percentile\")\n",
    "\n",
    "ax2 = ax.twinx()\n",
    "ax2.hist(timestep_mags, 200)\n",
    "ax2.set_ylim(0, ax2.get_ylim()[1])\n",
    "\n",
    "ax.set_title(\"Time Distribution for E2E Breakdown\")\n",
    "ax.set_xlabel(\"Time [s]\")\n",
    "ax.set_ylabel(\"Frequency\")\n",
    "ax.set_xlim(0, 2.01)\n",
    "ax.set_ylim(0, ax.get_ylim()[1])\n",
    "ax.legend()\n",
    "\n",
    "None"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "%%skip_if_false E2E_PLOT\n",
    "\n",
    "tree = relevant_trees[E2E_PLOT_TIMESTAMP]\n",
    "e2es = e2ess[E2E_PLOT_TIMESTAMP]\n",
    "e2e_paths = e2e_pathss[E2E_PLOT_TIMESTAMP]\n",
    "margin_y = .2\n",
    "margin_x = 0\n",
    "arr_width = 1 - margin_y\n",
    "\n",
    "\n",
    "def cb_str(inst: TrCallbackInstance):\n",
    "    cb: TrCallbackObject = inst.callback_obj\n",
    "    if not cb:\n",
    "        return None\n",
    "    ret_str = f\"- {inst.duration * 1e3:07.3f}ms \"\n",
    "    #if cb.callback_symbol:\n",
    "    #    ret_str = repr(sanitize(cb.callback_symbol.symbol))\n",
    "\n",
    "    match cb.owner:\n",
    "        case TrSubscriptionObject(subscription=sub):\n",
    "            sub: TrSubscription\n",
    "            ret_str = f\"{ret_str}{sub.node.path if sub.node else None} <- {sub.topic_name}\"\n",
    "        case TrTimer(period=p, node=node):\n",
    "            p: int\n",
    "            node: TrNode\n",
    "            ret_str = f\"{ret_str}{node.path if node else None} <- @{1 / (p * 1e-9):.2f}Hz\"\n",
    "    return ret_str\n",
    "\n",
    "\n",
    "def ts_str(inst, prev):\n",
    "    return f\"{(inst.timestamp - prev) * 1e3:+09.3f}ms\"\n",
    "\n",
    "\n",
    "def bw_str(bw):\n",
    "    return bytes_str(bw['mean'])\n",
    "\n",
    "\n",
    "def trunc_chars(string, w, e2e):\n",
    "    if w < e2e * .005:\n",
    "        return \"\"\n",
    "    n_chars = max(math.floor(w / (e2e * .17) * 65) - 5, 0)\n",
    "    if n_chars < 4:\n",
    "        return \"\"\n",
    "\n",
    "    return \"...\" + string[-n_chars:] if n_chars < len(string) else string\n",
    "\n",
    "\n",
    "#e2e_paths = sorted(e2e_paths, key=lambda path: path[-1].timestamp - path[0].timestamp, reverse=True)\n",
    "#for y, e2e_path in reversed(list(enumerate(e2e_paths))):\n",
    "#    last_pub_ts = None\n",
    "#    last_cb_end = None\n",
    "#    print(f\"=== {y}:\")\n",
    "#    for inst in e2e_path:\n",
    "\n",
    "#tree = sort_subtree(get_relevant_tree(tree, _leaf_filter))\n",
    "\n",
    "t_start = tree.head.timestamp\n",
    "t_min = t_start - e2e_lat(tree)\n",
    "t_e2e = t_start - t_min\n",
    "\n",
    "legend_entries = {}\n",
    "\n",
    "\n",
    "def plot_subtree(subtree: DepTree, ax: plt.Axes, y_labels, y=0, next_cb_start=0):\n",
    "    height = fanout(subtree)\n",
    "    inst = subtree.head\n",
    "\n",
    "    match inst:\n",
    "        case TrCallbackInstance(timestamp=t_cb, duration=d_cb):\n",
    "            is_sub = isinstance(inst.callback_obj.owner, TrSubscriptionObject)\n",
    "\n",
    "            r_x = t_cb - t_start + margin_x / 2\n",
    "            r_y = y + margin_y / 2\n",
    "            r_w = max(d_cb - margin_x, 0)\n",
    "            r_h = height - margin_y\n",
    "\n",
    "            r = mpatch.Rectangle((r_x, r_y), r_w, r_h,\n",
    "                                 ec=\"cadetblue\" if is_sub else \"indianred\", fc=\"lightblue\" if is_sub else \"lightcoral\",\n",
    "                                 zorder=9000)\n",
    "            ax.add_artist(r)\n",
    "\n",
    "            text = repr(sanitize(\n",
    "                inst.callback_obj.callback_symbol.symbol)) if inst.callback_obj and inst.callback_obj.callback_symbol else \"??\"\n",
    "            text = trunc_chars(text, r_w, t_e2e)\n",
    "            if text:\n",
    "                ax.text(r_x + r_w / 2, r_y + r_h / 2, text, ha=\"center\", va=\"center\", backgroundcolor=(1, 1, 1, .5),\n",
    "                        zorder=11000)\n",
    "\n",
    "            if is_sub and \"Subscription CB\" not in legend_entries:\n",
    "                legend_entries[\"Subscription CB\"] = r\n",
    "            elif not is_sub and \"Timer CB\" not in legend_entries:\n",
    "                legend_entries[\"Timer CB\"] = r\n",
    "\n",
    "            if next_cb_start is not None:\n",
    "                r_x = t_cb - t_start + d_cb - margin_x / 2\n",
    "                r_y = y + .5 - arr_width / 2\n",
    "                r_w = next_cb_start - (t_cb + d_cb) + margin_x\n",
    "                r_h = arr_width\n",
    "                r = mpatch.Rectangle((r_x, r_y), r_w, r_h, color=\"orange\")\n",
    "                ax.add_artist(r)\n",
    "\n",
    "                if is_sub:\n",
    "                    node = inst.callback_obj.owner.subscription.node\n",
    "                else:\n",
    "                    node = inst.callback_obj.owner.node\n",
    "                text = node.path\n",
    "\n",
    "                text = trunc_chars(text, r_w, t_e2e)\n",
    "                if text:\n",
    "                    ax.text(r_x + r_w / 2, r_y + r_h / 2, text, ha=\"center\", va=\"center\", backgroundcolor=(1, 1, 1, .5),\n",
    "                            zorder=11000)\n",
    "\n",
    "                if \"Idle\" not in legend_entries:\n",
    "                    legend_entries[\"Idle\"] = r\n",
    "\n",
    "            next_cb_start = t_cb\n",
    "        case TrPublishInstance(timestamp=t_pub, publisher=pub):\n",
    "            if not subtree.deps:\n",
    "                y_labels.append(pub.topic_name if pub else None)\n",
    "\n",
    "            scatter = ax.scatter(t_pub - t_start, y + .5, color=\"cyan\", marker=\".\", zorder=10000)\n",
    "\n",
    "            if \"Publication\" not in legend_entries:\n",
    "                legend_entries[\"Publication\"] = scatter\n",
    "\n",
    "            if next_cb_start is not None:\n",
    "                r_x = t_pub - t_start\n",
    "                r_y = y + .5 - arr_width / 2\n",
    "                r_w = max(next_cb_start - t_pub + margin_x / 2, 0)\n",
    "                r = mpatch.Rectangle((r_x, r_y), r_w, arr_width, color=\"lightgreen\")\n",
    "                ax.add_artist(r)\n",
    "                if pub:\n",
    "                    text = pub.topic_name\n",
    "                    text = trunc_chars(text, r_w, t_e2e)\n",
    "                    if text:\n",
    "                        ax.text(r_x + r_w / 2, r_y + arr_width / 2, text, ha=\"center\", va=\"center\",\n",
    "                                backgroundcolor=(1, 1, 1, .5), zorder=11000)\n",
    "\n",
    "                if \"DDS\" not in legend_entries:\n",
    "                    legend_entries[\"DDS\"] = r\n",
    "\n",
    "                topic_stats = msg_sizes.get(pub.topic_name) if BW_ENABLED else None\n",
    "                if topic_stats:\n",
    "                    size_str = bw_str(topic_stats)\n",
    "                    ax.text(r_x + r_w / 2, r_y + arr_width + margin_y, size_str, ha=\"center\",\n",
    "                            backgroundcolor=(1, 1, 1, .5), zorder=11000)\n",
    "            else:\n",
    "                print(\"[WARN] Tried to publish to another PublishInstance\")\n",
    "            next_cb_start = None\n",
    "\n",
    "    acc_fanout = 0\n",
    "    for dep in subtree.deps:\n",
    "        acc_fanout += plot_subtree(dep, ax, y_labels, y + acc_fanout, next_cb_start)\n",
    "    return height\n",
    "\n",
    "\n",
    "fig, ax = plt.subplots(figsize=(36, 20), num=\"path_viz\")\n",
    "ax.set_ylim(0, len(e2es))\n",
    "\n",
    "y_labels = []\n",
    "plot_subtree(tree, ax, y_labels)\n",
    "\n",
    "tree_e2e = e2e_lat(tree)\n",
    "plot_margin_x = .01 * tree_e2e\n",
    "ax.set_xlim(-tree_e2e - plot_margin_x, plot_margin_x)\n",
    "ax.set_yticks(np.array(range(len(y_labels))) + .5, y_labels)\n",
    "ax.set_title(f\"Timestep {E2E_PLOT_TIMESTAMP}: {(tree.head.timestamp - relevant_trees[0].head.timestamp):10.6f}s\")\n",
    "ax.set_xlabel(\"Time relative to output message [s]\")\n",
    "ax.set_ylabel(\"Start topic\")\n",
    "\n",
    "labels, handles = list(zip(*legend_entries.items()))\n",
    "ax.legend(handles, labels)\n",
    "print(len(y_labels))"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "# Find Top Critical Paths\n",
    "\n",
    "For each message tree, find its critical path and runner-up candidates. Then, for the whole timeseries, find the top critical paths by occurence count.\n",
    "Critical paths are abstracted by their E2E-latency and a list of passed CBs and topics.\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%% md\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "%%skip_if_false E2E_ENABLED\n",
    "\n",
    "critical_paths = {}\n",
    "print(len(relevant_trees))\n",
    "for tree in tqdm(relevant_trees):\n",
    "    crit = critical_path(tree)\n",
    "\n",
    "\n",
    "    def _owner(inst):\n",
    "        match inst:\n",
    "            case TrCallbackInstance(callback_obj=cb_obj):\n",
    "                cb_obj: TrCallbackObject\n",
    "                if cb_obj and cb_obj.callback_symbol:\n",
    "                    sym = repr(sanitize(cb_obj.callback_symbol.symbol))\n",
    "                else:\n",
    "                    sym = str(cb_obj.id)\n",
    "                return sym\n",
    "            case TrPublishInstance(publisher=pub):\n",
    "                pub: TrPublisher\n",
    "                topic = pub.topic_name\n",
    "                return topic\n",
    "            case _:\n",
    "                raise ValueError()\n",
    "\n",
    "\n",
    "    key = tuple(map(_owner, crit[::-1]))\n",
    "    if key not in critical_paths:\n",
    "        critical_paths[key] = []\n",
    "    critical_paths[key].append(crit)\n",
    "\n",
    "items = list(critical_paths.items())\n",
    "items.sort(key=lambda pair: len(pair[1]), reverse=True)\n",
    "\n",
    "out_df = pd.DataFrame(columns=[\"path\", \"timestamp\", \"e2e_latency\"])\n",
    "for key, paths in items:\n",
    "    path_records = [{\"path\": \" -> \".join(key),\n",
    "                     \"timestamp\": p[0].timestamp,\n",
    "                     \"e2e_latency\": p[0].timestamp - p[-1].timestamp}\n",
    "                    for p in paths]\n",
    "    out_df = out_df.append(path_records)\n",
    "    print(\n",
    "        f\"======== {len(paths)}x: {sum(map(lambda p: p[0].timestamp - p[-1].timestamp, paths)) / len(paths) * 1000:.3f}ms\")\n",
    "    paths_durations = []\n",
    "    for path in paths:\n",
    "        next_inst = None\n",
    "        durations = []\n",
    "        for inst in path:\n",
    "            match inst:\n",
    "                case TrCallbackInstance(timestamp=t, duration=d):\n",
    "                    if not next_inst:\n",
    "                        durations.append(d)\n",
    "                    else:\n",
    "                        durations.append(min(d, next_inst.timestamp - t))\n",
    "                case TrPublishInstance(timestamp=t):\n",
    "                    if not next_inst:\n",
    "                        durations.append(0.0)\n",
    "                    else:\n",
    "                        durations.append(next_inst.timestamp - t)\n",
    "                case _:\n",
    "                    raise ValueError()\n",
    "            next_inst = inst\n",
    "        paths_durations.append(durations)\n",
    "\n",
    "    durations = list(map(lambda l: sum(l) / len(l), zip(*paths_durations)))[::-1]\n",
    "    assert len(durations) == len(key)\n",
    "    perc = np.percentile(durations, [70, 90, 95, 100])\n",
    "    colors = [\"green\", \"yellow\", \"red\", \"magenta\"]\n",
    "    for part, duration in zip(key, durations):\n",
    "        E2E_PLOT_TIMESTAMP = 0\n",
    "        for j, p in enumerate(perc):\n",
    "            if duration < p:\n",
    "                break\n",
    "            E2E_PLOT_TIMESTAMP = j\n",
    "        dur_str = colored(f\"{duration * 1000 :>.3f}ms\", colors[E2E_PLOT_TIMESTAMP])\n",
    "        print(f\"  -> {dur_str} {part}\")\n",
    "\n",
    "out_df.to_csv(os.path.join(OUT_PATH, \"e2e.csv\"), sep=\"\\t\", index=False)\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "e7370f93d1d0cde622a1f8e1c04877d8463912d04d973331ad4851f04de6915a"
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}