今度は成功の予感
こんにちは、こんばんわ。かえるのクーです。
 |
注）もちろんボク ではありません。 モデルさんです。 |
ウェーブレットの重み基底の符号化器が「失敗」とおもいきや、なかなかの再現性能をみせたので、気を良くして長い音（曲）とかでも汎用的に実験できるよう、改めて設計してコーディングしてみます。（ウェーブレット専用）
「生きているから 明日があるから 地球がまわっているから」♪
プログラムデザイン
「重み」は学習しないのでこれはもう「自己符号化器」ではありません。ただの符号化器です。設計していきます。
| お題 | 内容 |
|---|---|
| プログラム名： | コレ（自己）ジャナイ符号化器 |
| コンセプト１： | 重み基底に各種周波数、位相のウェーブレットを予め設定しておき、このウェーブレット重みに対して符号化を行う。 |
| コンセプト２： | 符号化されたものを、これも同様のウェーブレット重みで復号化する。 |
| コンセプト３： | 重みを学習しない以外は自己符号化器の構造をそのまま使用する。 |
| コンセプト4： | 離散ウェーブレット変換と同等の符号化、復号化を行うため、ウェーブレット重みを所定の位置にシフトして演算できるようにする。 |
重み基底に使うウェブレットの設計
W1、Ｗ２ともに準備する理想的なウェーブレットを使用します。演算幅が違えば「パワー強度」も違ってくるので、準備する重みのほうで予め補正調整しておきます。
（ウェーブレットの幅が広ければ演算した結果が大きくなります。
（追記：修正）上「波形間の直積」と記載した部分は「波形間の内積」の誤りです。お詫びして訂正いたします。（追記ここまで）
| お題 | 内容 |
|---|---|
| 周波数バリエーション： | ピアノの最低音Ａからはじめ、10オクターブの12半音(平均率）を準備する。 |
| 位相バリエーション： | それぞれの周波数について、位相360°（２π）を16分割して準備する。 |
| エンベロープ： | ウェーブレットのエンベロープはハン窓（ハニング窓)を使用する。 |
| サンプリング： | 48KHzを基準とする。 |
| ウェーブレットの幅： | 2のべき乗幅とし、オクターブ毎に幅をかえる。また、各窓は16周期以上のsin波が収まるようにする。（低周波ほど窓幅が広い） |
| 重みの強度（振幅）： | 最終的な演算後の各周波数レンジのパワースペクトルが同じになるように振幅を調整する。２段の（リニア）乗算演算を行うため、ウェーブレット幅の平方根の逆数に比例するように設定する。 |
| 重みの数： | 基底重みの数は12（クロマ）×10（オクターブ）×16(フェーズ）=1920（重み個数）となる。 |
入力前処理の設計
44.1KHz ２チャンネルの入力でもそれ以外でもすべて内部処理のレートに合わせます。
| お題 | 内容 |
|---|---|
| サンプリングレート変換： | 内部で48KHzで演算するので、入力のサンプリングレートを48KHzに変換する。 |
| チャンネル数： | 内部でモノラルで演算するので、チャンネルを１にする（左右ミキシング） |
エンコードの設計
ランプ関数を使うところがミソでした。他のを使うと歪んだりノイズがでます。
前回の実験で失敗していたのは、「ハイパータンジェント」を使っていたことも原因でした。（マイナス値となる演算結果がノイズとしてはいってくる）
| お題 | 内容 |
|---|---|
| 重み枠シフト（スウィープ）幅： | ウェーブレット重みの1/4幅とする。 |
| 使用する重み基底： | 上記のウェーブレットのセットを用いる。 |
| 活性化関数： | ランプ関数（ReLU）を用いる。 |
| バイアス： | 活性化関数にバイアスは用いない。 |
| 内部演算： | 32Bitの浮動小数点とする。 |
デコードの設計
最終段リニアにしたほうがいいのは、身をもって体験しました。
| お題 | 内容 |
|---|---|
| 重み枠シフト（スウィープ）幅： | エンコードと同じとする。 |
| 使用する重み基底： | エンコードのウェーブレットのセットの転置を用いる（重み共有）。 |
| 活性化関数： | 恒等関数（Liner）を用いる。 |
| バイアス： | 活性化関数にバイアスは用いない。 |
| 内部演算： | 32Bitの浮動小数点とする。 |
波形の最終出力の設計
内部演算した結果を、ほぼそのまま出します。
| お題 | 内容 |
|---|---|
| サンプリングレート： | 一貫して内部48KHzで演算しているので、そのまま48KHzで出力する。 |
| Bit幅： | 16Bitに変換して出力する。 |
PG効率化の工夫
指定した音の前後にいちばん大きなウィンドウ幅の1/2サイズの無音をつけて処理し、復元したら、その無音部分を削除して出力する。そうすることにより、境界の考慮が不要になる部分があり、簡略化できる。（ただし、特に高音部分の配列で無音を担当する部分が増えるので、「若干」メモリ的に非効率になる）ループの外側であれば、多少速度や資源がエコでなくても、簡単にコーディングでき、シンプルなほうが修正もしやすくて幸せになれると思います。
予想
かなり視聴にたえるくらいのスペックにしたつもりなので、原音がかなりきれいに再現できると思います。
プログラム
実験用のコードをいれており、コアな部分以外でとても長くなっています。GitHubに置く予定はいまのところありません。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import numpy as np
from scipy import signal
import audioop
import wave
"""
-------------------------------------------------
Korejanai Encoder V1.0 2017.09.22 Koo Wells
-------------------------------------------------
"""
IN_WAVE_FILE = ".\\sample20170920.wav"
DECODE_FILE = ".\\DecodeSound.wav"
BASE_SAMPLING = 48000 # 内部処理の基本サンプリングレート
BASE_WINDOW_SIZE = 2**15 # いちばん広いウェーブレット重みの幅
HALF_WINDOW_SIZE = int(BASE_WINDOW_SIZE / 2) # BASE_WINDOW_SIZE幅の半分
FASE_SPLIT = 16 # 位相は１６分割(22.5°)
A0_NOTE_NO = 21 # A0(27.5Hz) Midi Note No. ピアノの一番低い音
A4_NOTE_NO = 69 # A4(440Hz) Midi Note No.
CROMATIC = 12 # 平均律12半音
WAVELET_SHIFT = 4 # 1/指定数 分づつシフト
# サウンドの読み込みと変換をします。
class soundtool():
def __init__(self):
pass
# 音のファイルを読み、基本的な情報を獲得します。
def fetch_soundData(self, filename):
wave_read = wave.open(filename, 'r')
w_channel = wave_read.getnchannels()
w_rate = wave_read.getframerate()
w_framenumber = wave_read.getnframes()
w_frame = wave_read.readframes(w_framenumber)
wave_read.close()
return w_channel, w_rate, w_framenumber, w_frame
# ストリームをサンプリングレート48KHz モノラルの音の配列（16bit)に変換します。
def conv48KMonoArray(self,aframe, achannel, arate):
if achannel == 2:
converted = audioop.tomono(aframe, 2, 0.5, 0.5)
else:
converted = aframe
new_frames = audioop.ratecv(converted, 2, 1, arate, BASE_SAMPLING, None)
array = np.array(np.frombuffer(new_frames[0], dtype="int16"))
return array
# ストリームをWaveファイルとして出力します。
def write_soundData(self, filename, stream):
s_write = wave.open(filename, 'w')
s_write.setparams((1, 2, BASE_SAMPLING, 0, 'NONE', 'Uncompressed'))
s_write.writeframes(stream)
s_write.close()
# ボクが欲しいのはコレジャナイ エンコーダ
class korejanai():
k_activation_Sigmoid = 'Sigmoid'
k_activation_ReLU = 'ReLU'
k_activation_Tanh = 'Tanh'
k_activation_Liner = 'Liner'
def __init__(self):
# 活性化関数とウェーブレットの設定
self.func = self.factory_activate_func(self.k_activation_ReLU)
self.func2 = self.factory_activate_func(self.k_activation_Liner)
# 21:MidiNote A0 27.5Hz
self.w1 = [
self.wavelet_init(2**(0/2), A0_NOTE_NO, A0_NOTE_NO + CROMATIC*1 - 1, FASE_SPLIT, 2**15),
self.wavelet_init(2**(1/2), A0_NOTE_NO + CROMATIC*1, A0_NOTE_NO + CROMATIC*2 - 1, FASE_SPLIT, 2**14),
self.wavelet_init(2**(2/2), A0_NOTE_NO + CROMATIC*2, A0_NOTE_NO + CROMATIC*3 - 1, FASE_SPLIT, 2**13),
self.wavelet_init(2**(3/2), A0_NOTE_NO + CROMATIC*3, A0_NOTE_NO + CROMATIC*4 - 1, FASE_SPLIT, 2**12),
self.wavelet_init(2**(4/2), A0_NOTE_NO + CROMATIC*4, A0_NOTE_NO + CROMATIC*5 - 1, FASE_SPLIT, 2**11),
self.wavelet_init(2**(5/2), A0_NOTE_NO + CROMATIC*5, A0_NOTE_NO + CROMATIC*6 - 1, FASE_SPLIT, 2**10),
self.wavelet_init(2**(6/2), A0_NOTE_NO + CROMATIC*6, A0_NOTE_NO + CROMATIC*7 - 1, FASE_SPLIT, 2**9),
self.wavelet_init(2**(7/2), A0_NOTE_NO + CROMATIC*7, A0_NOTE_NO + CROMATIC*8 - 1, FASE_SPLIT, 2**8),
self.wavelet_init(2**(8/2), A0_NOTE_NO + CROMATIC*8, A0_NOTE_NO + CROMATIC*9 - 1, FASE_SPLIT, 2**7),
self.wavelet_init(2**(9/2), A0_NOTE_NO + CROMATIC*9, A0_NOTE_NO + CROMATIC*10 - 1, FASE_SPLIT, 2**6),
]
self.w2 = [
self.w1[0].T,
self.w1[1].T,
self.w1[2].T,
self.w1[3].T,
self.w1[4].T,
self.w1[5].T,
self.w1[6].T,
self.w1[7].T,
self.w1[8].T,
self.w1[9].T,
]
self.z = [
[],
[],
[],
[],
[],
[],
[],
[],
[],
[],
]
# 活性化関数のファクトリ
def factory_activate_func(self, activate_func):
rtn_func = ActivateFunction()
if activate_func == self.k_activation_Sigmoid:
rtn_func = Sigmoid()
elif activate_func == self.k_activation_ReLU:
rtn_func = ReLU()
elif activate_func == self.k_activation_Tanh:
rtn_func = Tanh()
elif activate_func == self.k_activation_Liner:
rtn_func = Liner()
return rtn_func
# エンコード
def encode(self, x):
print('*** Encode now ***')
for i in range(len(self.z)):
print('range = %d (%d to %d Hz)' % (i, self.get_freq(i * CROMATIC + A0_NOTE_NO), self.get_freq((i + 1) * CROMATIC + A0_NOTE_NO)))
self.z[i] = []
w1 = self.w1[i]
ws1 = w1[0]
swin = x.size / ws1.size * WAVELET_SHIFT + 1
xwork = np.append(np.zeros(HALF_WINDOW_SIZE), x, axis=0)
xwork = np.append(xwork, np.zeros(HALF_WINDOW_SIZE), axis=0)
for s in range(int(swin)):
xwin = xwork[int(s * ws1.size / WAVELET_SHIFT):int(s * ws1.size / WAVELET_SHIFT) + ws1.size]
zwin = self.func.activate_func(np.dot(w1, xwin))
self.z[i].append(zwin)
return self.z
# デコード
def decode(self, z, acount):
print('*** Decode now ***')
x = [0] * (acount + BASE_WINDOW_SIZE + HALF_WINDOW_SIZE)
for i in range(len(self.z)):
print('range = %d (%d to %d Hz)' % (i, self.get_freq(i * CROMATIC + A0_NOTE_NO), self.get_freq((i + 1) * CROMATIC + A0_NOTE_NO)))
zwin = z[i]
w1 = self.w1[i]
ws1 = w1[0]
w2 = self.w2[i]
ws2 = w2[0]
swin = acount / ws1.size * WAVELET_SHIFT + 1
for s in range(int(swin)):
try:
xwk = self.func2.activate_func(np.dot(w2, zwin[s]))
except:
print('error 01')
print(s)
print(len(zwin))
try:
x[HALF_WINDOW_SIZE + int(s * ws1.size / WAVELET_SHIFT):HALF_WINDOW_SIZE + int(s * ws1.size / WAVELET_SHIFT) + ws1.size] += xwk
except:
print('error 02')
print(s)
print(ws1.size)
print(xwk.size)
return x[BASE_WINDOW_SIZE:acount + BASE_WINDOW_SIZE]
# ウェーブレットの初期設定
def wavelet_init(self, amp, astart_note, aend_note, afase, asample_size):
swav = np.empty(((aend_note - astart_note + 1) * afase, asample_size))
fs = BASE_SAMPLING
sec = asample_size / fs
freq = []
hann_window = signal.hann(asample_size)
for n in range(160)[astart_note:aend_note + 1]:
f = self.get_freq(n)
freq.append(f)
i = 0
for f in freq:
for fase in range(FASE_SPLIT):
s = amp * np.sin(2.0 * np.pi * (f * np.arange(fs * sec) / fs + (fase / FASE_SPLIT)))
swav[i] = s * hann_window
i += 1
return swav
# 指定周波数取得（平均律）
def get_freq(self, n):
note = n - A4_NOTE_NO
freq = 440.0 * (2 ** (note / CROMATIC))
return freq
class ActivateFunction(object):
def __init__(self):
self.name = self.__class__.__name__
@classmethod
# @abstractmethod
def activate_func(cls, x):
raise NotImplementedError()
@classmethod
# @abstractmethod
# 微分関数は現在つかってませんが、そのうち実験で使うかもしれないので残しています。
def differential_func(cls, x):
raise NotImplementedError()
# ---------------------------------------------------------------------------------------
# Define the activate function
# ---------------------------------------------------------------------------------------
class Sigmoid(ActivateFunction):
@classmethod
def activate_func(cls, x):
return 1. / (1. + np.exp(-x))
@classmethod
def differential_func(cls, x):
return x * (1. - x)
# ---------------------------------------------------------------------------------------
class Tanh(ActivateFunction):
@classmethod
def activate_func(cls, x):
return np.tanh(x)
@classmethod
def differential_func(cls, x):
return (1. - (np.tanh(x) * np.tanh(x)))
# ----------------------------------------------------------------------------------------
class ReLU(ActivateFunction):
@classmethod
def activate_func(cls, x):
return x * (x > 0)
@classmethod
def differential_func(cls, x):
return 1. * (x > 0)
# ----------------------------------------------------------------------------------------
class Liner(ActivateFunction):
@classmethod
def activate_func(cls, x):
return x
@classmethod
def differential_func(cls, x):
return np.ones(x.shape)
if __name__ == '__main__':
print('--------------------------------------------')
print(' _（ ）_ ') print(' |□Ｖ□| コレジャナイ Encoder V1.0') print(' | － | 2017 Koo Wells') print('--------------------------------------------')
st = soundtool()
w_channel, w_rate, w_framenumber, w_frame = st.fetch_soundData(IN_WAVE_FILE)
print('*** Start ***')
print('SoundFile = ' + IN_WAVE_FILE)
data48KMono = st.conv48KMonoArray(w_frame, w_channel, w_rate) # 48KHz モノラルの配列にする
k = korejanai() # コレジャナイエンコーダオブジェクトの取得
z = k.encode(data48KMono) # エンコードして活性化値を求める
count = len(data48KMono) # データ長を求めて置く
x_hat = k.decode(z, count) # デコードする
amax = max(np.max(x_hat), -np.min(x_hat)) # 振幅の最大値を求める
decode = x_hat / amax * 32768 * 0.98 # 16bitでほどよく収まるように振幅を調整する
decode.astype('int16')
work = b''
out_stream = b''
amp = 0
print('*** convert Array to Stream Now ***')
# これが遅いのですが、いいコーディングを見つけれていません。
# 一発でInt配列からByteストリームに変換する方法が見つかりません。
for i in range(len(decode)):
amp = int(decode[i])
ampbyte = amp.to_bytes(2, byteorder='little', signed=True) # 地道に1サンプルづつ変換します。
work = b''.join([work, ampbyte])
if i % 10000 == 0: # 一旦ワークに入れるのはそのままjoinしていくと指数的に遅くなるからです。
if i % 100000 == 0:
print('stream:%d' % (i)) # 遅いので、経過をレポートします。
out_stream = b''.join([out_stream, work])
work = b''
out_stream = b''.join([out_stream, work]) # 最後の残りをフラッシュします。
print('stream:%d' % (i))
print('*** output Now ***')
st.write_soundData(DECODE_FILE, out_stream) #ファイルに書き出します
print('DecodeFile = ' + DECODE_FILE)
print('*** Complete ! ***')
300行程度まで整理しました。
（実験編へつづく）
