Initial Commit

This commit is contained in:
colaclanth 2019-07-15 17:28:23 +01:00
parent 08d259ff79
commit 282d933517
6 changed files with 572 additions and 0 deletions

2
sstv/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from .command import SSTVCommand
from .decode import SSTVDecoder

9
sstv/__main__.py Normal file
View file

@ -0,0 +1,9 @@
#!/usr/bin/env python
import sstv
def main():
with sstv.SSTVCommand() as prog:
prog.start()
if __name__ == "__main__":
main()

114
sstv/command.py Normal file
View file

@ -0,0 +1,114 @@
#!usr/bin/env python
from sys import exit
from os import path
from .decode import SSTVDecoder
import argparse
class SSTVCommand(object):
""" Main class to handle the command line features """
examples_of_use = """
examples:
Decode local SSTV audio file named 'audio.ogg' to 'result.png':
$ sstv -d audio.ogg
Decode SSTV audio file in /tmp to './image.jpg':
$ sstv -d /tmp/signal.wav -o ./image.jpg
"""
def __init__(self):
""" Handle command line arguments """
self._audio_file = None
self._output_file = None
self.args = self.parse_args()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, traceback):
self.close()
def __del__(self):
self.close()
def init_args(self):
""" Initialise argparse parser """
version = "sstv 0.1"
parser = argparse.ArgumentParser(
prog="sstv",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=self.examples_of_use)
parser.add_argument("-d", "--decode", type=argparse.FileType('rb'),
help="SSTV audio file to decode",
dest="audio_file")
parser.add_argument("-o", "--output", type=argparse.FileType('wb'),
help="desination of output file",
default="result.png",
dest="output_file")
parser.add_argument("-V", "--version", action="version", version=version)
parser.add_argument("-v", "--verbose", action="count", default=1,
help="increase output to the terminal")
parser.add_argument("--list-modes", action="store_true",
dest="list_modes",
help="list supported SSTV modes")
parser.add_argument("--list-audio-formats", action="store_true",
dest="list_audio_formats",
help="list supported audio file formats")
parser.add_argument("--list-image-formats", action="store_true",
dest="list_image_formats",
help="list supported image file formats")
return parser
def parse_args(self):
""" Parse command line arguments """
parser = self.init_args()
args = parser.parse_args()
self._audio_file = args.audio_file
self._output_file = args.output_file
if args.list_modes:
self.list_supported_modes()
exit(0)
if args.list_audio_formats:
self.list_supported_audio_formats()
exit(0)
if args.list_image_formats:
self.list_supported_image_formats()
exit(0)
if self._audio_file is None:
parser.print_help()
exit(2)
return args
def start(self):
with SSTVDecoder(self._audio_file) as sstv:
img = sstv.decode()
try:
img.save(self._output_file)
except KeyError:
img.save("result.png")
def close(self):
""" Closes any input/output files if they exist """
if self._output_file is not None and not self._output_file.closed:
self._output_file.close()
if self._audio_file is not None and not self._audio_file.closed:
self._audio_file.close()
def list_supported_modes(self):
# FIXME hardcode
print("M1, S1")
def list_supported_audio_formats(self):
# FIXME hardcode
print("ogg, wav")
def list_supported_image_formats(self):
# FIXME hardcode
print("png, jpeg")

45
sstv/common.py Normal file
View file

@ -0,0 +1,45 @@
#!usr/bin/env python
from sys import stderr, stdout
from os import get_terminal_size
def log_message(message="", show=True, err=False, recur=False, prefix=True):
if not show:
return
out = stdout
if err:
out = stderr
end = '\n'
if recur:
end = '\r'
cols = get_terminal_size().columns
if cols < len(message):
message = message[:cols]
if prefix:
message = ' '.join(["[SSTV]", message])
print(message, file=out, end=end)
def progress_bar(progress, complete, message="", show=True):
if not show:
return
message = ' '.join(["[SSTV]", message])
cols = get_terminal_size().columns
percent_on = True
level = progress / complete
bar_size = min(cols - len(message) - 10, 100)
bar = ""
if bar_size > 5:
fill_size = round(bar_size * level)
bar = "[{}]".format(''.join(['#' * fill_size,
'.' * (bar_size - fill_size)]))
elif bar_size < -3:
percent_on = False
percent = ""
if percent_on:
percent = "{:4.0f}%".format(level * 100)
align_size = cols - len(message) - len(percent)
not_end = not progress == complete
log_message("{}{:>{width}}{}".format(message, bar, percent,
width=align_size), recur=not_end, prefix=False)

304
sstv/decode.py Normal file
View file

@ -0,0 +1,304 @@
#!usr/bin/env python
from . import spec
from .common import log_message, progress_bar
from PIL import Image, ImageDraw
from scipy.signal.windows import hann
import soundfile
import numpy as np
class SSTVDecoder(object):
def __init__(self, audio_file):
"""
Initialise SSTV decoder
audio_data - Can be a path to an audio file OR a tuple containing
samples and the sampling rate of audio data
"""
self.log_basic = True
self.mode = None
self._audio_file = audio_file
self._samples, self._sample_rate = soundfile.read(self._audio_file)
if self._samples.ndim > 1: # convert to mono if stereo
self._samples = self._samples.mean(axis=1)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, traceback):
self.close()
def __del__(self):
self.close()
def decode(self, skip=0):
"""
Attempts to decode the audio data as an SSTV signal
Returns a PIL image on success, and None on error
"""
header_start = self._find_header()
if header_start is None:
return None
vis_start = header_start + round(spec.HDR_SIZE * self._sample_rate)
vis_end = vis_start + round(spec.VIS_BIT_SIZE * 9 * self._sample_rate)
vis_section = self._samples[vis_start:vis_end]
self.mode = self._decode_vis(vis_section)
transmission_area = self._samples[vis_end:]
image_data = self._decode_image_data(transmission_area)
if image_data is None:
return None
return self._draw_image(image_data)
def close(self):
""" Closes any input files if they exist """
if self._audio_file is not None and not self._audio_file.closed:
self._audio_file.close()
def _barycentric_peak_interp(bins, x):
# Takes x as the index of the largest bin and interpolates the
# x value of the peak using neighbours in the bins array
# Make sure data is in bounds
if x <= 0:
y1 = bins[x]
else:
y1 = bins[x-1]
if x + 1 >= len(bins):
y3 = bins[x]
else:
y3 = bins[x+1]
denom = y3 + bins[x] + y1
if denom == 0:
return 0 # erroneous
return (y3 - y1) / denom + x
def _peak_fft_freq(self, data):
# Finds the peak frequency from a section of audio data
windowed_data = data * hann(len(data))
fft = np.abs(np.fft.rfft(windowed_data))
# Get index of bin with highest magnitude
x = np.argmax(fft)
# Interpolated peak frequency
peak = SSTVDecoder._barycentric_peak_interp(fft, x)
# Return frequency in hz
return peak * self._sample_rate / len(windowed_data)
def _find_header(self):
# Finds the approx sample of the calibration header
break_sample = round(spec.BREAK_OFFSET * self._sample_rate)
leader_sample = round(spec.LEADER_OFFSET * self._sample_rate)
vis_start_sample = round(spec.VIS_START_OFFSET * self._sample_rate)
header_size = round(spec.HDR_SIZE * self._sample_rate)
window_size = round(spec.HDR_WINDOW_SIZE * self._sample_rate)
jump_size = round(0.002 * self._sample_rate) # check every 2ms
# The margin of error created here will be negligible when decoding the
# vis due to each bit having a length of 30ms. We fix this error margin
# when decoding the image by aligning each sync pulse
current_sample = 0
for sample in self._samples[:-header_size+window_size]:
if current_sample % (jump_size * 256) == 0:
log_message("Searching for calibration header... {:.1f}s".format(
current_sample / self._sample_rate), self.log_basic, recur=True)
search_area = self._samples[current_sample:current_sample+header_size]
if abs(self._peak_fft_freq(search_area[0:window_size]) - 1900) < 50 and \
abs(self._peak_fft_freq(search_area[break_sample:break_sample+window_size]) - 1200) < 50 and \
abs(self._peak_fft_freq(search_area[leader_sample:leader_sample+window_size]) - 1900) < 50 and \
abs(self._peak_fft_freq(search_area[vis_start_sample:vis_start_sample+window_size]) - 1200) < 50:
log_message("Searching for calibration header... Found!{:>4}".format(
' '), self.log_basic)
return current_sample
current_sample += jump_size
log_message()
log_message("Couldn't find SSTV header in the given audio file.")
return None
def _decode_vis(self, vis_section):
# Decodes the vis value from the audio data and returns the SSTV mode
bit_size = round(spec.VIS_BIT_SIZE * self._sample_rate)
vis_bits = []
for bit_idx in range(8):
bit_offset = bit_idx * bit_size
freq = self._peak_fft_freq(vis_section[bit_offset:bit_offset+bit_size])
vis_bits.append(int(freq <= 1200))
# check for even parity in last bit
parity = sum(vis_bits) % 2 == 0
if not parity:
log_message("Error decoding VIS header (incorrect parity bit)")
return None
# LSB first so we must reverse and ignore the parity bit
vis_value = 0
for bit in vis_bits[-2::-1]:
vis_value = (vis_value << 1) | bit
if vis_value not in spec.VIS_MAP:
log_message("SSTV mode is unsupported (VIS: {})".format(vis_value))
return None
mode = spec.VIS_MAP[vis_value]
log_message("Detected SSTV mode {}".format(mode.NAME))
return mode
def _calc_lum(freq):
# Converts SSTV pixel frequency range into 0-255 luminance byte
lum = int(round((freq - 1500) / 3.1372549))
if lum > 255:
return 255
elif lum < 0:
return 0
else:
return lum
#def _yuv_to_rgb(y, ry, by):
# red = 0.003906 * ((298.082 * (y - 16.0)) + (408.583 * (ry - 128.0)))
# green = 0.003906 * ((298.082 * (y - 16.0)) + (-100.291 * (by - 128.0)) \
# + (-208.12 * (ry - 128.0)))
# blue = 0.003906 * ((298.082 * (y - 16.0)) + (516.411 * (by - 128.0)))
# rgb = round(red), round(green), round(blue)
# return rgb
def _align_sync(self, align_section, start_of_sync=True):
# Returns sample where the beginning of the sync pulse was found
sync_window = round(self.mode.SYNC_PULSE * 1.4 * self._sample_rate)
search_end = len(align_section) - sync_window
current_sample = 0
while current_sample < search_end:
freq = self._peak_fft_freq(align_section[current_sample:current_sample+sync_window])
if freq > 1350:
break
current_sample += 1
end_sync = current_sample + (sync_window // 2)
if start_of_sync:
return end_sync - round(self.mode.SYNC_PULSE * self._sample_rate)
else:
return end_sync
def _decode_image_data(self, transmission):
# Decodes image data from the transmission section of an sstv signal
if self.mode == spec.M1:
window_factor = 2.34
if self.mode == spec.S1:
window_factor = 2.48
if self.mode == spec.R36:
window_factor = 7.83
pixel_window = round(self.mode.PIXEL_TIME * window_factor * self._sample_rate)
centre_window_time = (self.mode.PIXEL_TIME * window_factor) / 2
image_data = []
if self.mode.HAS_START_SYNC:
# Start at the end of the initial sync pulse
seq_start = self._align_sync(transmission, start_of_sync=False)
else:
seq_start = 0
for line in range(self.mode.LINE_COUNT):
image_data.append([])
if self.mode.CHAN_SYNC > 0 and line == 0:
# align seq_start to the beginning of the sync pulse in the past
seq_start -= round((self.mode.CHAN_OFFSETS[self.mode.CHAN_SYNC] \
+ self.mode.SCAN_TIME) * self._sample_rate)
for chan in range(self.mode.CHAN_COUNT):
image_data[line].append([])
if chan == self.mode.CHAN_SYNC:
if line > 0 or chan > 0:
seq_start += round(self.mode.LINE_TIME * self._sample_rate)
# align to start of sync pulse
seq_start += self._align_sync(transmission[seq_start:])
pixel_time = self.mode.PIXEL_TIME
if self.mode.HAS_MERGE_SCAN:
if chan % 2 == 1:
pixel_time = self.mode.MERGE_PIXEL_TIME
pixel_window = round(pixel_time * window_factor * self._sample_rate)
centre_window_time = (pixel_time * window_factor) / 2
for px in range(self.mode.LINE_WIDTH):
chan_offset = self.mode.CHAN_OFFSETS[chan]
px_sample = round(seq_start + (chan_offset + px * pixel_time \
- centre_window_time) * self._sample_rate)
freq = self._peak_fft_freq(transmission[px_sample:px_sample+pixel_window])
image_data[line][chan].append(SSTVDecoder._calc_lum(freq))
progress_bar(line, self.mode.LINE_COUNT - 1,
"Decoding image... ", self.log_basic)
log_message("...Done!")
return image_data
def _draw_image(self, image_data):
# Renders the image from the decoded sstv signal
if self.mode.COLOR == spec.COL_FMT.YUV:
col_mode = "YCbCr"
else:
col_mode = "RGB"
image = Image.new(col_mode, (self.mode.LINE_WIDTH, self.mode.LINE_COUNT))
pixel_data = image.load()
for y in range(self.mode.LINE_COUNT):
ryby = y % 2
for x in range(self.mode.LINE_WIDTH):
if self.mode.COLOR == spec.COL_FMT.GBR:
pixel = image_data[y][2][x], image_data[y][0][x], image_data[y][1][x]
elif self.mode.COLOR == spec.COL_FMT.YUV:
pixel = (image_data[y][0][x], image_data[y-ryby][1][x],
image_data[y-(ryby-1)][1][x])
else:
pixel = image_data[y][0][x], image_data[y][1][x], image_data[y][2][x]
pixel_data[x, y] = pixel
if self.mode.COLOR == spec.COL_FMT.YUV:
image = image.convert("RGB")
return image

98
sstv/spec.py Normal file
View file

@ -0,0 +1,98 @@
#!usr/bin/env python
from enum import Enum
BREAK_OFFSET = 0.300
LEADER_OFFSET = 0.010 + BREAK_OFFSET
VIS_START_OFFSET = 0.300 + LEADER_OFFSET
HDR_SIZE = 0.030 + VIS_START_OFFSET
HDR_WINDOW_SIZE = 0.010
VIS_BIT_SIZE = 0.030
class COL_FMT(Enum):
RGB = 1
GBR = 2
YUV = 3
BW = 4
class SSTV(object):
HAS_START_SYNC = False
HAS_MERGE_SCAN = False
CHAN_SYNC = 0
class M1(SSTV):
NAME = "Martin 1"
VIS_CODE = 44
COLOR = COL_FMT.GBR
LINE_WIDTH = 320
LINE_COUNT = 256
SCAN_TIME = 0.146432
SYNC_PULSE = 0.004862
SYNC_PORCH = 0.000572
SEP_PULSE = 0.000572
CHAN_COUNT = 3
CHAN_TIME = SEP_PULSE + SCAN_TIME
# CHAN_OFFSETS = [SYNC_PULSE + SYNC_PORCH + c * CHAN_TIME for c in range(CHAN_COUNT)]
CHAN_OFFSETS = [SYNC_PULSE + SYNC_PORCH]
CHAN_OFFSETS.append(CHAN_OFFSETS[0] + CHAN_TIME)
CHAN_OFFSETS.append(CHAN_OFFSETS[1] + CHAN_TIME)
LINE_TIME = SYNC_PULSE + SYNC_PORCH + 3 * CHAN_TIME
PIXEL_TIME = SCAN_TIME / LINE_WIDTH
class S1(SSTV):
NAME = "Scottie 1"
HAS_START_SYNC = True
VIS_CODE = 60
COLOR = COL_FMT.GBR
LINE_WIDTH = 320
LINE_COUNT = 256
SCAN_TIME = 0.138240
SYNC_PULSE = 0.009000
SYNC_PORCH = 0.001500
SEP_PULSE = 0.001500
CHAN_COUNT = 3
CHAN_SYNC = 2
CHAN_TIME = SEP_PULSE + SCAN_TIME
CHAN_OFFSETS = [SYNC_PULSE + SYNC_PORCH + CHAN_TIME]
CHAN_OFFSETS.append(CHAN_OFFSETS[0] + CHAN_TIME)
CHAN_OFFSETS.append(SYNC_PULSE + SYNC_PORCH)
LINE_TIME = SYNC_PULSE + 3 * CHAN_TIME
PIXEL_TIME = SCAN_TIME / LINE_WIDTH
class R36(SSTV):
NAME = "Robot 36"
HAS_MERGE_SCAN = True
VIS_CODE = 8
COLOR = COL_FMT.YUV
LINE_WIDTH = 320
LINE_COUNT = 240
SCAN_TIME = 0.088000
MERGE_SCAN_TIME = 0.044000
SYNC_PULSE = 0.009000
SYNC_PORCH = 0.003000
SEP_PULSE = 0.004500
SEP_PORCH = 0.001500
CHAN_COUNT = 2
CHAN_TIME = SEP_PULSE + SCAN_TIME
CHAN_OFFSETS = [SYNC_PULSE + SYNC_PORCH]
CHAN_OFFSETS.append(CHAN_OFFSETS[0] + CHAN_TIME + SEP_PORCH)
LINE_TIME = CHAN_OFFSETS[1] + MERGE_SCAN_TIME
PIXEL_TIME = SCAN_TIME / LINE_WIDTH
MERGE_PIXEL_TIME = MERGE_SCAN_TIME / LINE_WIDTH
VIS_MAP = {44: M1, 60: S1, 8: R36}