Stormworks HTTPデータロガーの仕組み

local PORT = 8080
local SAMPLE_INTERVAL = 1
local CHANNEL_COUNT = 8
local tick = 0

function onTick()
  tick = tick + 1

  if not input.getBool(1) then
    return
  end

  if tick % SAMPLE_INTERVAL ~= 0 then
    return
  end

  local query = {"/log?tick=", tostring(tick)}

  for channel = 1, CHANNEL_COUNT do
    local value = input.getNumber(channel)
    query[#query + 1] = string.format(
      "&c%d=%.6f",
      channel,
      value
    )
  end

  async.httpGet(PORT, table.concat(query))
end

function httpReply(port, request, reply)
  -- 最小構成では応答本文を使わない
end

from csv import writer
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from threading import Lock
from urllib.parse import parse_qs, urlparse

PORT = 8080
CHANNELS = range(1, 9)
OUTPUT = Path(__file__).with_name("stormworks_log.csv")
WRITE_LOCK = Lock()


class LogHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        request = urlparse(self.path)

        if request.path != "/log":
            self.send_error(404)
            return

        query = parse_qs(request.query)
        row = [
            datetime.now().isoformat(timespec="milliseconds"),
            query.get("tick", [""])[0],
            *[
                query.get(f"c{channel}", [""])[0]
                for channel in CHANNELS
            ],
        ]

        with WRITE_LOCK:
            needs_header = (
                not OUTPUT.exists()
                or OUTPUT.stat().st_size == 0
            )

            with OUTPUT.open(
                "a",
                newline="",
                encoding="utf-8",
            ) as stream:
                csv_writer = writer(stream)

                if needs_header:
                    csv_writer.writerow([
                        "PC_Time",
                        "Tick",
                        *[
                            f"Channel{channel}"
                            for channel in CHANNELS
                        ],
                    ])

                csv_writer.writerow(row)

        self.send_response(204)
        self.end_headers()

    def log_message(self, format, *args):
        return


if __name__ == "__main__":
    server = ThreadingHTTPServer(
        ("localhost", PORT),
        LogHandler,
    )
    print(f"Stormworks logger: http://localhost:{PORT}")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()