WIP: Netplay

main.py

@@ -303,6 +303,12 @@ def loop():
         oled.show()
     state.clean()
 
+def net():
+    import netplay
+    switch.select(3)
+    pot.write(0, 128)
+    pot.write(1, 128)
+    netplay.test()
 
 while True:
     loop()

netplay.py

@@ -0,0 +1,99 @@
+import machine
+import math
+import socket
+import struct
+
+I2S_ID = 1
+WS_PIN = 27
+SD_PIN = 26
+SCK_PIN = 25
+BUFFER_SIZE = 40000
+
+
+class Player:
+    def __init__(self, bits=16, format=machine.I2S.STEREO, rate=22050):
+        self.i2s = machine.I2S(
+            I2S_ID,
+            ws=machine.Pin(WS_PIN),
+            sd=machine.Pin(SD_PIN),
+            sck=machine.Pin(SCK_PIN),
+            mode=machine.I2S.TX,
+            bits=bits,
+            format=format,
+            rate=rate,
+            ibuf=BUFFER_SIZE,
+        )
+        self.silence = bytearray(1000)
+        self.samples = bytearray(10000)
+        self.samples_mv = memoryview(self.samples)
+        self.start()
+
+    def start(self):
+        self.i2s.irq(self.i2s_callback)
+        self.i2s.write(self.silence)
+
+    def i2s_callback(self, arg):
+        self.i2s.write(self.silence)
+
+    def stop(self):
+        self.i2s.deinit()
+
+class NetPlayer(Player):
+    def __init__(self, bits=16, format=machine.I2S.STEREO, rate=22050):
+        self.socket = socket.socket()
+        super().__init__(bits=bits, format=format, rate=rate)
+
+    def start(self):
+        addrinfo = socket.getaddrinfo("zephyrus.local", 8001, 0, socket.SOCK_STREAM)
+        print("Opening socket to ", addrinfo)
+        self.socket.connect(addrinfo[0][-1])
+        self.socket.setblocking(False)
+        print("Connected")
+        super().start()
+
+    def i2s_callback(self, arg):
+        num_read = self.socket.readinto(self.samples_mv)
+        if num_read and num_read > 0:
+            print("CB AUDIO", num_read)
+            self.i2s.write(self.samples_mv[:num_read])
+        else:
+            print("CB SILENCE")
+            self.i2s.write(self.silence)
+
+    def stop(self):
+        self.socket.close()
+        super().stop()
+
+def make_tone(rate, bits, frequency):
+    # create a buffer containing the pure tone samples
+    samples_per_cycle = rate // frequency
+    sample_size_in_bytes = bits // 8
+    samples = bytearray(samples_per_cycle * sample_size_in_bytes)
+    volume_reduction_factor = 2 # 32
+    range = pow(2, bits) // 2 // volume_reduction_factor
+
+    if bits == 16:
+        format = "<h"
+    else:  # assume 32 bits
+        format = "<l"
+
+    for i in range(samples_per_cycle):
+        sample = range + int(
+            (range - 1) * math.sin(2 * math.pi * i / samples_per_cycle)
+        )
+        struct.pack_into(format, samples, i * sample_size_in_bytes, sample)
+
+    return samples
+
+
+class TonePlayer(Player):
+    def __init__(self):
+        self.tone = make_tone(22050, 16, 440)
+        super().__init__(format=machine.I2S.MONO)
+
+    def i2s_callback(self, arg):
+        self.i2s.write(self.tone)
+
+
+def test():
+    return TonePlayer()