Django + Plotly.js でグラフをリアルタイム更新

最近、測定器からデータを取得してリアルタイムにデータを表示して確認したいという願望があってそれを実現する方法を考えてました。その１つの解決策としてDjangoを使ってWebアプリケーションのような形で実現する方法を思いつき、試しに作ってみたのでそれを共有したいと思います。

この記事では、プロットするデータは測定器から発生していますがPythonで得られるデータであればなんでもこの方法は応用できます。

Photo by Isaac Smith on Unsplash

なぜ Django?

• 測定器からのデータは GPIB 通信というもので来ていたのでそれを処理できる必要がある -> Python, C
• グラフを動的に変化させたい -> JavaScript

という要件があったので、Djangoを使ってWebアプリケーションとして書けば要件を満たすことができると考えました。

この記事で書かないこと

PyVISAを用いた測定器との通信方法

環境

• Python 3.9.6
• Django 3.1.5
• Plotly.js 2.3.0

構成

フロントエンド側 (HTML + JavaScript)

HTML

<script
  type="text/javascript"
  src="{% static 'core/js/plotly-2.3.0.min.js' %}"
></script>

<div id="canvas" style="width: 700px; height: 400px"></div>
<form id="tds-settings" action="{% url 'core:tds_data' %}">
  <div id="stage-position-settings">
    <div id="start">
      <p>Start:</p>
      <input type="number" id="start-position" /> &mu;m
    </div>
    <div id="end">
      <p>End:</p>
      <input type="number" id="end-position" /> &mu;m
    </div>
    <div id="step">
      <p>Step:</p>
      <input type="number" id="moving-step" /> &mu;m
    </div>
  </div>
  <p>Lock-in time:</p>
  <input type="number" id="lockin-time" /> ms
  <input type="submit" value="Start" id="start-button" />
</form>

グラフをプロットするエリア(id="canvas")と測定の設定を決めるフォームを用意します。フォームがsubmitされたときのアクションにはDjango側で用意するAPIのアドレスを入れます。

JavaScript

var trace1 = {
  x: [],
  y: [],
  type: "scatter",
};

var data = [trace1];
var layout = {
  width: 600,
  height: 400,
  margin: { l: 50, r: 0, b: 3, t: 20, pad: 5 },
  showlegend: true,
  legend: { orientation: "h" },
};

Plotly.newPlot("canvas", data, layout);

const _sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

document
  .getElementById("tds-settings")
  .addEventListener("submit", async (e) => {
    e.preventDefault();
    tds_measurement();
  });

async function tds_measurement() {
  let url = document.getElementById("tds-settings").action;
  let boot = tds_boot_url;

  let start = Number(document.getElementById("start-position").value);
  let end = Number(document.getElementById("end-position").value);
  let step = Number(document.getElementById("moving-step").value);
  let lockin = Number(document.getElementById("lockin-time").value);

  if (start >= end || start < 0 || end <= 0 || step <= 0 || lockin <= 0) {
    alert("Invalid Parameters");
    return;
  }

  let finished = false;
  fetch(boot, {
    method: "POST",
    body: `start=${start}&end=${end}&step=${step}&lockin=${lockin}`,
    headers: {
      "Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
    },
  }).catch((error) => {
    finished = true;
  });
  await _sleep(1000);
  while (!finished) {
    await fetch(url, {
      method: "POST",
      body: "",
      headers: {
        "Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
      },
    })
      .then((response) => {
        return response.json();
      })
      .then((responseJson) => {
        if (responseJson.status === "finished") {
          finished = true;
        }
        let x = responseJson.x;
        let y = responseJson.y;
        data[0].x = x;
        data[0].y = y;

        Plotly.update("canvas", data, layout);
      })
      .catch((error) => {
        alert("Error while measurement");
        finished = true;
      });

    await _sleep(500);
  }
}

JS側では、フォームがsubmitされるとまず、画面遷移が生じるないようにします(e.preventDefault())。Django側へは２回APIアクセスします。それぞれの役割は

• 測定を開始する
• 測定データを得る

となっていますが、なぜこのように２つに分けたかというと、１つにまとめてしまうと測定が終わるまでデータ点が得られないためです。Python側でエンドポイントを分けてあげることで測定シーケンスを進めるスレッドと測定データをフロント側に渡すスレッドができることになります。

測定を開始するリクエストを送ったあとは、測定が終了したという通知をPythonから受け取るまでは500msごとに測定データを要求します。

バックエンド側 (Python)

エンドポイントの設定

from django.urls import path

from . import views

app_name = "core"
urlpatterns = [
    path("tds/", views.TDS.as_view(), name="tds"),
    path("tds-data/", views.tds_data, name="tds_data"),
    path("tds-boot/", views.tds_boot, name="tds_boot"),
]

１つめは最初に書いたHTMLをテンプレートとするビュー、あとの２つはjsonを返すビューです。

ビュー

from django.views import View
from django.http import JsonResponse

wave_tds = WaveForm()
tds_running = False

class TDS(View):
    def get(self, request):
        return render(request, "core/tds.html")

def tds_boot(request):
    global tds_running

    tds_running = True
    wave_tds.clear()
    start = int(request.POST.get("start"))
    end = int(request.POST.get("end"))
    step = int(request.POST.get("step"))
    lockin = float(request.POST.get("lockin"))

    api_ops.tds_scan(start, end, step, lockin, wave_tds)  # 測定シーケンスの中身

    tds_running = False

    return JsonResponse({})


def tds_data(request):
    x = wave_tds.x
    y = wave_tds.y

    status = "running" if tds_running else "finished"

    return JsonResponse({"x": x, "y": y, "status": status})

測定シーケンスが走っているかいないかはグローバル変数 tds_runningで判断します。

実際の動作 (動画)

実際に動作している様子を動画にしました。データは適当にsinカーブを生成するようにしています。マウスを近づけると値が読めたり、詳しく見たい部分をズームアップできるのは嬉しいですね。
streamable.com

現場で使える Django の教科書《基礎編》（3.2 対応版）