def main()

in polymetis/python/scripts/viz.py [0:0]

• 42 lines of code
• 4 McCabe index (conditional complexity)

def main(cfg):
    print(f"Config:\n{omegaconf.OmegaConf.to_yaml(cfg)}")

    # Connect to server...
    visualizer = RobotStateVisualizer(
        server_ip=cfg.server_ip,
        server_port=cfg.server_port,
        max_queue_size=cfg.max_queue_size,
        downsampling_ratio=cfg.downsampling_ratio,
        log_keys=set(cfg.log_keys),
        logfile=cfg.logfile,
    )

    # Create initial figure
    fig = initialize_graphs(visualizer)

    # Create app
    style = omegaconf.OmegaConf.to_container(cfg.app.external_stylesheets, resolve=True)
    app = dash.Dash("controller_manager_visualizer", external_stylesheets=style)

    # Define app layout
    app.layout = html.Div(
        children=[
            html.H1(children="Controller manager visualization"),
            dcc.Graph(id="live-update-graph", figure=fig),
            dcc.Interval(
                id="interval-component",
                interval=cfg.update_interval_seconds * 1000,
                n_intervals=0,
            ),
        ]
    )

    @app.callback(
        Output("live-update-graph", "extendData"),
        [Input("interval-component", "n_intervals")],
    )
    def update_graph_data(n_intervals):
        dataframes = visualizer.process_queue()

        updated_data = {"x": [], "y": []}
        num_traces = 0

        # Add graph objects
        for df_name, df in dataframes.items():
            datetimes = [x.strftime("%H:%M:%S") for x in df.index]
            for column in df.columns:
                num_traces += 1
                updated_data["x"].append(datetimes)
                updated_data["y"].append(df[column])

        trace_indices = list(range(num_traces))
        return [updated_data, trace_indices]

    app.logger.setLevel(logging.WARNING)
    app.run_server(debug=cfg.app.debug, host=cfg.app.host, port=cfg.app.port)