|
python用的时间不长，一般用来做字符串处理、简单测试的一些小程序。最近工作中需要做一个简单的GUI应用，使用麦克录音并存成wave文件。然后就想拿wxPython练练手。
一、概述
GUI开发采用wxPython，界面编辑工具采用wxGlade，声音采集和播放采用PyAudio，小数据库采用sqlite3，最后使用py2exe打包发布。基本的应用开发流程都包括了。
二、wxGlade
界面编辑工具也是找了几个，比如wxFormBuilder，甚至是收费的DialogBlocks，wxFormBuilder 很漂亮，不过bug好像较多，经常自动退出。wxGlade有经典的Linux GUI界面风格，分立式窗体，了解了基本的原理后用起来很方便。主要是其中的sizer，add slot、insert slot增加空位，然后添加控件。也可以添加自定义的控件，只需要设置自定义控件的Class属性。然后在MainFrame的Extra code for this widget增加from YourModule import YourCLass。因为界面比较简单，我没有采用XRC资源导入的模式，而是直接生成MainFrame的代码。由于界面设计可能会变，在应用中新建一 个MainFrameEx类继承MainFrame，将事件处理放在继承类中完成。这样每次使用wxGlade编辑界面后可以直接覆盖生成的代码。
三、PyAudio
PyAudio是从PortAudio移植的，现在还是alpha版。不过使用起来还真是方便，看看网站上提供的example就可以了。没有什么大问题。需要注意多线程的问题，PyAudio对象尽量复用。注意线程中刷新wxWidget需要使用wx.CallAfter方法。
# -*- coding: UTF-8 -*-
import pyaudio
import wave
import threading
import wx
import datetime
import traceback
import os
import logging
logger = logging.getLogger("root")
class BMRecord(threading.Thread):
CHUNK = 1024
FORMAT = pyaudio.paInt16 # 至少为16位
DEVICE = 1
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 5
def __init__(self, window, audio, device, prefix):
threading.Thread.__init__(self)
self.window = window
self.audio = audio
self.prefix = prefix # wave文件命名的前缀
self.DEVICE = device
self.CHANNELS = 2 # 双通道采集
self.RECORD_SECONDS = int(window.params["length"])
self.RATE = int(window.params["rate"])
self.FORMAT = pyaudio.paInt16
def record(self):
self.filename = ""
self.filetime = ""
try :
stream = self.audio.open(format=self.FORMAT,
channels=self.CHANNELS,
rate=self.RATE,
input=True,
input_device_index=self.DEVICE,
frames_per_buffer=self.CHUNK)
frames = []
for i in range(0, int(self.RATE * self.RECORD_SECONDS / self.CHUNK)):
data = stream.read(self.CHUNK)
frames.append(data)
i = i
stream.stop_stream()
stream.close()
now = datetime.datetime.now()
strnow = now.strftime('%Y%m%d%H%M%S')
self.filetime = now.strftime('%Y-%m-%d %H:%M:%S')
savepath = self.checkPath(strnow)
# 双声道存为两个单声道文件
frames1 = []
frames2 = []
wavedata = b''.join(frames)
for i in range(len(wavedata) / 4):
frames1.append(wavedata[i * 4 : i * 4 + 2])
frames2.append(wavedata[i * 4 + 2 : i * 4 + 4])
# 双声道存储
fullpath = savepath + "/" + strnow + self.prefix + "X.wav"
wf = wave.open(fullpath, 'wb')
wf.setnchannels(2)
wf.setsampwidth(self.audio.get_sample_size(self.FORMAT))
wf.setframerate(self.RATE)
wf.writeframes(wavedata)
wf.close()
# 两个单声道存储
filenames = [strnow + self.prefix + "Z.wav", strnow + self.prefix + "Y.wav"]
fullpath = savepath + "/" + filenames[0]
wf = wave.open(fullpath, 'wb')
wf.setnchannels(1)
wf.setsampwidth(self.audio.get_sample_size(self.FORMAT))
wf.setframerate(self.RATE)
wf.writeframes(b''.join(frames1))
wf.close()
fullpath = savepath + "/" + filenames[1]
wf = wave.open(fullpath, 'wb')
wf.setnchannels(1)
wf.setsampwidth(self.audio.get_sample_size(self.FORMAT))
wf.setframerate(self.RATE)
wf.writeframes(b''.join(frames2))
wf.close()
self.message = "录制成功"
self.filenames = filenames
logger.info(filenames[0] + "," + filenames[1] + ", recorded")
return 0
except Exception:
self.message = traceback.format_exc()
logger.error(traceback.format_exc())
return -1
def checkPath(self, pathname):
curpath = os.path.abspath(os.curdir)
strdate = pathname[0:8]
fullpath = curpath + "/data/" + strdate
if not os.path.exists(fullpath) :
os.makedirs(fullpath)
return fullpath
def run(self):
ret = self.record()
wx.CallAfter(self.window.recordResult, ret, self.filenames, self.filetime, self.message)
四、sqlite
sqlite模块是Python内置的用起来很方便：
import sqlite3
import datetime
class BMDatabase():
def loadData(self, whichDay):
conn = sqlite3.connect("data/bmon.db")
cur = conn.cursor()
start = datetime.datetime.strptime(whichDay, "%Y-%m-%d")
end = start + datetime.timedelta(days=1)
res = cur.execute("select * from CheckRecord where rec_time between ? and ? order by rec_time", (start, end))
return res.fetchall()
def getFile(self, waveFile):
conn = sqlite3.connect("data/bmon.db")
cur = conn.cursor()
res = cur.execute("select * from CheckRecord where rec_file=?", (waveFile,))
row = res.fetchone()
return row
def save(self, filename, filetime, result):
conn = sqlite3.connect("data/bmon.db")
cur = conn.cursor()
record = [(filename, filetime, result)]
cur.executemany('INSERT INTO CheckRecord (rec_file,rec_time,result) VALUES (?,?,?)', record)
conn.commit()
五、自定义wxWidget控件
自绘控件主要是处理EVT_PAINT事件： self.Bind(wx.EVT_PAINT, self.on_paint)
import wx
class WavePane(wx.StaticText):
waveData = None
spectrum = None
def __init__(self, parent, nid=wx.ID_ANY, caption=""):
wx.StaticText.__init__(self, parent)
self.SetBackgroundStyle(wx.BG_STYLE_CUSTOM)
self.Bind(wx.EVT_SIZE, self.on_size)
self.Bind(wx.EVT_PAINT, self.on_paint)
def on_size(self, event):
self.Refresh()
event.Skip()
def on_paint(self, event):
w, h = self.GetClientSize()
dc = wx.AutoBufferedPaintDC(self)
brush = wx.Brush(wx.Color(0, 0, 0x80), wx.SOLID)
dc.SetBrush(brush)
dc.Clear()
dc.SetPen(wx.Pen(wx.BLACK, 1))
dc.SetTextForeground(wx.Color(0, 0xFF, 0))
font = dc.GetFont()
font.SetPointSize(8)
dc.SetFont(font)
dc.DrawRectangle(0, 0, w - 1, h - 1)
if self.waveData <> None:
dc.BeginDrawing()
dc.SetPen(wx.Pen(wx.Color(0, 0xFF, 0), 1))
au = self.waveData
step = int(au.nframes / w)
height = au.height # 或者65536 / 2.0
i = 0
j = 0
while i < au.nframes:
if au.frames[2 * i + 1] >= 0x80: # 负数
value = au.frames[2 * i] + au.frames[2 * i + 1] * 256 - 65536
else:
value = au.frames[2 * i] + au.frames[2 * i + 1] * 256
dc.DrawLine(j, int(h / 2.0), j, int(h / 2.0 * (1 - value * 1.0 / height)))
i += step
j += 1
if 2 * i + 1 >= au.nframes * 2:
break;
dc.DrawText(str(au.maxValue), 1, 1)
dc.DrawText(str(au.minValue), 1, h - 16)
dc.EndDrawing()
elif self.spectrum <> None :
dc.BeginDrawing()
dc.SetPen(wx.Pen(wx.Color(0, 0xFF, 0), 1))
brush = wx.Brush(wx.Color(0, 0xFF, 0), wx.SOLID)
dc.SetBrush(brush)
dc.SetTextForeground(wx.RED)
barWidth = int(w / 72)
i = 0
j = barWidth
while i < 36:
y = int((1 - self.spectrum[i] * 1.0 / self.maxSpectrum) * h)
dc.DrawRectangle(j, y , barWidth, h - y - 1)
if y < h - 2:
dc.FloodFill(j + 1, y + 1, wx.Color(0, 0xFF, 0), wx.FLOOD_BORDER)
# Bar的编号
if i == 0 or(i + 1) % 5 == 0:
if i < 9:
dc.DrawText(str(i + 1), j + 4, h - 13)
else:
dc.DrawText(str(i + 1), j , h - 13)
i += 1
j += barWidth * 2
dc.SetTextForeground(wx.Color(0, 0xFF, 0))
dc.DrawText(str(self.maxSpectrum), 1, 1)
dc.EndDrawing()
def setWaveData(self, waveData):
self.waveData = waveData
self.Refresh()
def setSpectrum(self, spectrum):
self.spectrum = spectrum
if spectrum <> None:
maxValue = 0
for i in range(0, 36):
if spectrum[i] > maxValue :
maxValue = spectrum[i]
self.maxSpectrum = maxValue
self.Refresh()
六、日志的使用
使用循环日志
import logging.handlers
logger = logging.getLogger("root")
handler = logging.handlers.RotatingFileHandler(os.path.join(os.getcwd(), 'bmon.log'),
maxBytes=5 * 1024 * 1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
七、py2exe
最后一步就是打包成可执行文件。setup.py：
from distutils.core import setup
import py2exe
# setup(console=["hello.py"])
py2exe_options = dict({
"includes":['sip', 'encodings', 'encodings.ascii', 'encodings.utf_8', 'encodings.cp866'],
"dll_excludes":["MSVCP90.dll"]})
setup(version="1.0",
description="Bearing Monitor",
name="bmon",
zipfile=None,
dist_dir="bmon",
windows=["bmon.py"],
options={'py2exe': py2exe_options},
icon_resources=[(1, "check_all.ico")],
data_files=[("", ["check_all.ico"])]
)
然后命令行下执行：python setup.py py2exe，就可以生成dist发布目录
八、小结
这个简单应用涉及的主要模块就这么几个，组合成了一个简单的GUI应用。Python开发还真是很简单，前提是得熟悉各种模块。
|
