Remove Visualizer code from xyz.py

Author: chutsu

src/xyz.py

@@ -25,7 +25,6 @@
 - CALIBRATION
 - SIMULATION
 - CONTROL
-- Visualizer
 
 """
 import os
@@ -13484,301 +13483,6 @@ def on_key(event, fig):
       plt.show()
 
 
-###############################################################################
-# Visualizer
-# class DevServer
-# class MultiPlot
-###############################################################################
-
-try:
-  import websockets
-except:
-  pip_install("websockets")
-
-
-class DevServer:
-  """ Dev server """
-  def __init__(self, loop_fn):
-    self.host = "127.0.0.1"
-    self.port = 5000
-    self.loop_fn = loop_fn
-
-  def __del__(self):
-    process = Popen([f"lsof", "-i", ":{self.port}"], stdout=PIPE, stderr=PIPE)
-    stdout, _ = process.communicate()
-    for process in str(stdout.decode("utf-8")).split("\n")[1:]:
-      data = [x for x in process.split(" ") if x != '']
-      if len(data) <= 1:
-        continue
-      print(f"killing {data[1]}")
-      os.kill(int(data[1]), signal.SIGKILL)
-
-  def run(self):
-    """ Run server """
-    kwargs = {"ping_timeout": 1, "close_timeout": 1}
-    server = websockets.serve(self.loop_fn, self.host, self.port, **kwargs)
-    loop = asyncio.get_event_loop()
-    loop.run_until_complete(server)
-    loop.run_forever()
-
-  @staticmethod
-  def stop():
-    """ Stop server """
-    asyncio.get_event_loop().stop()
-
-
-class MultiPlot:
-  """ MultiPlot """
-  def __init__(self, has_gnd=False):
-    self.plots = []
-    self.add_pos_xy_plot(has_gnd=has_gnd)
-    self.add_pos_z_plot(has_gnd=has_gnd)
-    self.add_roll_plot(has_gnd=has_gnd)
-    self.add_pitch_plot(has_gnd=has_gnd)
-    self.add_yaw_plot(has_gnd=has_gnd)
-    self.add_pos_error_plot()
-    self.add_att_error_plot()
-    self.add_reproj_error_plot()
-
-    self.plot_data = {}
-    self.emit_rate = 8.0  # Hz
-    self.last_updated = datetime.now()
-
-  def _add_plot(self, title, xlabel, ylabel, trace_names, **kwargs):
-    conf = {}
-    conf["title"] = title
-    conf["width"] = kwargs.get("width", 300)
-    conf["height"] = kwargs.get("height", 280)
-    conf["buf_size"] = kwargs.get("buf_size", 100)
-    conf["trace_names"] = trace_names
-    conf["xlabel"] = xlabel
-    conf["ylabel"] = ylabel
-    conf["show_legend"] = (len(trace_names) > 1)
-    self.plots.append(conf)
-
-  def add_pos_xy_plot(self, **kwargs):
-    """ Add Position X-Y Data """
-    title = "Position X-Y"
-    xlabel = "x [m]"
-    ylabel = "y [m]"
-    trace_names = ["Estimate"]
-    if kwargs.get("has_gnd"):
-      trace_names.append("Ground-Truth")
-
-    self._add_plot(title, xlabel, ylabel, trace_names)
-
-  def add_pos_z_plot(self, **kwargs):
-    """ Add Position Z Data """
-    xlabel = "Time [s]"
-    ylabel = "y [m]"
-    trace_names = ["Estimate"]
-    if kwargs.get("has_gnd"):
-      trace_names.append("Ground-Truth")
-
-    self._add_plot("Position Z", xlabel, ylabel, trace_names)
-
-  def add_roll_plot(self, **kwargs):
-    """ Add Roll Data """
-    xlabel = "Time [s]"
-    ylabel = "Attitude [deg]"
-    trace_names = ["Estimate"]
-    if kwargs.get("has_gnd"):
-      trace_names.append("Ground-Truth")
-
-    self._add_plot("Roll", xlabel, ylabel, trace_names)
-
-  def add_pitch_plot(self, **kwargs):
-    """ Add Roll Data """
-    xlabel = "Time [s]"
-    ylabel = "Attitude [deg]"
-    trace_names = ["Estimate"]
-    if kwargs.get("has_gnd"):
-      trace_names.append("Ground-Truth")
-
-    self._add_plot("Pitch", xlabel, ylabel, trace_names)
-
-  def add_yaw_plot(self, **kwargs):
-    """ Add Yaw Data """
-    xlabel = "Time [s]"
-    ylabel = "Attitude [deg]"
-    trace_names = ["Estimate"]
-    if kwargs.get("has_gnd"):
-      trace_names.append("Ground-Truth")
-
-    self._add_plot("Yaw", xlabel, ylabel, trace_names)
-
-  def add_pos_error_plot(self):
-    """ Add Position Error Data """
-    title = "Position Error"
-    xlabel = "Time [s]"
-    ylabel = "Position Error [m]"
-    trace_names = ["Error"]
-    self._add_plot(title, xlabel, ylabel, trace_names)
-
-  def add_att_error_plot(self):
-    """ Add Attitude Error Data """
-    title = "Attitude Error"
-    xlabel = "Time [s]"
-    ylabel = "Position Error [m]"
-    trace_names = ["Error"]
-    self._add_plot(title, xlabel, ylabel, trace_names)
-
-  def add_reproj_error_plot(self):
-    """ Add Reprojection Error Data """
-    title = "Reprojection Error"
-    xlabel = "Time [s]"
-    ylabel = "Reprojection Error [px]"
-    trace_names = ["Mean", "RMSE"]
-    self._add_plot(title, xlabel, ylabel, trace_names)
-
-  def _form_plot_data(self, plot_title, time_s, **kwargs):
-    gnd = kwargs.get("gnd")
-    est = kwargs.get("est")
-    err = kwargs.get("err")
-
-    conf = {plot_title: {}}
-    if gnd:
-      conf[plot_title]["Ground-Truth"] = {"x": time_s, "y": gnd}
-
-    if est:
-      conf[plot_title]["Estimate"] = {"x": time_s, "y": est}
-
-    if err:
-      conf[plot_title]["Error"] = {"x": time_s, "y": err}
-
-    self.plot_data.update(conf)
-
-  def add_pos_xy_data(self, **kwargs):
-    """ Add Position X-Y Data """
-    plot_title = "Position X-Y"
-    conf = {plot_title: {}}
-
-    if "gnd" in kwargs:
-      gnd = kwargs["gnd"]
-      conf[plot_title]["Ground-Truth"] = {"x": gnd[0], "y": gnd[1]}
-
-    if "est" in kwargs:
-      est = kwargs["est"]
-      conf[plot_title]["Estimate"] = {"x": est[0], "y": est[1]}
-
-    self.plot_data.update(conf)
-
-  def add_pos_z_data(self, time_s, **kwargs):
-    """ Add Position Z Data """
-    self._form_plot_data("Position Z", time_s, **kwargs)
-
-  def add_roll_data(self, time_s, **kwargs):
-    """ Add Roll Data """
-    self._form_plot_data("Roll", time_s, **kwargs)
-
-  def add_pitch_data(self, time_s, **kwargs):
-    """ Add Roll Data """
-    self._form_plot_data("Pitch", time_s, **kwargs)
-
-  def add_yaw_data(self, time_s, **kwargs):
-    """ Add Yaw Data """
-    self._form_plot_data("Yaw", time_s, **kwargs)
-
-  def add_pos_error_data(self, time_s, error):
-    """ Add Position Error Data """
-    self._form_plot_data("Position Error", time_s, err=error)
-
-  def add_att_error_data(self, time_s, error):
-    """ Add Attitude Error Data """
-    self._form_plot_data("Attitude Error", time_s, err=error)
-
-  def add_reproj_error_data(self, time_s, reproj_rmse, reproj_mean):
-    """ Add Reprojection Error Data """
-    plot_title = "Reprojection Error"
-    conf = {plot_title: {}}
-    conf[plot_title]["Mean"] = {"x": time_s, "y": reproj_rmse}
-    conf[plot_title]["RMSE"] = {"x": time_s, "y": reproj_mean}
-    self.plot_data.update(conf)
-
-  def get_plots(self):
-    """ Get plots """
-    return json.dumps(self.plots)
-
-  def get_plot_data(self):
-    """ Get plot data """
-    return json.dumps(self.plot_data)
-
-  async def emit_data(self, ws):
-    """ Emit data """
-    time_now = datetime.now()
-    time_diff = (time_now - self.last_updated).total_seconds()
-    if time_diff > (1.0 / self.emit_rate):
-      await ws.send(self.get_plot_data())
-      self.last_updated = time_now
-
-
-async def fake_loop(ws, _):
-  """ Simulates a simulation or dev loop """
-  # Setup plots
-  print("Connected to client!")
-  multi_plot = MultiPlot(has_gnd=True)
-  await ws.send(multi_plot.get_plots())
-
-  # Loop
-  index = 0
-
-  while True:
-    index += 1
-
-    t = index
-    x = np.random.random()
-    y = np.random.random()
-    z = np.random.random()
-    gnd = np.random.random(3)
-    est = np.random.random(3)
-    multi_plot.add_pos_xy_data(est=est, gnd=gnd)
-    multi_plot.add_pos_z_data(t, est=z, gnd=x)
-    multi_plot.add_roll_data(t, est=x, gnd=y)
-    multi_plot.add_pitch_data(t, est=x, gnd=y)
-    multi_plot.add_yaw_data(t, est=x, gnd=y)
-    multi_plot.add_pos_error_data(t, y)
-    multi_plot.add_att_error_data(t, x)
-    multi_plot.add_reproj_error_data(t, x, y)
-    await multi_plot.emit_data(ws)
-
-  # Important
-  await ws.close()
-  DevServer.stop()
-
-
-class TestViz(unittest.TestCase):
-  """ Test Viz """
-  def test_multiplot(self):
-    """ Test MultiPlot() """
-    t = 0
-    x = np.random.random()
-    y = np.random.random()
-    z = np.random.random()
-    gnd = np.random.random(3)
-    est = np.random.random(3)
-    multi_plot = MultiPlot(has_gnd=True)
-    multi_plot.add_pos_xy_data(est=est, gnd=gnd)
-    multi_plot.add_pos_z_data(t, est=z, gnd=x)
-    multi_plot.add_roll_data(t, est=x, gnd=y)
-    multi_plot.add_pitch_data(t, est=x, gnd=y)
-    multi_plot.add_yaw_data(t, est=x, gnd=y)
-    multi_plot.add_pos_error_data(t, y)
-    multi_plot.add_att_error_data(t, x)
-    multi_plot.add_reproj_error_data(t, x, y)
-
-    # import pprint
-    # pprint.pprint(multi_plot.get_plots())
-
-    self.assertTrue(multi_plot is not None)
-
-  @unittest.skip("")
-  def test_server(self):
-    """ Test DevServer() """
-    viz_server = DevServer(fake_loop)
-    viz_server.run()
-    self.assertTrue(viz_server is not None)
-
-
 ###############################################################################
 # Sandbox
 ###############################################################################