spacecruft
/
sstvdec
1
0
Fork 0
Code Releases Activity

Improved formatting

master
colaclanth 2019-07-15 17:46:41 +01:00
parent 282d933517
commit 8ecf722670
6 changed files with 58 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
sstv/__init__.py
View File

@ -1,2 +1,4 @@
#!/usr/bin/env python
from .command import SSTVCommand from .command import SSTVCommand
from .decode import SSTVDecoder from .decode import SSTVDecoder

3
sstv/__main__.py
View File

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

10
sstv/command.py
View File

@ -1,9 +1,10 @@
#!usr/bin/env python #!usr/bin/env python
from sys import exit from sys import exit
from os import path
from .decode import SSTVDecoder from .decode import SSTVDecoder
import argparse import argparse
class SSTVCommand(object): class SSTVCommand(object):
""" Main class to handle the command line features """ """ Main class to handle the command line features """
@ -48,7 +49,8 @@ examples:
help="desination of output file", help="desination of output file",
default="result.png", default="result.png",
dest="output_file") dest="output_file")
parser.add_argument("-V", "--version", action="version", version=version) parser.add_argument("-V", "--version", action="version",
version=version)
parser.add_argument("-v", "--verbose", action="count", default=1, parser.add_argument("-v", "--verbose", action="count", default=1,
help="increase output to the terminal") help="increase output to the terminal")
parser.add_argument("--list-modes", action="store_true", parser.add_argument("--list-modes", action="store_true",
@ -69,7 +71,7 @@ examples:
self._audio_file = args.audio_file self._audio_file = args.audio_file
self._output_file = args.output_file self._output_file = args.output_file
if args.list_modes: if args.list_modes:
self.list_supported_modes() self.list_supported_modes()
exit(0) exit(0)
@ -93,7 +95,7 @@ examples:
img.save(self._output_file) img.save(self._output_file)
except KeyError: except KeyError:
img.save("result.png") img.save("result.png")
def close(self): def close(self):
""" Closes any input/output files if they exist """ """ Closes any input/output files if they exist """
if self._output_file is not None and not self._output_file.closed: if self._output_file is not None and not self._output_file.closed:

13
sstv/common.py
View File

@ -1,7 +1,9 @@
#!usr/bin/env python #!usr/bin/env python
from sys import stderr, stdout from sys import stderr, stdout
from os import get_terminal_size from os import get_terminal_size
def log_message(message="", show=True, err=False, recur=False, prefix=True): def log_message(message="", show=True, err=False, recur=False, prefix=True):
if not show: if not show:
return return
@ -18,7 +20,8 @@ def log_message(message="", show=True, err=False, recur=False, prefix=True):
message = ' '.join(["[SSTV]", message]) message = ' '.join(["[SSTV]", message])
print(message, file=out, end=end) print(message, file=out, end=end)
def progress_bar(progress, complete, message="", show=True): def progress_bar(progress, complete, message="", show=True):
if not show: if not show:
return return
@ -38,8 +41,8 @@ def progress_bar(progress, complete, message="", show=True):
percent = "" percent = ""
if percent_on: if percent_on:
percent = "{:4.0f}%".format(level * 100) percent = "{:4.0f}%".format(level * 100)
align_size = cols - len(message) - len(percent) align = cols - len(message) - len(percent)
not_end = not progress == complete not_end = not progress == complete
log_message("{}{:>{width}}{}".format(message, bar, percent, log_message("{}{:>{width}}{}".format(message, bar, percent, width=align),
width=align_size), recur=not_end, prefix=False) recur=not_end, prefix=False)

64
sstv/decode.py
View File

@ -1,7 +1,8 @@
#!usr/bin/env python #!usr/bin/env python
from . import spec from . import spec
from .common import log_message, progress_bar from .common import log_message, progress_bar
from PIL import Image, ImageDraw from PIL import Image
from scipy.signal.windows import hann from scipy.signal.windows import hann
import soundfile import soundfile
import numpy as np import numpy as np
@ -12,7 +13,7 @@ class SSTVDecoder(object):
def __init__(self, audio_file): def __init__(self, audio_file):
""" """
Initialise SSTV decoder Initialise SSTV decoder
audio_data - Can be a path to an audio file OR a tuple containing audio_data - Can be a path to an audio file OR a tuple containing
samples and the sampling rate of audio data samples and the sampling rate of audio data
""" """
@ -39,7 +40,7 @@ class SSTVDecoder(object):
def decode(self, skip=0): def decode(self, skip=0):
""" """
Attempts to decode the audio data as an SSTV signal Attempts to decode the audio data as an SSTV signal
Returns a PIL image on success, and None on error Returns a PIL image on success, and None on error
""" """
@ -61,7 +62,7 @@ class SSTVDecoder(object):
return None return None
return self._draw_image(image_data) return self._draw_image(image_data)
def close(self): def close(self):
""" Closes any input files if they exist """ """ Closes any input files if they exist """
if self._audio_file is not None and not self._audio_file.closed: if self._audio_file is not None and not self._audio_file.closed:
@ -149,7 +150,8 @@ class SSTVDecoder(object):
for bit_idx in range(8): for bit_idx in range(8):
bit_offset = bit_idx * bit_size bit_offset = bit_idx * bit_size
freq = self._peak_fft_freq(vis_section[bit_offset:bit_offset+bit_size]) section = vis_section[bit_offset:bit_offset+bit_size]
freq = self._peak_fft_freq(section)
vis_bits.append(int(freq <= 1200)) vis_bits.append(int(freq <= 1200))
# check for even parity in last bit # check for even parity in last bit
@ -182,14 +184,6 @@ class SSTVDecoder(object):
else: else:
return lum 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): def _align_sync(self, align_section, start_of_sync=True):
# Returns sample where the beginning of the sync pulse was found # Returns sample where the beginning of the sync pulse was found
sync_window = round(self.mode.SYNC_PULSE * 1.4 * self._sample_rate) sync_window = round(self.mode.SYNC_PULSE * 1.4 * self._sample_rate)
@ -197,7 +191,8 @@ class SSTVDecoder(object):
current_sample = 0 current_sample = 0
while current_sample < search_end: while current_sample < search_end:
freq = self._peak_fft_freq(align_section[current_sample:current_sample+sync_window]) section = align_section[current_sample:current_sample+sync_window]
freq = self._peak_fft_freq(section)
if freq > 1350: if freq > 1350:
break break
current_sample += 1 current_sample += 1
@ -219,7 +214,8 @@ class SSTVDecoder(object):
if self.mode == spec.R36: if self.mode == spec.R36:
window_factor = 7.83 window_factor = 7.83
pixel_window = round(self.mode.PIXEL_TIME * window_factor * self._sample_rate) pixel_window = round(self.mode.PIXEL_TIME * window_factor *
self._sample_rate)
centre_window_time = (self.mode.PIXEL_TIME * window_factor) / 2 centre_window_time = (self.mode.PIXEL_TIME * window_factor) / 2
image_data = [] image_data = []
@ -234,16 +230,18 @@ class SSTVDecoder(object):
image_data.append([]) image_data.append([])
if self.mode.CHAN_SYNC > 0 and line == 0: if self.mode.CHAN_SYNC > 0 and line == 0:
# align seq_start to the beginning of the sync pulse in the past # align seq_start to the beginning of the previous sync pulse
seq_start -= round((self.mode.CHAN_OFFSETS[self.mode.CHAN_SYNC] \ sync_offset = self.mode.CHAN_OFFSETS[self.mode.CHAN_SYNC]
+ self.mode.SCAN_TIME) * self._sample_rate) seq_start -= round((sync_offset + self.mode.SCAN_TIME)
* self._sample_rate)
for chan in range(self.mode.CHAN_COUNT): for chan in range(self.mode.CHAN_COUNT):
image_data[line].append([]) image_data[line].append([])
if chan == self.mode.CHAN_SYNC: if chan == self.mode.CHAN_SYNC:
if line > 0 or chan > 0: if line > 0 or chan > 0:
seq_start += round(self.mode.LINE_TIME * self._sample_rate) seq_start += round(self.mode.LINE_TIME *
self._sample_rate)
# align to start of sync pulse # align to start of sync pulse
seq_start += self._align_sync(transmission[seq_start:]) seq_start += self._align_sync(transmission[seq_start:])
@ -253,20 +251,22 @@ class SSTVDecoder(object):
if chan % 2 == 1: if chan % 2 == 1:
pixel_time = self.mode.MERGE_PIXEL_TIME pixel_time = self.mode.MERGE_PIXEL_TIME
pixel_window = round(pixel_time * window_factor * self._sample_rate) pixel_window = round(pixel_time * window_factor *
self._sample_rate)
centre_window_time = (pixel_time * window_factor) / 2 centre_window_time = (pixel_time * window_factor) / 2
for px in range(self.mode.LINE_WIDTH): for px in range(self.mode.LINE_WIDTH):
chan_offset = self.mode.CHAN_OFFSETS[chan] chan_offset = self.mode.CHAN_OFFSETS[chan]
px_sample = round(seq_start + (chan_offset + px * pixel_time \ px_sample = round(seq_start + (chan_offset + px *
- centre_window_time) * self._sample_rate) pixel_time - centre_window_time) *
freq = self._peak_fft_freq(transmission[px_sample:px_sample+pixel_window]) self._sample_rate)
section = transmission[px_sample:px_sample+pixel_window]
freq = self._peak_fft_freq(section)
image_data[line][chan].append(SSTVDecoder._calc_lum(freq)) image_data[line][chan].append(SSTVDecoder._calc_lum(freq))
progress_bar(line, self.mode.LINE_COUNT - 1, progress_bar(line, self.mode.LINE_COUNT - 1,
"Decoding image... ", self.log_basic) "Decoding image... ", self.log_basic)
@ -281,7 +281,8 @@ class SSTVDecoder(object):
else: else:
col_mode = "RGB" col_mode = "RGB"
image = Image.new(col_mode, (self.mode.LINE_WIDTH, self.mode.LINE_COUNT)) image = Image.new(col_mode,
(self.mode.LINE_WIDTH, self.mode.LINE_COUNT))
pixel_data = image.load() pixel_data = image.load()
for y in range(self.mode.LINE_COUNT): for y in range(self.mode.LINE_COUNT):
@ -290,12 +291,17 @@ class SSTVDecoder(object):
for x in range(self.mode.LINE_WIDTH): for x in range(self.mode.LINE_WIDTH):
if self.mode.COLOR == spec.COL_FMT.GBR: if self.mode.COLOR == spec.COL_FMT.GBR:
pixel = image_data[y][2][x], image_data[y][0][x], image_data[y][1][x] pixel = (image_data[y][2][x],
image_data[y][0][x],
image_data[y][1][x])
elif self.mode.COLOR == spec.COL_FMT.YUV: elif self.mode.COLOR == spec.COL_FMT.YUV:
pixel = (image_data[y][0][x], image_data[y-ryby][1][x], pixel = (image_data[y][0][x],
image_data[y-(ryby-1)][1][x]) image_data[y-ryby][1][x],
image_data[y-(ryby-1)][1][x])
else: else:
pixel = image_data[y][0][x], image_data[y][1][x], image_data[y][2][x] pixel = (image_data[y][0][x],
image_data[y][1][x],
image_data[y][2][x])
pixel_data[x, y] = pixel pixel_data[x, y] = pixel
if self.mode.COLOR == spec.COL_FMT.YUV: if self.mode.COLOR == spec.COL_FMT.YUV:

5
sstv/spec.py
View File

@ -1,4 +1,5 @@
#!usr/bin/env python #!usr/bin/env python
from enum import Enum from enum import Enum
BREAK_OFFSET = 0.300 BREAK_OFFSET = 0.300
@ -10,17 +11,20 @@ HDR_WINDOW_SIZE = 0.010
VIS_BIT_SIZE = 0.030 VIS_BIT_SIZE = 0.030
class COL_FMT(Enum): class COL_FMT(Enum):
RGB = 1 RGB = 1
GBR = 2 GBR = 2
YUV = 3 YUV = 3
BW = 4 BW = 4
class SSTV(object): class SSTV(object):
HAS_START_SYNC = False HAS_START_SYNC = False
HAS_MERGE_SCAN = False HAS_MERGE_SCAN = False
CHAN_SYNC = 0 CHAN_SYNC = 0
class M1(SSTV): class M1(SSTV):
NAME = "Martin 1" NAME = "Martin 1"
VIS_CODE = 44 VIS_CODE = 44
@ -35,7 +39,6 @@ class M1(SSTV):
CHAN_COUNT = 3 CHAN_COUNT = 3
CHAN_TIME = SEP_PULSE + SCAN_TIME 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 = [SYNC_PULSE + SYNC_PORCH]
CHAN_OFFSETS.append(CHAN_OFFSETS[0] + CHAN_TIME) CHAN_OFFSETS.append(CHAN_OFFSETS[0] + CHAN_TIME)
CHAN_OFFSETS.append(CHAN_OFFSETS[1] + CHAN_TIME) CHAN_OFFSETS.append(CHAN_OFFSETS[1] + CHAN_TIME)