pythonテクニック 演奏機能付きコード譜エディタ
pythonプログラム教室の題材として、tkinterでコード譜エディタを作って、ハモンドっぽい音で演奏する機能を付加してみます。
音を出している裏で、次のコードの音を生成する必要があるので、マルチプロセス、共有メモリを使います。
使うモジュールは、こんな感じ、他にも色々。
tkinter ＧＵＩ機能
pyaudio
from concurrent.futures.process import ProcessPoolExecutor
from logging import StreamHandler, Formatter, INFO, getLogger
from multiprocessing import Manager
１．プログラム構造
クラス sc77 コード名をもらって、PCMデータを生成
クラス hanko コード譜のGUI、sc77の起動
最下層 マルチプロセス起動、発音プロセス
名前がhankoになってるのは、電子判子ソフトを改造して作ったためです。
クラス hankoがコード１個ごとのpcmデータを生成して、共有メモリ経由で発音プロセスに渡すと、音がでて、その間にクラスsc77が次のコードのPCMを作成しておき、前の音が終わると、今作ったPCMを発音プロセスにわたす、という手順です。
２．画面の起動
self.xwidth = 800
self.ywidth = 600
print('chord diagram editor')
print('version',version_txt,version_date)
self.c0 = tk.Tk()
self.c0.title('Chord Diagram')
self.c0.geometry(str(self.xwidth)+'x'+str(self.ywidth))ごく当たり前にtkinterのウインドウ起動です。
self.canvas = tk.Canvas(self.fm1,bg='white')
self.canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
# 拡大許可でpack()
self.c0.update()
self.draw_dia() # 画面描画ルーチン
self.c0.bind('<Configure>',self.resize_dia) # ウィンドウをドラッグされたら
# 呼ぶ。 画面の大きさを可変にしたいので、expand=Trueでcanvasを定義します。
ルート画面のインスタンスself.coにbindを設定して、ドラッグされたら再描画するようにします。
３．再描画
def resize_dia(self,event):
# ウインドウのサイズを変えると飛んでくる。
self.cur_time = time.time()
delta_time = self.cur_time - self.past_time
# print ('dt=',delta_time)
self.past_time = self.cur_time
# ウインドウをドラッグしている間は書き直さない。
if not self.duty_flag:
self.c0.after(1000,self.resize_after) # 1秒後に書き直す設定
# print ('set after')
self.duty_flag = Trueずりずりとドラッグされている最中に毎回、再描画していると固まってしまうので、１秒タイマを仕掛けてそれまでは描画を止めます。
def draw_dia(self):
width = self.canvas.winfo_width()
height = self.canvas.winfo_height()
self.xwidth = self.c0.winfo_width()
self.ywidth = self.c0.winfo_height()
self.xvar.set(self.xwidth)
self.yvar.set(self.ywidth)
self.canvas.create_rectangle(10,10,width*0.8,height-10,tags='rect')
deltay = height/self.numline
deltax = (width*0.6)/self.numcolumn再描画で、ウインドウの大きさを取得して、相対座標でwidgetを描いていきます。
４．発音のプロセス
def mp_play_chord(sh_dict , sh_status , sh_chunk):
# 発音プロセス
global stream
# 音声デバイス初期化
p = pyaudio.PyAudio()
print ('mp_play_chord pyaudio done.')
stream = p.open(format=pyaudio.paFloat32,
channels=1, rate=44100, output=1)
print ('start mp_play_chord')
while True:
# 発音トリガ待ち
while sh_status.value != 1:
usleep(10)
#
chunkp = sh_chunk[:] # list型で受信
chunk =np.array(chunkp).astype(np.float32) # ndarray型に変換
stream.write(chunk.tostring())
sh_status.value = 2
# endwhile注意するのは、「stream = p.open(format=pyaudio.paFloat32,」のところで、よくあるpyaudioの例ではpyaudioから入力したデータをそのまま発音する設定ですが、そんなときにはformat=pyaudio.paInt32などになっていますが、今回使うデータはnumpyの三角関数を使うのでpaFloat32となります。
５．マルチプロセス
def start_multi():
sh_status.value = 2
futures = []
with ProcessPoolExecutor(max_workers=2) as executor:
# 発音プロセスを起動
results1 = executor.submit(mp_play_chord,sh_dict,sh_status,sh_chunk)
# 譜面プロセスの起動
results2 = executor.submit(hanko,11)
# shutdown待ち
print ('2 process execute was done.')
while sh_shutdown.value == 0:
usleep(1000000)
# print ('sh_shutdown=',sh_shutdown.value)
print('shutdown...')
results1.cancel()
results2.cancel()
for process in executor._processes.values():
process.kill()
print ('kill',process)
submitで関数をふたつ起動します。
マルチプロセスで起動すると、sys.exit()などでは終了できなくなるので、起動したあとにシャットダウン待ちに入って、共有メモリ経由で終了フラグ受け取って、２つのプロセスをキャンセルします。
６．共有メモリ
if name=="main":
manager = Manager()
sh_chunk = manager.list() # shared list for chunk
sh_dict = manager.dict() # dummy
sh_status = manager.Value('i',0) # shared integer value
sh_shutdown = manager.Value('i',0) # shared interger value
sh_currentbar = manager.Value('i',0)
main() マルチプロセスで起動したプロセス間の変数やり取りは、 from multiprocessing import Manager
を使います。
list型は普通に使えますが、Value型は、
if sh_status.value != 1:
などと使います。
三角関数の出力は、numpyのarray型なので、一度list型に変換してから共有メモリに格納し、発音側でもとに戻してから使います。
共有メモリのarray型もありますが、固定長になるので使いにくいです。
７．オルガン風のドローバー機能
def hamond(self,frequency,length , rate):
# 3rd:1.25 5th:1.5
self.draw_list = [
[ 1.0 ],
[0.5 , 1.0 , 3.0 ],
[0.5 , 0.75 , 1.0 , 2.0 ,3.0 ],
[0.5 , 0.75 , 1.0 , 2.0 ]
]
self.drawbar = self.draw_list[self.tone]
src = []
for fl in self.drawbar:
src.append( self.sine(frequency * fl , length,rate))
res = np.array([0] * len(src[0]))
for s in src:
res = res + s
res *= 0.5
return res ドローバー左側４本の周波数は、オクターブ下、下の５度、１度、オクターブ上になってます。
各周波数のsin波を生成して、加算していくだけです。
これをコードの構成音毎にやると、コードの音が出来上がります。
８．BPMの設定
def sine(self,frequency, length, rate):
length = int(length * rate * self.bpm)
factor = float(frequency) * (math.pi * 2) / rate
wave = np.sin(np.arange(length) * factor)
return wave クラス hankoの画面でBPMを設定すると、下の計算式で係数を得ます。
aa.bpm = 1.0 /(float(self.en1.get()) / 240.0)
サンプリングレートが44.1kHzなので、データを44100個並べると、1秒間発音します。
現状はとりあえず4拍子限定なので、240BPMを基準として,rate=44100 , length=1.0で240bpmになるようにしています。