#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PONT OSC → EARMOTION (version web)
==================================
Reçoit de l'OSC en UDP (Max/MSP, Pure Data, TouchOSC, QLab, Reaper…) et le remet à
EarMotion ouvert dans un navigateur.
python3 emt-osc-bridge.py
Puis, dans EarMotion : panneau « OSC » → « Connecter le pont ».
───────────────────────────────────────────────────────────────────────────────────────────
POURQUOI CE DÉTOUR PAR UNE PETITE FENÊTRE
───────────────────────────────────────────────────────────────────────────────────────────
Un navigateur n'ouvre pas de socket UDP : OSC ne peut pas lui arriver directement. Il faut
donc un pont local. Mais la voie évidente — la page https://earmotion.net ouvre elle-même une
WebSocket vers 127.0.0.1 — NE MARCHE PAS, et échoue de la pire façon.
Mesuré le 11/08/2026 sur Chrome 150 / macOS :
· `new WebSocket('ws://127.0.0.1:9131')` depuis https://earmotion.net reste en `CONNECTING`
INDÉFINIMENT. Aucun `error`, aucun `close`, aucun message en console. Même après un vrai
clic. La connexion TCP touche le pont puis se referme sans qu'un octet HTTP soit envoyé.
· `fetch(..., {targetAddressSpace:'local'})` échoue en 51 ms, sans jamais atteindre le pont.
· `navigator.permissions.query({name:'local-network-access'})` répond `prompt` : l'accès au
réseau local est désormais derrière une PERMISSION, et aucune invite ne s'est affichée.
D'où le détour, qui lui fonctionne : **EarMotion n'émet aucune requête vers le réseau local,
il ouvre une FENÊTRE.** Une navigation de premier niveau n'a jamais été soumise à cette
permission — sans quoi taper « localhost » dans la barre d'adresse serait impossible. Cette
fenêtre, servie par le pont, est donc de la MÊME ORIGINE que lui et lui parle librement ; elle
transmet ensuite à EarMotion par `postMessage`, qui traverse les origines sans rien demander.
Max/MSP ──OSC/UDP──▶ pont ──WebSocket──▶ petite fenêtre ──postMessage──▶ earmotion.net
(même origine) (origine vérifiée des deux côtés)
Conséquence à accepter : **la petite fenêtre doit rester ouverte.** La fermer coupe le pont.
C'est le prix de la permission qu'on n'a pas à demander.
───────────────────────────────────────────────────────────────────────────────────────────
CE QUE LE PONT NE SAIT PAS, ET NE DOIT PAS SAVOIR
───────────────────────────────────────────────────────────────────────────────────────────
Il ne connaît AUCUN verbe d'EarMotion. Il transmet `{adresse, arguments}` tels quels ; c'est
`msOscVersAction` (bloc pur `MS-OSC-PURE`) qui traduit, côté page, à partir du catalogue
d'actions. Ajouter un verbe au catalogue le rend joignable en OSC sans toucher à ce fichier.
C'est la même règle que pour l'API web : une seule liste de verbes dans tout le projet.
Bibliothèque standard uniquement — le projet n'a aucun gestionnaire de dépendances.
"""
import base64
import hashlib
import json
import socket
import struct
import sys
import threading
VERSION = '1.0'
PORT_OSC = 9130 # UDP, ce qu'écoute le pont
PORT_WEB = 9131 # TCP, la petite fenêtre ET sa WebSocket
SORTIE_DEFAUT = ('127.0.0.1', 9132) # retour OSC vers le contrôleur
# Les pages autorisées à recevoir l'OSC. Une page qui n'y figure pas peut ouvrir la fenêtre du
# pont, mais celle-ci ne lui transmettra rien.
PARENTS = ['https://earmotion.net', 'https://www.earmotion.net',
'http://localhost:8000', 'http://127.0.0.1:8000']
GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
clients = []
clients_lock = threading.Lock()
compteur = {'recus': 0}
# ═══════════════════════════════════════════════════════════════════════════════════════════
# OSC
# ═══════════════════════════════════════════════════════════════════════════════════════════
def _chaine(buf, i):
"""Chaîne OSC : terminée par un zéro, puis alignée sur 4 octets."""
fin = buf.index(b'\0', i)
return buf[i:fin].decode('utf-8', 'replace'), (fin + 4) & ~3
def osc_parse(data):
"""Rend [(adresse, [arguments])]. Traverse les bundles."""
if data[:8] == b'#bundle\0':
out, i = [], 16
while i + 4 <= len(data):
(taille,) = struct.unpack_from('>i', data, i)
i += 4
if taille <= 0 or i + taille > len(data):
break
out.extend(osc_parse(data[i:i + taille]))
i += taille
return out
try:
addr, i = _chaine(data, 0)
if not addr.startswith('/'):
return []
args = []
if i < len(data) and data[i:i + 1] == b',':
tags, i = _chaine(data, i)
for t in tags[1:]:
if t == 'i':
args.append(struct.unpack_from('>i', data, i)[0]); i += 4
elif t == 'f':
args.append(round(struct.unpack_from('>f', data, i)[0], 6)); i += 4
elif t == 'd':
args.append(struct.unpack_from('>d', data, i)[0]); i += 8
elif t == 'h':
args.append(struct.unpack_from('>q', data, i)[0]); i += 8
elif t in 'sS':
s, i = _chaine(data, i); args.append(s)
elif t == 'b':
(n,) = struct.unpack_from('>i', data, i)
i += 4 + ((n + 3) & ~3)
args.append(None)
elif t == 'T':
args.append(True)
elif t == 'F':
args.append(False)
elif t in 'NI':
args.append(None)
return [(addr, args)]
except Exception:
return []
def osc_build(addr, args):
# ⚠️ Le reste modulo 4 est repris modulo 4. Sans le second `% 4`, une chaîne DÉJÀ alignée
# reçoit quatre octets de trop et le lecteur d'en face cherche l'argument suivant au
# mauvais endroit : toute adresse de longueur multiple de 4 sort vide. Vu, et corrigé.
def pad(b):
return b + b'\0' * ((4 - len(b) % 4) % 4)
tete = pad(addr.encode('utf-8') + b'\0')
tags, corps = ',', b''
for a in args:
if isinstance(a, bool):
tags += 'T' if a else 'F'
elif isinstance(a, int):
tags += 'i'; corps += struct.pack('>i', a)
elif isinstance(a, float):
tags += 'f'; corps += struct.pack('>f', a)
else:
tags += 's'; corps += pad(str(a).encode('utf-8') + b'\0')
return tete + pad(tags.encode() + b'\0') + corps
# ═══════════════════════════════════════════════════════════════════════════════════════════
# La petite fenêtre — servie par le pont, donc de sa propre origine
# ═══════════════════════════════════════════════════════════════════════════════════════════
def page(port_web, port_osc):
return """
Pont OSC — EarMotion
Pont OSC
connexion…
- Écoute
- OSC/UDP
__PORT_OSC__
- Reçus
- 0
- Dernier
- —
Garder cette fenêtre ouverte : la fermer coupe le pont.
""" \
.replace('__PORT_OSC__', str(port_osc)) \
.replace('__PARENTS__', json.dumps(PARENTS)) \
.replace('__VERSION__', VERSION)
# ═══════════════════════════════════════════════════════════════════════════════════════════
# WebSocket (RFC 6455, le strict nécessaire)
# ═══════════════════════════════════════════════════════════════════════════════════════════
def ws_frame(charge):
b = charge.encode('utf-8')
n = len(b)
if n < 126:
return struct.pack('>BB', 0x81, n) + b
if n < 65536:
return struct.pack('>BBH', 0x81, 126, n) + b
return struct.pack('>BBQ', 0x81, 127, n) + b
def ws_lire(sock):
def recevoir(n):
b = b''
while len(b) < n:
m = sock.recv(n - len(b))
if not m:
return None
b += m
return b
t = recevoir(2)
if not t:
return None
op, masque, n = t[0] & 0x0F, t[1] & 0x80, t[1] & 0x7F
if n == 126:
d = recevoir(2)
if not d: return None
n = struct.unpack('>H', d)[0]
elif n == 127:
d = recevoir(8)
if not d: return None
n = struct.unpack('>Q', d)[0]
if n > (1 << 20):
return None
cle = recevoir(4) if masque else b'\0\0\0\0'
if cle is None:
return None
corps = recevoir(n) if n else b''
if corps is None:
return None
if masque:
corps = bytes(c ^ cle[i % 4] for i, c in enumerate(corps))
return op, corps
def servir(sock, port_web, port_osc, sortie, cible_udp):
"""Une connexion : soit la page, soit une WebSocket."""
inscrit = False
try:
requete = b''
while b'\r\n\r\n' not in requete:
m = sock.recv(4096)
if not m:
return
requete += m
if len(requete) > 16384:
return
lignes = requete.decode('latin-1').split('\r\n')
entetes = {}
for l in lignes[1:]:
if ':' in l:
k, v = l.split(':', 1)
entetes[k.strip().lower()] = v.strip()
if entetes.get('upgrade', '').lower() != 'websocket':
corps = page(port_web, port_osc).encode('utf-8')
sock.sendall(b'HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n'
b'Cache-Control: no-store\r\n'
b'Content-Length: ' + str(len(corps)).encode() + b'\r\n'
b'Connection: close\r\n\r\n' + corps)
return
origine = entetes.get('origin', '')
cle = entetes.get('sec-websocket-key', '')
# La fenêtre du pont est servie par le pont : sa WebSocket est de la MÊME origine.
# Toute autre origine est refusée — c'est la seule défense qui tienne, et le
# navigateur pose `Origin` lui-même : une page ne peut pas mentir dessus.
miennes = {'http://127.0.0.1:%d' % port_web, 'http://localhost:%d' % port_web}
if not cle or origine not in miennes:
print(' refus : origine « %s »' % (origine or 'absente'))
sock.sendall(b'HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n')
return
accept = base64.b64encode(hashlib.sha1((cle + GUID).encode()).digest()).decode()
sock.sendall(('HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n'
'Connection: Upgrade\r\nSec-WebSocket-Accept: ' + accept + '\r\n\r\n').encode())
with clients_lock:
clients.append(sock)
inscrit = True
print(' fenêtre connectée')
while True:
r = ws_lire(sock)
if r is None:
break
op, corps = r
if op == 8:
break
if op == 9:
sock.sendall(struct.pack('>BB', 0x8A, len(corps)) + corps)
elif op == 1:
try:
m = json.loads(corps.decode('utf-8'))
except Exception:
continue
if isinstance(m, dict) and m.get('t') == 'osc' and m.get('addr'):
sortie.sendto(osc_build(str(m['addr']), m.get('args') or []), cible_udp)
except Exception:
pass
finally:
with clients_lock:
if sock in clients:
clients.remove(sock)
try:
sock.close()
except Exception:
pass
if inscrit:
print(' fenêtre fermée')
def diffuser(obj):
trame = ws_frame(json.dumps(obj, ensure_ascii=False))
with clients_lock:
morts = []
for c in clients:
try:
c.sendall(trame)
except Exception:
morts.append(c)
for c in morts:
clients.remove(c)
# ═══════════════════════════════════════════════════════════════════════════════════════════
def main(argv):
port_osc, port_web = PORT_OSC, PORT_WEB
cible_udp = SORTIE_DEFAUT
i = 1
while i < len(argv):
a = argv[i]
if a in ('-h', '--aide', '--help'):
print(__doc__)
return 0
if a == '--osc' and i + 1 < len(argv):
port_osc = int(argv[i + 1]); i += 2; continue
if a == '--web' and i + 1 < len(argv):
port_web = int(argv[i + 1]); i += 2; continue
if a == '--sortie' and i + 1 < len(argv):
h, _, p = argv[i + 1].partition(':')
cible_udp = (h or '127.0.0.1', int(p or 9132)); i += 2; continue
print('option inconnue : ' + a); return 2
sortie = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
# 127.0.0.1 et jamais 0.0.0.0 : le pont ne s'expose pas au réseau.
srv.bind(('127.0.0.1', port_web))
except OSError as e:
print('Impossible d\'écouter sur le port %d : %s' % (port_web, e))
print('Un autre pont tourne déjà ? Sinon : --web ')
return 1
srv.listen(8)
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
udp.bind(('0.0.0.0', port_osc))
except OSError as e:
print('Impossible d\'écouter l\'OSC sur le port %d : %s' % (port_osc, e))
return 1
def accepter():
while True:
try:
s, _ = srv.accept()
except OSError:
return
threading.Thread(target=servir,
args=(s, port_web, port_osc, sortie, cible_udp),
daemon=True).start()
threading.Thread(target=accepter, daemon=True).start()
print('┌─ Pont OSC EarMotion %s' % VERSION)
print('│ OSC/UDP entrant : %d' % port_osc)
print('│ Retour OSC : %s:%d' % cible_udp)
print('│ Fenêtre du pont : http://127.0.0.1:%d' % port_web)
print('└─ Dans EarMotion : panneau « OSC » → Connecter le pont. (Ctrl-C pour arrêter)')
while True:
data, _ = udp.recvfrom(65535)
for addr, args in osc_parse(data):
compteur['recus'] += 1
diffuser({'t': 'osc', 'addr': addr, 'args': args})
if __name__ == '__main__':
try:
sys.exit(main(sys.argv) or 0)
except KeyboardInterrupt:
print('\nPont arrêté.')