Compare commits
1 commit
Author | SHA1 | Date | |
---|---|---|---|
381bb143fc |
2 changed files with 137 additions and 0 deletions
6
main.py
6
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()
|
||||
|
|
131
netplay.py
Normal file
131
netplay.py
Normal 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()
|
Loading…
Reference in a new issue