Compare commits

...

1 commit

Author SHA1 Message Date
381bb143fc WIP: Netplay 2023-07-20 14:51:19 -04:00
2 changed files with 137 additions and 0 deletions

View file

@ -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()

131
netplay.py Normal file
View file

@ -0,0 +1,131 @@
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()
class DumbNetPlayer(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")
while True:
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 i2s_callback(self, arg):
...
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 = 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):
...
def start(self):
while True:
self.i2s.write(self.tone)
def test():
return DumbNetPlayer()