relikd/LiberPrayground
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

761

Browse Source
This commit is contained in:
relikd
2021-03-04 22:07:33 +01:00
parent 9f466cc02c
commit dbc709da07
4 changed files with 744 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
other/761/mp3.py Executable file
View File

@@ -0,0 +1,362 @@
#!/usr/bin/env python3
import os
import sys
if len(sys.argv) > 1 and os.path.isfile(sys.argv[1]):
INPUT_FILE = sys.argv[1]
else:
INPUT_FILE = 'audio_files/761.MP3' # '761.MP3' 'index.mp3'
# print('File not found.')
# exit()
class MP3Header(object):
# https://id3.org/mp3Frame
# http://mpgedit.org/mpgedit/mpeg_format/MP3Format.html
# http://www.mpgedit.org/mpgedit/mpeg_format/mpeghdr.htm
# Starts with 11x 1-byte
SYNC = '' # 11 bit
ID = { # 2 bit (but reduced to 1 later)
0b00: 'MPEG Version 2.5',
0b01: 'reserved',
0b10: 'MPEG Version 2 (ISO/IEC 13818-3)',
0b11: 'MPEG Version 1 (ISO/IEC 11172-3)'}
LAYER = { # 2 bit (but -1 later)
0b00: 'reserved',
0b01: 'Layer III',
0b10: 'Layer II',
0b11: 'Layer I'}
PROTECTION = { # 1 bit
0: 'Protected', # 16bit CRC after header
1: 'Not Protected'}
BITRATE = { # 4 bit
# MPEG2-Layer3, M2-L2, M2-L1, M1-L3, M1-L2, M1-L1 in kbit/s
0b0000: [0, 0, 0, 0, 0, 0], # free
0b0001: [8000, 32000, 32000, 32000, 32000, 32000],
0b0010: [16000, 48000, 64000, 40000, 48000, 64000],
0b0011: [24000, 56000, 96000, 48000, 56000, 96000],
0b0100: [32000, 64000, 128000, 56000, 64000, 128000],
0b0101: [64000, 80000, 160000, 64000, 80000, 160000],
0b0110: [80000, 96000, 192000, 80000, 96000, 192000],
0b0111: [56000, 112000, 224000, 96000, 112000, 224000],
0b1000: [64000, 128000, 256000, 112000, 128000, 256000],
0b1001: [128000, 160000, 288000, 128000, 160000, 288000],
0b1010: [160000, 192000, 320000, 160000, 192000, 320000],
0b1011: [112000, 224000, 352000, 192000, 224000, 352000],
0b1100: [128000, 256000, 384000, 224000, 256000, 384000],
0b1101: [256000, 320000, 416000, 256000, 320000, 416000],
0b1110: [320000, 384000, 448000, 320000, 384000, 448000],
0b1111: [0, 0, 0, 0, 0, 0]} # bad
FREQUENCY = { # 2 bit in Hz
# MPEG-2, MPEG-1, MPEG-2.5 (not used)
0b00: [22050, 44100, 11025],
0b01: [24000, 48000, 12000],
0b10: [16000, 32000, 8000],
0b11: [0, 0, 0]} # reserved
PADDING = { # 1 bit
0: 'Padded', # +1 byte to frame length
1: 'Not Padded'}
PRIVATE = { # 1 bit
0: 'free', # freely used for whatever
1: 'free'}
MODE = { # 2 bit
0b00: 'Stereo',
0b01: 'Joint stereo (Stereo)',
0b10: 'Dual channel (2 mono channels)',
0b11: 'Single channel (Mono)'}
MODE_EXTENSION = { # 2 bit
# Layer I & II Layer III
# Intensity stereo MS stereo
# 0b00 bands 4 to 31 off off
# 0b01 bands 8 to 31 on off
# 0b10 bands 12 to 31 off on
# 0b11 bands 16 to 31 on on
}
COPYRIGHT = { # 1 bit
0: 'Not Copyrighted',
1: 'Copyrighted'}
ORIGINAL = { # 1 bit
0: 'Copy of Original',
1: 'Original'}
EMPHASIS = { # 2 bit
0b00: 'none',
0b01: '50/15 ms',
0b10: 'reserved',
0b11: 'CCIT J.17'}
MULTIPLY = [144, 144, 12] # frame length multiplier
# FRAMESIZE = [0, 1152, 1152, 384] # in samples
# SLOTS = [0, 1, 1, 4] # in bytes
def init_from_bytes(self, b0, b1, b2, b3):
self.emphasis = b3 & 0b11
b3 >>= 2
self.original = b3 & 1
b3 >>= 1
self.copyright = b3 & 1
b3 >>= 1
self.mode_extension = b3 & 0b11
b3 >>= 2
self.mode = b3 & 0b11
self.private = b2 & 1
b2 >>= 1
self.pad = b2 & 1
b2 >>= 1
self.frequency = b2 & 0b11
if self.frequency == 3:
raise ValueError('Reserved sample rate')
b2 >>= 2
self.bitrate = b2 & 0b1111
if self.frequency == 0b1111:
raise ValueError('Invalid bitrate')
self.protection = b1 & 1
b1 >>= 1
self.layer = b1 & 0b11 # Layer I-III
if self.layer == 0:
raise ValueError('Reserved MPEG-Layer')
b1 >>= 2
self.id = b1 & 0b11
b1 >>= 2
self.sync = (b0 << 3) + (b1 & 0b111)
if self.sync != 0b11111111111:
raise ValueError('Not a MP3 header')
def __init__(self, b0, b1, b2, b3):
self.init_from_bytes(b0, b1, b2, b3)
i_lyr = self.layer - 1 # because arrays
i_id = self.id & 1 # because arrays
br = self.BITRATE[self.bitrate][i_lyr + i_id * 3]
sr = self.FREQUENCY[self.frequency][i_id]
self.framelength = self.MULTIPLY[i_lyr] * br / sr
if self.pad:
self.framelength += 1
if i_lyr == 2: # LAYER-1
self.framelength *= 4
# TODO: check whether CRC length must be added
# if self.protection == 0:
self.framelength = int(self.framelength)
def as_bytes(self):
b = self.sync
b = b << 2 | self.id
b = b << 2 | self.layer
b = b << 1 | self.protection
b = b << 4 | self.bitrate
b = b << 2 | self.frequency
b = b << 1 | self.pad
b = b << 1 | self.private
b = b << 2 | self.mode
b = b << 2 | self.mode_extension
b = b << 1 | self.copyright
b = b << 1 | self.original
b = b << 2 | self.emphasis
return b.to_bytes(4, 'big')
def __str__(self):
f = '{:011b} {:02b} {:02b} {:b} {:04b} {:02b} {:b} {:b} {:02b} {:02b} {:b} {:b} {:02b}'
return f.format(
self.sync, self.id, self.layer, self.protection, self.bitrate,
self.frequency, self.pad, self.private, self.mode,
self.mode_extension, self.copyright, self.original, self.emphasis)
def bin_to_hex(binary_str):
ret = ''
for i in range(0, len(binary_str), 8):
ret += '{:02X}'.format(int(binary_str[i:i + 8], 2))
return ret
def bin_to_text(binary_str):
ret = ''
for i in range(0, len(binary_str), 8):
ret += chr(int(binary_str[i:i + 8], 2))
return ret
def flip_bits(bits):
return bits.replace('1', '_').replace('0', '1').replace('_', '0')
# def read_mp3_headers(bytes, to_file):
# with open(to_file, 'w') as fo:
# counter = 0
# offset = 0
# for byte in bytes:
# if offset < 6000: # skip ID3
# offset += 8
# continue
# for x in [128, 64, 32, 16, 8, 4, 2, 1]:
# offset += 1
# z = 1 if byte & x else 0
# if z:
# counter += 1
# else:
# if counter >= 13:
# fo.write('{}\n'.format(offset))
# counter = 0
# def prepare_mp3_headers(bytes, header_file):
# with open(header_file, 'r') as f:
# indices = [int(x) for x in f.readlines()]
# all_of_them = []
# for i in indices[:10]:
# i -= 14 # beginning of header, 13 + 1 for prev bit
# major = i // 8
# minor = i % 8
# raw_int = 0
# for u in range(5):
# raw_int += bytes[major + u] << (32 - u * 8)
# bit_str = ''
# for x in range(7 - minor + 32, 7 - minor, -1):
# bit_str += '1' if raw_int & (1 << x) else '0'
# try:
# all_of_them.append((i, MP3Header(bit_str)))
# except ValueError:
# pass
# return all_of_them
# def analyze_mp3_headers(bytes, prepared_obj):
# txt = ''
# for i, head in prepared_obj:
# print('{:06d} {} {}'.format(i, head, head.framelength))
# # if head == '00':
# # txt += head[7]
# print(txt)
# print(bin_to_text(txt))
# read_mp3_headers(bytes, to_file='mp3_header_indices.txt')
# anlz = prepare_mp3_headers(bytes, header_file='mp3_header_indices.txt')
# analyze_mp3_headers(bytes, anlz)
def parse_mp3_header(bytes):
for i, x in enumerate(bytes):
if x != 0xFF:
continue
if bytes[i + 1] >> 5 == 0b111:
try:
obj = MP3Header(*bytes[i:i + 4])
next_at = i + obj.framelength
except ValueError:
continue
try:
MP3Header(*bytes[next_at:next_at + 4])
return next_at, obj
except ValueError:
continue
def enum_mp3_header(bytes):
i, header = parse_mp3_header(bytes)
while header and i < len(bytes):
header = MP3Header(*bytes[i:i + 4])
yield i, header
i += header.framelength
with open(INPUT_FILE, 'rb') as f:
bytes = f.read()
uniq = [set(), set(), set(), set(), set(),
set(), set(), set(), set(), set(), set()]
keyz = ['id', 'layer', 'protection', 'frequency', 'pad', 'private',
'mode_extension', 'copyright', 'original', 'emphasis', 'framelength']
txt_chr = ''
txt_bit = ''
count_header = 0
# # Modify existing new file (a copy)
# last_i = 0
# with open(INPUT_FILE + '.modified.mp3', 'wb') as f:
# for i, header in enum_mp3_header(bytes):
# f.write(bytes[last_i:i])
# header.mode_extension = 3
# f.write(header.as_bytes())
# last_i = i + 4
# # Split in chunks
# if not os.path.isdir('tmp'):
# os.mkdir('tmp')
# if not os.path.isdir('tmp/mp3_frames'):
# os.mkdir('tmp/mp3_frames')
# last_i = 0
# running_i = 0
# for i, header in enum_mp3_header(bytes):
# with open('tmp/mp3_frames/{:06d}.mp3'.format(running_i), 'wb') as f:
# running_i += 1
# f.write(bytes[last_i:i])
# last_i = i
# exit()
txt = [''] * 624
# Parse and analyze header info
for i, header in enum_mp3_header(bytes):
# for x in range(1, 624):
# txt[x] += '1' if bytes[i - x] & 7 else '0'
# print(header)
count_header += 1
txt_chr += chr(bytes[i - 1])
txt_bit += '1' if bytes[i - 1] & 1 else '0'
for i, k in enumerate(keyz):
uniq[i].add(getattr(header, k))
for x in range(624):
if txt[x]:
print(bin_to_text(txt[x]))
# exit()
print('The unique values per header field:')
print({x: y for x, y in zip(keyz, uniq)})
print()
def print_bits(bits):
print('\nBinary:')
print(bits)
print('\nText (normal):')
print(bin_to_text(bits))
print('\nText (reverse):')
print(bin_to_text(bits[::-1]))
print('\nText (inverse):')
print(bin_to_text(flip_bits(bits)))
print('\nText (reverse, inverse):')
print(bin_to_text(flip_bits(bits[::-1])))
print()
print('Last byte per chunk:')
print(txt_chr)
print()
print('Last bit per chunk:')
print_bits(txt_bit)
# find header fields that differ
for i in range(len(uniq) - 1, -1, -1):
if len(uniq[i]) == 1:
del uniq[i]
del keyz[i]
else:
uniq[i] = uniq[i].pop() # good luck if there are three
if not uniq:
print('Nothing to do. No header changes value')
else:
txt = [''] * len(uniq)
# skip_once = True
for i, header in enum_mp3_header(bytes):
# if skip_once:
# skip_once = False
# continue
for i, k in enumerate(keyz):
txt[i] += '1' if getattr(header, k) == uniq[i] else '0'
for i, k in enumerate(keyz):
print('Header field:', k)
print_bits(txt[i])
print()
print('Number of headers: {}'.format(count_header))
print()

38
other/761/mpg123.py Normal file
View File

@@ -0,0 +1,38 @@
from mpg123 import Mpg123
# import wave
# import struct
# https://github.com/20tab/mpg123-python
def bin_to_text(binary_str):
ret = ''
for i in range(0, len(binary_str), 8):
ret += chr(int(binary_str[i:i + 8], 2))
return ret
mp3 = Mpg123('761.MP3')
# rate, channels, encoding = mp3.get_format()
# wav = wave.open('761.wav', 'wb')
# wav.setnchannels(channels)
# wav.setframerate(rate)
# wav.setsampwidth(mp3.get_width_by_encoding(encoding))
# # fill the wave file
# for frame in mp3.iter_frames():
# wav.writeframes(frame)
# wav.close()
txt = [''] * 8
for i, frame in enumerate(mp3.iter_frames()):
# bytes = struct.unpack('H' * (len(frame) // 2), frame)
for u in range(1):
for b in range(0, len(frame), 167):
txt[u] += '1' if frame[b] & (1 << 0) else '0'
# for x in bytes:
# txt += '1' if x & 1 else '0'
# if i > 5:
# break
for t in txt:
print(bin_to_text(t))

BIN
other/761/peaks.txt Normal file
View File

Binary file not shown.

344
other/761/wav.py Executable file
View File

@@ -0,0 +1,344 @@
#!/usr/bin/env python3
import os
import wave
import struct
TRACK_LEN = 146.468
SAMPLE_LEN = 6459264 # hard coded so we dont need to load the file
SAMPLING = 8 # take every x-th frame
SMOOTH_WINDOW = 2 # gaussian window size X +- window
END_TIMES = [
10.639, 15.914, 28.937, 32.239, 33.590, 38.875, 42.408,
45.919, 49.475, 54.763, 72.301, 74.172, 81.219, 82.339,
92.900, 99.753, 100.654, 105.919, 114.771, 146.468]
def flip_bits(bits):
return bits.replace('1', '_').replace('0', '1').replace('_', '0')
def bin_to_hex(binary_str):
ret = ''
for i in range(0, len(binary_str), 8):
ret += '{:02X}'.format(int(binary_str[i:i + 8], 2))
return ret
def bin_to_text(binary_str):
ret = ''
for i in range(0, len(binary_str), 8):
ret += chr(int(binary_str[i:i + 8], 2))
return ret
def oneChannel(fname, chanIdx, maxread=None):
f = wave.open(fname, 'rb')
c_chn = f.getnchannels()
c_frm = f.getnframes()
if maxread:
c_frm = min(maxread, c_frm)
assert f.getsampwidth() == 2
s = f.readframes(c_frm)
f.close()
unpstr = '<{0}h'.format(c_frm * c_chn)
x = list(struct.unpack(unpstr, s))
return x[chanIdx::c_chn]
def find_db_peaks(wav_filename, threshold, write_to=None):
res = oneChannel(wav_filename, 1) # 100000
if len(res) != SAMPLE_LEN:
print('WARN: file sample rate mismatch with SAMPLE_LEN')
with open(write_to, 'wb') as fo:
# apply a rough gaussian smoothing
ftlr_rng = range(-SMOOTH_WINDOW, SMOOTH_WINDOW + 1)
for i in range(SMOOTH_WINDOW, len(res) - SMOOTH_WINDOW, SAMPLING):
z = [res[i + x] * (1 / (abs(x) + 1)) for x in ftlr_rng]
z = sum(z)
f = abs(z) > 400 # threshold
fo.write(b'\xFF' if f else b'\x00')
def fill_gaps(fname, window_size, min_count, threshold=128, write_to=None):
window = [0] * window_size
with open(write_to, 'wb') as fo:
with open(fname, 'rb') as fi:
for x in fi.read():
window.pop(0)
window.append(1 if x > threshold else 0)
f = sum(window) > min_count
fo.write(b'\xFF' if f else b'\x00')
def find_db_change(fname, threshold=128, write_to=None):
res = [(0, False)]
prev = False
with open(fname, 'rb') as fi:
for i, x in enumerate(fi.read()):
f = x > threshold
if f != prev:
prev = f
res.append((i, f))
with open(write_to, 'w') as fo:
for x in res:
fo.write('{}: {}\n'.format(*x))
# fo.write('\n'.join(['{}: {}'.format(*x) for x in res]))
def find_signal_midpoints(fname):
res = [] # (pos, width, dist_to_prev)
prev = 0
with open(fname, 'r') as fi:
lines = fi.readlines()
for x, y in zip(lines[1::2], lines[2::2]):
x = int(x.split(':')[0])
y = int(y.split(':')[0])
w = y - x
x += int(w / 2) # center point
res.append((x, w, x - prev))
prev = x
return res
def analyze_midpoints(midpoints_list, min_frames):
res = [] # (frame-no, time, dist-to-prev, type 'S-M-E')
typ = 'E' # marks first as [S]tart
for x, width, dist in midpoints_list:
if width < min_frames:
continue
typ = 'S' if typ == 'E' else 'M'
x *= SAMPLING
at_time = x / SAMPLE_LEN * TRACK_LEN
dist *= SAMPLING
for i, end_time in enumerate(END_TIMES):
if abs(end_time - at_time) < 0.100: # accurate within 100 ms
typ = 'E'
del END_TIMES[i] # keeps count if all are used up
break
res.append((x, at_time, dist, typ))
if len(END_TIMES) > 0:
if END_TIMES[0] > res[-1][1]:
for x in END_TIMES:
fn = round(x / TRACK_LEN * SAMPLE_LEN)
res.append((fn, x, fn - res[-1][0], 'E'))
else:
print('These endpoints were not found:')
print(END_TIMES) # double check
return res
def find_common_frame_dist(arr):
arr = [x[2] for x in arr if x[3] != 'S']
min_dist = min(arr)
print('Smallest common divisor: {}'.format(min_dist))
best_match = min_dist
best_sum = 999999
for tx in range(min_dist - 200, min_dist + 200 + 1):
subsum = 0
for x in arr:
x /= tx
x -= round(x)
subsum += x * x # least square distance
if subsum < best_sum:
best_sum = subsum
best_match = tx
print('Best matching frame dist: {}'.format(best_match))
return best_match
def analyze_db_peaks(wav_file, force=False):
print('761')
print('===')
print('Track length:', TRACK_LEN)
print('Total frames:', SAMPLE_LEN)
if not os.path.isdir('tmp'):
os.mkdir('tmp')
tmp1 = 'tmp/wav-peak-analysis_1.dat'
tmp2 = 'tmp/wav-peak-analysis_2.dat'
tmp3 = 'tmp/wav-peak-analysis_3.txt'
if force or not os.path.isfile(tmp1):
find_db_peaks(wav_file, 400, write_to=tmp1)
if force or not os.path.isfile(tmp2):
fill_gaps(tmp1, window_size=80, min_count=20, write_to=tmp2)
# force = True
if force or not os.path.isfile(tmp3):
find_db_change(tmp2, write_to=tmp3)
points = find_signal_midpoints(tmp3)
points = analyze_midpoints(points, min_frames=10)
freq = find_common_frame_dist(points)
# if times between 96.68-96.79 and 70.10-70.21 are sampled differently
# freq /= 2 # use *2 or /2 to decrease or increase sampling frequency
print('''
The columns are as follows:
Type Time(s) Time(frame) dist-to-prev
- Type is one of [S]tart point, [M]id-point, or [E]nd point
- dist-to-prev is frame distance to previous signal divided by frame-dist
''')
bits = ['']
nums = [[]]
t_between = []
t_lengths = []
since_start = 0
for x, at, dist, typ in points:
def time_diff_tpl(diff):
return (round(diff / freq), diff / SAMPLE_LEN * TRACK_LEN)
in_samples = round(dist / freq)
print('{} {:.2f} {} {}'.format(typ, at, x, in_samples))
if typ == 'S':
# bits[-1] += '0' * (in_samples - 1) # consider space between
bits[-1] += '1'
t_between.append(time_diff_tpl(dist))
since_start = 0
elif typ == 'E':
bits[-1] += '0' * (in_samples - 1)
bits[-1] += '0' # or 1?
missing_bits = 8 - len(bits[-1]) % 8
if missing_bits != 8:
# bits[-1] = '0' * missing_bits + bits[-1]
bits[-1] += '0' * missing_bits
since_start += dist
t_lengths.append(time_diff_tpl(since_start))
bits.append('')
nums[-1].append(in_samples)
nums.append([])
else:
since_start += dist
bits[-1] += '0' * (in_samples - 1)
bits[-1] += '1'
nums[-1].append(in_samples)
if bits[-1] == '':
del bits[-1]
if not nums[-1]:
del nums[-1]
print()
print('Distance between transmissions:')
print(', '.join(['{} ({:.2f}s)'.format(x, y) for x, y in t_between]))
print()
print('Lengths of transmission:')
print(', '.join(['{} ({:.2f}s)'.format(x, y) for x, y in t_lengths]))
print()
print('Individual signals:')
for i, x in enumerate(nums):
print(' {:2}: {}'.format(i, x))
print()
print('Individual signals (total time):')
for i, x in enumerate(nums):
r = [0]
for n in x:
r.append(r[-1] + n)
print(' {:2}: {}'.format(i, r[1:]))
print()
print('''
The following assumes that each transmission:
- begins with a 1 bit
- end is always a 0 bit
- midpoints are '0' * (dist-to-prev - 1) + '1'
- no counting in-between transmissions
Here is a representation of the individual transmissions,
as well as the full string at the end. Results are:
(0): signals are 1 bit, read left-to-right
(1): reverse bit order (aka. read right-to-left)
(2): as (0) but with inverted bits
(3): reversed and inverted
Interpreting individual transmissions:
''')
def print_arr_w_alternates(bits, fn):
print('0\n{}'.format([fn(x) for x in bits]))
print('1\n{}'.format([fn(x[::-1]) for x in bits]))
print('2\n{}'.format([fn(flip_bits(x)) for x in bits]))
print('3\n{}'.format([fn(flip_bits(x)[::-1]) for x in bits]))
print()
def print_str_w_alternates(bits, fn):
not_bits = flip_bits(bits)
print('0: {}'.format(fn(bits)))
print('1: {}'.format(fn(bits[::-1])))
print('2: {}'.format(fn(not_bits)))
print('3: {}'.format(fn(not_bits[::-1])))
print()
# print('As numbers:')
# print_arr_w_alternates(bits, lambda x: int(x, 2))
# print('As binary:')
# print_arr_w_alternates(bits, lambda x: x)
# print('As hex:')
# print_arr_w_alternates(bits, lambda x: bin_to_hex(x))
print('As text:')
print_arr_w_alternates(bits, lambda x: bin_to_text(x))
print('Interpreting as a whole:')
print()
concat = ''.join([x for x in bits])
# print('As numbers:')
# print_str_w_alternates(concat, lambda x: int(x, 2))
# print('As binary:')
# print_str_w_alternates(concat, lambda x: x)
# print('As hex:')
# print_str_w_alternates(concat, lambda x: bin_to_hex(x))
print('As text:')
print_str_w_alternates(concat, lambda x: bin_to_text(x))
# Least Significant Bit Analysis
# https://medium.com/analytics-vidhya/get-secret-message-from-audio-file-8769421205c3
def analyze_lsb(wav_filename):
obj = wave.open(wav_filename, 'rb')
# print(obj.getparams())
fcount = obj.getnframes()
fcount = 1000
bytes = bytearray(list(obj.readframes(fcount)))
obj.close()
bytes = struct.unpack('H' * (len(bytes) // 2), bytes)
# if not os.path.isdir('tmp'):
# os.mkdir('tmp')
# Every frame LSB
for z in range(1, 2):
for u in range(z):
txt = ''
for i in range(u, len(bytes), z):
f = bytes[i] & (1 << 0)
txt += '1' if f else '0'
# txt += chr(bytes[i])
# print(txt)
print(bin_to_text(txt))
# Alternating frame LSB
# left = bytes[::2]
# right = bytes[1::2]
# for z in range(1, 2):
# for u in range(z):
# txt = ''
# for i in range(u, len(left), 2):
# if i % 2 == 0:
# txt += str(left[i] & 1)
# # txt += chr(left[i])
# else:
# txt += str(right[i] & 1)
# # txt += chr(right[i])
# # print(txt)
# print(bin_to_text(txt))
analyze_db_peaks('audio_files/761_convergePitch_2.wav', force=False)
# analyze_lsb('audio_files/761.wav')