def render_dot()

in common_components/monitoring/json_to_dot.py [0:0]

• 47 lines of code
• 13 McCabe index (conditional complexity)

def render_dot(nodes,
               left_right=False,
               seconds_per_block=1,
               fontsize=24,
               debug=False):
    """Render graph into a DOT graph

    Parameters
    ----------
    nodes : Nodes
        Node graph for rendering
    left_right : bool
        Whether to use left to right or top to bottom for query timelines
    seconds_per_block : int
        Number of seconds per block size
    fontsize : int
        Font size for nodes
    debug : bool
        Whether to produce a DOT graph that is debuggable
    """
    start_time = min([v["start_time"] for v in nodes.values() if v["start_time"]])
    end_time = max([v["end_time"] for v in nodes.values() if v["end_time"]])

    def find_block(time_stamp):
        return int(time_stamp - start_time) // seconds_per_block

    print("digraph {\n")
    print('newrank=true;')
    print('compound=true;')
    print(f'rankdir="{"LR" if left_right else "TB"}";')
    print('ranksep="equally";')
    print('overlap="false";')
    print(f'node [fontsize={fontsize}, shape = plaintext];')

    print('\n/* Timeline */\n')
    print('edge [style="invis"];')
    for blk in range(find_block(end_time)):
        print(f"{blk} -> {blk+1}")
        print(f"{blk} [label=\"{get_elapsed_time(blk * seconds_per_block)}\"] ")

    print('\n/* Main Styling */\n')
    print('edge [style="solid"];')
    if debug:
        print('node [shape="box"];')

    print('\n/* Main Data */\n')

    for node_id, node in nodes.items():

        print(f"subgraph cluster_{node_id} {{")
        print(f'  bgcolor="{"red" if "critical" in node else "steelblue1"}"')
        print("  style=\"rounded\"")

        # Find start and end time block for alignment
        start_align = find_block(node["start_time"])
        end_align = find_block(node["end_time"]) - 1

        if start_align < end_align:
            print(f"  {node_id}_start [fontsize={fontsize}, label=\"{node['name']}\" width=0]")
            print(f"  {node_id} [style=\"invis\" label=\"{node['name']}\"]")
        else:
            print(f"  {node_id} [fontsize={fontsize}, label=\"{node['name']}\"]")

        print("}")

        if start_align < end_align:
            print(f"{{ rank=same; {start_align} {node_id}_start }}")
            print(f"{{ rank=same; {end_align} {node_id} }}")
        else:
            print(f"{{ rank=same; {start_align} {node_id} }}")

        # Upstream dependencies
        for depends in node["depends_on"]:
            print(f"{depends} -> {node_id}{'_start' if start_align < end_align else ''} " +
                  f"[ltail=\"cluster_{depends}\" lhead=\"cluster_{node_id}\"]")

    print("}\n")