Pulse width encoding
Working hello world transfer in sim and irl
This commit is contained in:
Joppe Blondel
+95
-93
@@ -1,133 +1,135 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import argparse
|
||||
import math
|
||||
import struct
|
||||
import wave
|
||||
from pathlib import Path
|
||||
|
||||
class FSKWriter:
|
||||
def __init__(self, sample_rate: int, c_low: int, c_high: int, f_low: float, f_high: float, amplitude: float, block_spacing: float):
|
||||
|
||||
class PulseWriter:
|
||||
def __init__(
|
||||
self,
|
||||
sample_rate: int,
|
||||
t_short_us: float,
|
||||
t_long_us: float,
|
||||
amplitude: float,
|
||||
) -> None:
|
||||
self.sample_rate = sample_rate
|
||||
self.c_low = c_low
|
||||
self.c_high = c_high
|
||||
self.f_low = f_low
|
||||
self.f_high = f_high
|
||||
self.amplitude = amplitude
|
||||
self.samples_per_low = (sample_rate * c_low) // f_low
|
||||
self.samples_per_high = (sample_rate * c_high) // f_high
|
||||
self.phase = 0.0
|
||||
self.block_spacing = block_spacing
|
||||
self.short_samples = self._duration_samples(t_short_us)
|
||||
self.long_samples = self._duration_samples(t_long_us)
|
||||
self.high = round(32767 * amplitude)
|
||||
self.low = -self.high
|
||||
self.samples = bytearray()
|
||||
|
||||
self.space(1.0)
|
||||
self.space(0.1)
|
||||
|
||||
def finalize(self):
|
||||
self.space(1.0)
|
||||
def space(self, time: float) -> None:
|
||||
self._level(0, int(time*self.sample_rate))
|
||||
|
||||
def space(self, time:float = 0.03):
|
||||
self.samples.extend(b'\0\0'*int(time*self.sample_rate))
|
||||
def _duration_samples(self, duration_us: float) -> int:
|
||||
sample_count = round(duration_us * self.sample_rate / 1_000_000)
|
||||
if sample_count < 1:
|
||||
raise ValueError(
|
||||
f"pulse duration {duration_us:g} us is shorter than one sample"
|
||||
)
|
||||
return sample_count
|
||||
|
||||
def _tone(self, frequency: float, sample_count: int) -> None:
|
||||
phase_step = 2.0 * math.pi * frequency / self.sample_rate
|
||||
scale = int(32767 * self.amplitude)
|
||||
phase = self.phase
|
||||
out = self.samples
|
||||
def _level(self, value: int, sample_count: int) -> None:
|
||||
self.samples.extend(struct.pack("<h", value) * sample_count)
|
||||
|
||||
for _ in range(int(sample_count)):
|
||||
sample = int(round(scale * math.sin(phase)))
|
||||
out.extend(struct.pack("<h", sample))
|
||||
phase += phase_step
|
||||
if phase >= 2.0 * math.pi:
|
||||
phase -= 2.0 * math.pi
|
||||
def short_pulse(self) -> None:
|
||||
self._level(self.high, self.short_samples)
|
||||
self._level(self.low, self.long_samples)
|
||||
|
||||
self.phase = phase
|
||||
def long_pulse(self) -> None:
|
||||
self._level(self.high, self.long_samples)
|
||||
self._level(self.low, self.short_samples)
|
||||
|
||||
def leader(self) -> None:
|
||||
self._tone(self.f_high, (200*self.sample_rate)//self.f_high)
|
||||
self._tone(self.f_low, (200*self.sample_rate)//self.f_low)
|
||||
for _ in range(8):
|
||||
self.bit(1)
|
||||
self.bit(0)
|
||||
self.byte(0xf0)
|
||||
def marker(self) -> None:
|
||||
self._level(self.high, 2*self.long_samples)
|
||||
self._level(self.low, 2*self.long_samples)
|
||||
|
||||
def bit(self, value: int) -> None:
|
||||
if value:
|
||||
self._tone(self.f_high, self.samples_per_high)
|
||||
self.long_pulse()
|
||||
else:
|
||||
self._tone(self.f_low, self.samples_per_low)
|
||||
self.short_pulse()
|
||||
|
||||
def byte(self, value: int) -> None:
|
||||
# MSB first
|
||||
for bit_no in range(7, -1, -1):
|
||||
self.bit((value >> bit_no) & 1)
|
||||
# Transmit bytes most-significant bit first.
|
||||
for bit_number in range(7, -1, -1):
|
||||
self.bit((value >> bit_number) & 1)
|
||||
|
||||
def bytes(self, data: bytes) -> None:
|
||||
for value in data:
|
||||
def bytes(self, value: bytes) -> None:
|
||||
for byte in value:
|
||||
self.byte(byte)
|
||||
|
||||
def leader(self) -> None:
|
||||
self.space(0.2)
|
||||
for _ in range(50):
|
||||
self._level(self.high, self.short_samples)
|
||||
self._level(self.low, self.short_samples)
|
||||
for _ in range(50):
|
||||
self._level(self.high, self.long_samples)
|
||||
self._level(self.low, self.long_samples)
|
||||
self.marker()
|
||||
for value in (0x55, 0x5A, 0xC3):
|
||||
self.byte(value)
|
||||
|
||||
def block(self, nr: int, data: bytes) -> None:
|
||||
print(f'> {data}')
|
||||
self.leader()
|
||||
self.byte(nr)
|
||||
l = len(data)
|
||||
self.byte(l>>8)
|
||||
self.byte(l&0xff)
|
||||
#self.samples.extend(b'\0'*int(self.samples_per_low*8))
|
||||
self.bytes(data)
|
||||
self.byte(self.chsum(data))
|
||||
self.space(self.block_spacing)
|
||||
|
||||
def chsum(self, data: bytes) -> int:
|
||||
cs = 0
|
||||
for d in data:
|
||||
cs += int(d)
|
||||
return cs
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Encode a binary file as two tone FSK WAV")
|
||||
parser.add_argument("input", type=Path, help="Binary input file")
|
||||
parser.add_argument("output", type=Path, help="output WAV; default <input>.wav")
|
||||
parser.add_argument("--sample-rate", type=int, default=48000)
|
||||
parser.add_argument("--low", type=float, default=2000.0, help="low tone in Hz")
|
||||
parser.add_argument("--high", type=float, default=4000.0, help="high tone in Hz")
|
||||
parser.add_argument("--low-cycles", type=int, default=2, help="low tone cycle count for a 0")
|
||||
parser.add_argument("--high-cycles", type=int, default=4, help="high tone cycle count for a 1")
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate the pulse-width encoded tape leader"
|
||||
)
|
||||
parser.add_argument("output", type=Path, help="output WAV file")
|
||||
parser.add_argument("--sample-rate", type=int, default=48_000)
|
||||
parser.add_argument(
|
||||
"--t-short",
|
||||
type=float,
|
||||
default=250.0,
|
||||
metavar="MICROSECONDS",
|
||||
help="short pulse interval in microseconds (default: 250)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--t-long",
|
||||
type=float,
|
||||
default=500.0,
|
||||
metavar="MICROSECONDS",
|
||||
help="long pulse interval in microseconds (default: 500)",
|
||||
)
|
||||
parser.add_argument("--amplitude", type=float, default=0.75)
|
||||
parser.add_argument("--block-size", type=int, default=128)
|
||||
parser.add_argument("--block-spacing", type=float, default=1.0)
|
||||
args = parser.parse_args()
|
||||
|
||||
output_path = args.output or args.input.with_name(args.input.stem + ".wav")
|
||||
if args.sample_rate <= 0:
|
||||
parser.error("--sample-rate must be positive")
|
||||
if args.t_short <= 0 or args.t_long <= 0:
|
||||
parser.error("pulse intervals must be positive")
|
||||
if args.t_short >= args.t_long:
|
||||
parser.error("--t-short must be less than --t-long")
|
||||
if not 0.0 <= args.amplitude <= 1.0:
|
||||
parser.error("--amplitude must be between 0 and 1")
|
||||
|
||||
tones = FSKWriter(args.sample_rate, args.low_cycles, args.high_cycles, args.low, args.high, args.amplitude, args.block_spacing)
|
||||
try:
|
||||
pulses = PulseWriter(
|
||||
args.sample_rate,
|
||||
args.t_short,
|
||||
args.t_long,
|
||||
args.amplitude,
|
||||
)
|
||||
except ValueError as error:
|
||||
parser.error(str(error))
|
||||
|
||||
# Create header block
|
||||
nrblocks = math.ceil(os.path.getsize(args.input)/args.block_size)
|
||||
print(f"Number of blocks {nrblocks}*2")
|
||||
pulses.leader()
|
||||
pulses.bytes(b'Hello world!\0')
|
||||
|
||||
tones.block(0, struct.pack('<H', nrblocks*2) + args.input.name.encode() + b'\0')
|
||||
tones.space(1.0)
|
||||
|
||||
block = 1
|
||||
with open(args.input, 'rb') as f:
|
||||
while True:
|
||||
data = f.read(args.block_size)
|
||||
if not len(data):
|
||||
break
|
||||
tones.block(block, data)
|
||||
tones.block(block, data)
|
||||
block += 1
|
||||
|
||||
tones.finalize()
|
||||
with wave.open(str(output_path), "wb") as wav:
|
||||
pulses.space(1.0)
|
||||
with wave.open(str(args.output), "wb") as wav:
|
||||
wav.setnchannels(1)
|
||||
wav.setsampwidth(2)
|
||||
wav.setframerate(args.sample_rate)
|
||||
wav.writeframes(tones.samples)
|
||||
wav.writeframes(pulses.samples)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user