Light speed with Python and JS

We've gone to plaid!

Summary

Nope, this article is not about making Python and JS run very fast, it's about making a light speed effect for the fun of a hundred of drunk people.

Take a midi board, connect it using mido and fastapi, send the signals to some JS over websocker, draw lines on top of a trippy video when you receive a message...

You now can serve the drinks.

Those are not the perfs you are looking for

Yeah, I embraced the clickbait on this one. It’s not about making Python fast.

Every year, we make a big party on Halloween turning an entire house into whatever madness we have decided on. In 2023, we went sci-fi, and transformed it all into a giant spaceship. It took about 2 months and 20 persons to setup:

You can hear me dying in the background. And one friend trying the piano one last time before it becomes part of a cardboard nuclear reactor.

Just like with professional projects, it starts with very a serious meeting where we make a coherent plan. Then we proceed to never follow it and move in absolute chaos until the deadline forces us to make compromises.

One of my missions was to spice up the cockpit, so I installed a laser projector with a video of colorful space animations behind a frame. The end result is super nice:

Smoke and dim light make everything more impressive

I have a dusty midi controller I thought would make a cool looking part of the dashboard:

On one hand I should clean it, on the other hand, my keyboard is in a worse shape

But since it was there, why not connect it and make it do something?

So I decided to hack a light speed effect in top of the space video, activated by the pad. This way people could play with it between two mojitos:

We gonna need a lot of duck tape.

Getting the signal

Thanks to the library mido, we can read inputs from a midi controller without too much fuss.

To get an idea of how to use it, I’ll show snippets first. We instantiate a backend (we'll take the generic one) and connect to the device. The general idea is this:

backend = mido.Backend("alsa_midi.mido_backend")
device = backend.get_input_names()[0] # if you have only one
midi_input = backend.open_input(device)

Then you get the messages:

for message in port.iter_pending():
    if message.type == "note_on":
        # that's a note being played, so a little square button
        print(message.note)
        name = NOTE_MAPPING.get(message.note, "Unknown")
        messagereceived(name, message.velocity)

    elif message.type == "control_change":
        # that's a control changing value, so a radio button, 
        # a slider, etc.
        name = CONTROL_MAPPING.get(
            message.control, "Unknown"
        )
        messagereceived(name, message.value)

I did a few trial and errors to map the names of the buttons on the pad and the codes I was receiving, and it looked like this:

CONTROL_MAPPING = {
    9: "K2",
    20: "F1",
...
}

NOTE_MAPPING = {
    22: "Pad_12",
    23: "Pad_13",
...
}

Then the day after, it stopped working. Turns out depending of the way you plug the USB C cable, it changes.

So I marked the cable, orientation and plug, and didn't deviate from that.

Then I needed to write the function to react to the messages received:

def messagereceived(key, value):
    print(key, value)
    try:
        requests.post(f"http://localhost:8000/relay/?key={key}&value={value}")
    except Exception as e:
        # High quality code for a high quality project
        print("error", e)
        pass

It makes two assumptions:

• I'm going to setup an endpoint at http://localhost:8000/relay/.

• The HTTP request to the endpoint is going to be fast enough to keep up with the midi input.

Of course, using it more and more, you start noting things like latency and disconnections. You wonder what to do when you plug multiple devices, and so on.

Then you realize the date is approaching and you don't have that much time left between a full time job, a blog to write and a deep love of naps.

So you ship a blob. That’s the real code we are going to dig from the past and try to understand the state of mind of the author at the time (colored version here):

import os
import queue
import re
import time

import mido
import requests

# I don't even remember why it's there. Permission issue I worked 
#  around? Maybe I replaced that with an udev rules?
os.geteuid = lambda: print("nope") or 0

CONTROL_MAPPING = {
    9: "K2",
    20: "F1",
    21: "F2",
    28: "K1",
}

NOTE_MAPPING = {
    22: "Pad_12",
    23: "Pad_13",
    24: "Pad_14",
    25: "Pad_15",
    26: "Pad_16",
    11: "Pad_1",
    12: "Pad_2",
    13: "Pad_3",
    14: "Pad_4",
    15: "Pad_5",
    16: "Pad_6",
    17: "Pad_7",
    18: "Pad_8",
    19: "Pad_9",
    20: "Pad_10",
    21: "Pad_11",
}


def messagereceived(key, value):
    print(key, value)
    try:
        requests.post(f"http://localhost:8000/relay/?key={key}&value={value}")
    except Exception as e:
        print("error", e)
        pass


# There is zero reason for a class to be used here. I probably 
# had the grand delusion of making it pluggable.
class Controller:

    def __init__(self):
        self.midi_backend = mido.Backend("alsa_midi.mido_backend")

    def listen(
        self,
        device_name: str,
        # Yeah, definitely tried to make that generic.
        device_name_is_regex: bool,
        first_discovery_timeout: int = 30,
    ):
        # Seems I loop trying to find the device, which is likely
        # to allow for plugging and unplugging it
        error = ""
        for i in range(first_discovery_timeout * 2):
            try:
                device = self.find_device(device_name, device_name_is_regex)
                break
            except SystemError as e:
                if not i:
                    print("Waiting for device...")
                error = str(e)
                time.sleep(0.5)
        else:
            # We can't find a device, no point in holding to life
            raise SystemExit(str(error))

        port = self.midi_backend.open_input(device)

        # That's an event loop folks. Don't let people tell you otherwise.
        while True:
            try:
                if self.is_device_connected(device_name, device_name_is_regex):
                    with self.midi_backend.open_input(device) as port:
                        print("Device connected:", device_name)

                        while self.is_device_connected(
                            device_name, device_name_is_regex
                        ):
                            try:
                                for message in port.iter_pending():
                                    if message.type == "note_on":
                                        print(message.note)
                                        name = NOTE_MAPPING.get(message.note, "Unknown")
                                        messagereceived(name, message.velocity)

                                    elif message.type == "control_change":
                                        name = CONTROL_MAPPING.get(
                                            message.control, "Unknown"
                                        )
                                        messagereceived(name, message.value)

                            except queue.Empty:
                                time.sleep(0.1)
            # unplug tolerance again, and beautiful black whole for any other source
            # of error
            except OSError:
                pass

            print("Device disconnected", name)
            print("Waiting for device...")
            time.sleep(0.5)

    def list_devices(self):
        return self.midi_backend.get_input_names()

    def find_device(self, name, regex=False):

        # I don't even think I own another midi device, but hey

        inputs = self.list_devices()

        if not regex:
            matches = [n for n in inputs if name.lower() in n.lower()]
        else:
            matches = [n for n in inputs if re.search(name, n)]

        if not matches:
            raise SystemError(f"No device matching '{name}' found")

        if len(matches) > 1:
            raise SystemError(f"{len(matches)} devices matching '{name}' found")

        return matches[0]

    def is_device_connected(self, name, regex=False):
        try:
            return self.find_device(name, regex)
        # When you catch SystemError, you know you reached the bottom
        except SystemError:
            return False

If you ever needed reassurance for your imposter syndrome, there it is.

Let's move on to the web server.

Putting micro services on your resume

To put the light speed effect on top of the video, I can't play it with VLC anymore. The easiest path is to load it in a Web browser since it got a powerful video and canvas API I can leverage without any special lib. I really don't want to program that with QT.

At this stage, I realize I should I’ve tried web midi first, instead of creating this Rube Goldberg machine but I'm committed now.

However I need to let the browser know I received the midi signals, and for that, a bit of websocket would work wonder. So let's pop fastapi:

from fastapi import FastAPI, WebSocket
from starlette.responses import FileResponse
from starlette.websockets import WebSocketDisconnect

app = FastAPI()
active_connections = []


@app.websocket("/ws/")
async def websocket_endpoint(websocket: WebSocket):
    # Any JS script connecting to use stores a reference to 
    # itself here
    await websocket.accept()
    active_connections.append(websocket)
    # Wait until disconnection
    try:
        while True:
            data = await websocket.receive_text()
    except WebSocketDisconnect:
        active_connections.remove(websocket)

# Remember our "http://localhost:8000/relay/?key={key}&value={value}" 
# endpoint? It relays all the midi signals to our websocket client
@app.post("/relay/")
async def relay_data(key: str, value: str):
    for connection in active_connections:
        await connection.send_json({"key": key, "value": value})
    return {"status": "data relayed to all active websockets"}

# Since we have a server, let's serve the file and avoid any CORS 
# errors. Don't think websocket have them but just in case.
@app.get("/")
def serve_html():
    return FileResponse("index.html")

Since I don't want to manage threads, I start this and the midi controller in 2 different processes manually. Somewhere, some Go dev is laughing.

Connecting the client

Moving to the frontend, it's time to write some JS. We begin with the websocket:

let ws;
let reconnectInterval = 1000;  // Starting with 1 second
const maxReconnectInterval = 30000;  // Max delay is 30 seconds
const reconnectDecay = 1.5;  // Increase the reconnect interval by 1.5 times each attempt

// What button does what. Turns out I decided to add more features
// than just the light speed effect. Because of course I did.
// Each button displays a different message.
const key_to_alerts = {
  "Pad_1": "password_required",
  "Pad_2": "password_required",
  "Pad_3": "password_required",
  "Pad_4": "password_required",
  "Pad_5": "note",
  "Pad_6": "password_required",
  "Pad_7": "hal",
  "Pad_8": "password_required",
  "Pad_9": "password_required",
  "Pad_10": "warning",
  "Pad_11": "password_required",
  "Pad_12": "error",
  "Pad_13": "dashboard",
  "Pad_14": "password_required",
  "Pad_15": "success",
  "Pad_16": "password_required",
}

function connectWebSocket() {

  ws = new WebSocket('ws://127.0.0.1:8000/ws/');

  // Manual attempt to connect, or reconnect if something fails, with
  // a little bit of exponential backoff
  ws.onopen = function () {
    console.log('WebSocket is open now.');
    reconnectInterval = 1000;  // Reset to 1 second after success
  };

  ws.onclose = function (e) {
    console.log('WebSocket is closed now. Reconnecting...');
    setTimeout(function () {
      reconnectInterval *= reconnectDecay;
      reconnectInterval = Math.min(maxReconnectInterval, reconnectInterval);
      connectWebSocket();
    }, reconnectInterval);
  };

  ws.onerror = function (err) {
    console.error('WebSocket encountered an error: ', err);
    ws.close();
  };

  // the meat of the game: every time we get a message from fastapi 
  // (so really from our midi controller), we do something
  ws.onmessage = function (event) {
    const data = JSON.parse(event.data);
    console.log(`Key: ${data.key}, Value: ${data.value}`)

    if (data.key === 'F2') {
      // Either we update the speed of the space ship
      // That's a global variable. Sorry.
      speed = (data.value + 1) / 127 * 8;
    } else if (data.key === 'Unkown') {
      // Or we display a message on the screen of the space ship.
      showAlert("password_required");
    } else {
      showAlert(key_to_alerts[data.key] || "password_required");
    }
  };
};

connectWebSocket();

Alert!

Once the button is pressed, we display modals. Basically jokes like "the virus database is up to date". They look like this:

That’s French for “Critrical mass (Your mum)”. I’m adult. I pay taxes.

The default screen requires a password so that I don't have to map ALL buttons.

And the rest load a pic that disappears after 5 seconds. We lock it to prevent overlapping:

let lock = false;

alerts = {
  password_required: `
    <form action="" class="container">
      <div class="input-container">
          <div class="input-content">
              <div class="input-dist">
                  <div class="input-type">
                      <h1 placeholder="User" required="" type="text" class="input-is">Password required</h1>
                      <span placeholder="Password" required="" type="password" style="font-weight: normal;" class="input-is">  </span>
                  </div>
              </div>
          </div>
      </div>
</div>
  `,
  dashboard: `<iframe id="dashboard" src="dashboard.html">  </iframe>`,
  success: `<div class="container"> <img class="alert" src="success.png"/></div>`,
  warning: `<div class="container"> <img class="alert" src="warning.png"/></div>`,
  error: `<div class="container"> <img class="alert" src="error.png"/></div>`,
  note: `<div class="container"> <img class="alert" src="note.png"/></div>`,
  hal: `<div class="container"> <img class="alert" src="hal.gif"/><h3 class="link-sf font-w300">I'm sorry, Dave...</h3></div>`,
}

function showAlert(name) {

  if (lock) {
    return
  }

  lock = true;
  const alertElem = document.createElement('div', { id: 'alertContainer' })
  alertElem.innerHTML += alerts[name];
  document.body.appendChild(alertElem);

  // There is CSS with that, but this article is long enough as it is
  alertElem.classList.add('animate-in');

  setTimeout(() => {
    alertElem.classList.remove('animate-in');
    alertElem.classList.add('animate-out');
    setTimeout(() => {
      alertElem.remove();
      lock = false;
    }, 1000);  // Assuming the animation duration is 1 second
  }, 5000);
}

Ludicrous speed

For the light speed effect, we have a big fat global variable named speed, and we always display stars moving on top of the video according to that. When the value changes, the stars go faster and have a longer tail (colored version):

// the HTML Video element
const bgVideo = document.getElementById('bgVideo');

// The canvas element on which we display the animation
const canvas = document.getElementById('spaceCanvas');
const ctx = canvas.getContext('2d');
canvas.width = window.innerWidth;
canvas.height = window.innerHeight;

// I create a second canvas for performance reasons (explained later)
const offscreenCanvas = document.createElement('canvas');
offscreenCanvas.width = canvas.width;
offscreenCanvas.height = canvas.height;
const offscreenCtx = offscreenCanvas.getContext('2d');

// You'll note from there the totally scientific constants and magic  
// numbers that I had in no way brute forced until it looked good
const numStars = 1000;
let stars = [];
let speed = 2;

const mid_wd = canvas.width / 2;
const mid_ht = canvas.height / 2;
const width = canvas.width;
const height = canvas.height;

// Each star is a dot in space with a tail and the length depends of
// the speed so we store previous positions which we keep drawing.
class Star {
  constructor() {
    this.x = (Math.random() - 0.5) * width;
    this.y = (Math.random() - 0.5) * height;
    this.z = Math.random() * width;  // Faking depth
    this.prevX = 0;
    this.prevY = 0;
    this.prevPositions = [];
  }

  update() {
    // Move the star away from the center of the screen
    this.z -= speed;

    const prevX = this.x * (width / this.z) + mid_wd;
    const prevY = this.y * (width / this.z) + mid_ht;
    this.prevPositions.unshift({ x: prevX, y: prevY });

    const threshold = 50;

    if (this.z <= threshold) {
      this.prevPositions = [];  // remove the tail
    }

    if (this.z <= 0) {
      this.x = (Math.random() - 0.5) * width;
      this.y = (Math.random() - 0.5) * height;
      this.z = width;
    }

    // Limit the number of previous positions based on speed
    const maxPrevPositions = Math.floor(speed * 5);
    while (this.prevPositions.length > maxPrevPositions) {
      this.prevPositions.pop();
    }
  }

  draw() {
    const scale = width / this.z;
    const x = this.x * scale + mid_wd;
    const y = this.y * scale + mid_ht;

    if (speed > 2 && this.prevPositions.length > 1) {
      // We draw all the positions on the canvas
      const furthestPos = this.prevPositions[this.prevPositions.length - 1];
      ctx.strokeStyle = 'white';
      ctx.lineWidth = scale / 4;
      ctx.beginPath();
      ctx.moveTo(furthestPos.x, furthestPos.y);
      ctx.lineTo(x, y);
      ctx.stroke();
    } else {
      // just one dot
      ctx.fillStyle = 'white';
      ctx.beginPath();
      ctx.arc(x, y, scale, 0, Math.PI * 2);
      ctx.fill();
    }
  }

}

// Initialize stars (Say it aloud with a robot voice)
for (let i = 0; i < numStars; i++) {
  stars.push(new Star());
}

function animate() {
  // Clear offscreen canvas
  ctx.drawImage(bgVideo, 0, 0, canvas.width, canvas.height);

  // Optimization trick: each star is drawned first on
  // an offscreen canvas to avoid rerendering everytime. 
  // When all of them are updated, we copy
  // the result on the main canvas.
  for (let star of stars) {
    star.update();
    star.draw(offscreenCtx);
  }

  // Draw the offscreen canvas onto the main canvas
  ctx.drawImage(offscreenCanvas, 0, 0);

  requestAnimationFrame(animate);
}

animate()

Even with the optimization trick, the HD video on a gigantic screen + the animation was too much for my poor laptop. I had to re-encode the video with less pixels, lower the screen resolution… and move from Firefox to Chrome :(

There is also a HUD but I'm reaching 500 lines for one post, so I'll stop there.

Disposable

The day after the party, you put everything away:

Two months to build, one day to take down

Don't worry, every cardboard and plastic we used were collected from the street, or trash donated by surrounding shops, and all of it was transferred to the city recycling facility.

We did use a lot of paint and duck tape, though.

And crappy code, that you write once, only to never touch it again.