File indexing completed on 2024-04-28 07:51:11
0001 # -*- coding: utf-8 -*- 0002 0003 """ 0004 Copyright (C) 2010-2016 Wolfgang Rohdewald <wolfgang@rohdewald.de> 0005 0006 SPDX-License-Identifier: GPL-2.0 0007 0008 """ 0009 0010 import os 0011 import tarfile 0012 import subprocess 0013 import datetime 0014 from io import BytesIO 0015 from hashlib import md5 0016 0017 from common import Debug, Internal, StrMixin, cacheDir 0018 from util import which, removeIfExists, uniqueList, elapsedSince 0019 from log import logWarning, i18n, logDebug, logException 0020 0021 from qt import QStandardPaths 0022 0023 from tile import Tile 0024 0025 # Phonon does not work with short files - it plays them 0026 # simultaneously or only parts of them. Mar 2010, KDE 4.4. True for mp3 0027 # and for wav. Also, mpg123 often plays distorted sounds. Kubuntu 9.10. 0028 # So we use ogg123 and ogg sound files. 0029 # self.audio = Phonon.MediaObject(self) 0030 # self.audioOutput = Phonon.AudioOutput(Phonon.GameCategory, self) 0031 # Phonon.createPath(self.audio, self.audioOutput) 0032 # self.audio.enqueue(Phonon.MediaSource(wavName)) 0033 # self.audio.play() 0034 0035 if os.name == 'nt': 0036 import winsound # pylint: disable=import-error 0037 0038 0039 class Sound: 0040 0041 """the sound interface. Use class variables and class methods, 0042 thusly ensuring no two instances try to speak""" 0043 __oggBinary = None 0044 __bonusOgg = None 0045 playProcesses = [] 0046 lastCleaned = None 0047 0048 @staticmethod 0049 def findOggBinary(): 0050 """set __oggBinary to exe name or an empty string""" 0051 if Sound.__oggBinary is None: 0052 if os.name == 'nt': 0053 Sound.__oggBinary = os.path.join('share', 'kajongg', 'voices', 'oggdec.exe') 0054 msg = '' # we bundle oggdec.exe with the kajongg installer, it must be there 0055 else: 0056 oggBinary = 'ogg123' 0057 msg = i18n( 0058 'No voices will be heard because the program %1 is missing', 0059 oggBinary) 0060 if which(oggBinary): 0061 Sound.__oggBinary = oggBinary 0062 else: 0063 Sound.__oggBinary = '' 0064 Internal.Preferences.useSounds = False 0065 # checks again at next reenable 0066 if msg: 0067 logWarning(msg) 0068 if Debug.sound: 0069 logDebug('ogg123 found:' + Sound.__oggBinary) 0070 return Sound.__oggBinary 0071 0072 @staticmethod 0073 def cleanProcesses(): 0074 """terminate ogg123 children""" 0075 now = datetime.datetime.now() 0076 if Sound.lastCleaned and (now - Sound.lastCleaned).seconds < 2: 0077 return 0078 Sound.lastCleaned = now 0079 remaining = [] 0080 for process in Sound.playProcesses: 0081 if process.poll() is not None: 0082 # ogg123 self-finished 0083 continue 0084 diff = now - process.startTime 0085 if diff.seconds > 10: 0086 try: 0087 process.kill() 0088 except OSError: 0089 pass 0090 try: 0091 os.waitpid(process.pid, 0) 0092 except OSError: 0093 pass 0094 if Debug.sound: 0095 game = Internal.scene.game 0096 game.debug('10 seconds passed. Killing %s' % process.name) 0097 else: 0098 remaining.append(process) 0099 Sound.playProcesses = remaining 0100 0101 @staticmethod 0102 def speak(what): 0103 """this is what the user of this module will call.""" 0104 # pylint: disable=too-many-branches 0105 if not Internal.Preferences.useSounds: 0106 return 0107 game = Internal.scene.game 0108 reactor = Internal.reactor 0109 if game and not game.autoPlay and Sound.playProcesses: 0110 # in normal play, wait a moment between two speaks. Otherwise 0111 # sometimes too many simultaneous speaks make them ununderstandable 0112 lastSpeakStart = max(x.startTime for x in Sound.playProcesses) 0113 if elapsedSince(lastSpeakStart) < 0.3: 0114 reactor.callLater(1, Sound.speak, what) 0115 return 0116 if os.path.exists(what): 0117 oggBinary = Sound.findOggBinary() 0118 if oggBinary: 0119 if os.name == 'nt': 0120 # convert to .wav, store .wav in cacheDir 0121 name, ext = os.path.splitext(what) 0122 assert ext == '.ogg', 'what: {} name: {} ext: {}'.format(what, name, ext) 0123 if 'bell' in name: 0124 nameParts = ['bell'] 0125 else: 0126 nameParts = os.path.normpath(name).split(os.sep) 0127 nameParts = nameParts[nameParts.index('voices') + 1:] 0128 wavName = os.path.normpath( 0129 '{}/{}.wav'.format(cacheDir(), 0130 '_'.join(nameParts))) 0131 if not os.path.exists(wavName): 0132 args = [oggBinary, '-a', '-w', wavName, os.path.normpath(what)] 0133 startupinfo = subprocess.STARTUPINFO() 0134 startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW 0135 subprocess.call(args, startupinfo=startupinfo) 0136 if Debug.sound: 0137 logDebug('converted {} to wav {}'.format(what, wavName)) 0138 try: 0139 winsound.PlaySound( 0140 wavName, 0141 winsound.SND_FILENAME | winsound.SND_NODEFAULT) 0142 except RuntimeError: 0143 pass 0144 else: 0145 args = [oggBinary, '-q', what] 0146 if Debug.sound: 0147 game.debug(' '.join(args)) 0148 process = subprocess.Popen(args) 0149 process.startTime = datetime.datetime.now() 0150 process.name = what 0151 Sound.playProcesses.append(process) 0152 reactor.callLater(3, Sound.cleanProcesses) 0153 reactor.callLater(6, Sound.cleanProcesses) 0154 0155 @staticmethod 0156 def bonus(): 0157 """ring some sort of bell, if we find such a file""" 0158 if Sound.__bonusOgg is None: 0159 Sound.__bonusOgg = '' 0160 for oggName in ( 0161 '/usr/share/sounds/KDE-Im-Message-In.ogg', 0162 'share/sounds/bell.ogg'): 0163 if os.path.exists(oggName): 0164 Sound.__bonusOgg = oggName 0165 if Debug.sound: 0166 logDebug('Bonus sound found:{}'.format(oggName)) 0167 break 0168 if Debug.sound and not Sound.__bonusOgg: 0169 logDebug('No bonus sound found') 0170 if Sound.__bonusOgg: 0171 Sound.speak(Sound.__bonusOgg) 0172 0173 class Voice(StrMixin): 0174 0175 """this administers voice sounds. 0176 0177 When transporting voices between players, a compressed tarfile 0178 is generated at source and transferred to destination. At 0179 destination, no tarfile is written, only the content. It makes 0180 only sense to cache the voice in a tarfile at source.""" 0181 0182 __availableVoices = [] 0183 md5sumLength = 32 # magical constant 0184 0185 def __init__(self, directory, content=None): 0186 """give this name a voice""" 0187 self.__md5sum = None 0188 if not os.path.split(directory)[0]: 0189 if Debug.sound: 0190 logDebug('place voice %s in %s' % (directory, cacheDir())) 0191 directory = os.path.join(cacheDir(), directory) 0192 self.directory = directory 0193 self.__setArchiveContent(content) 0194 0195 def __str__(self): 0196 return self.directory 0197 0198 def language(self): 0199 """the language code of this voice. Locally defined voices 0200 have no language code and return 'local'. 0201 remote voices received from other clients return 'remote', 0202 they always get predecence.""" 0203 if len(self.directory) == self.md5sumLength: 0204 if os.path.split(self.directory)[1] == self.md5sum: 0205 return 'remote' 0206 if 'HOME' in os.environ: 0207 home = os.environ['HOME'] 0208 elif 'HOMEPATH' in os.environ: 0209 home = os.environ['HOMEPATH'] 0210 else: 0211 logException('have neither HOME nor HOMEPATH') 0212 if home: 0213 if self.directory.startswith(home): 0214 return 'local' 0215 result = os.path.split(self.directory)[0] 0216 result = os.path.split(result)[1] 0217 if result == 'voices': 0218 result = 'en_US' 0219 return result 0220 0221 @staticmethod 0222 def availableVoices(): 0223 """a list of all voice directories""" 0224 if not Voice.__availableVoices: 0225 result = [] 0226 directories = QStandardPaths.locateAll( 0227 QStandardPaths.AppDataLocation, 'voices', QStandardPaths.LocateDirectory) 0228 directories.insert(0, os.path.join('share', 'kajongg', 'voices')) 0229 for parentDirectory in directories: 0230 for (dirpath, _, _) in os.walk(parentDirectory, followlinks=True): 0231 if os.path.exists(os.path.join(dirpath, 's1.ogg')): 0232 result.append(Voice(dirpath)) 0233 group = Internal.kajonggrc.group('Locale') 0234 prefLanguages = uniqueList( 0235 ':'.join(['local', str(group.readEntry('Language')), 'en_US']).split(':')) 0236 prefLanguages = dict(enumerate(prefLanguages)) 0237 result = sorted( 0238 result, key=lambda x: prefLanguages.get(x.language(), 9999)) 0239 if Debug.sound: 0240 logDebug('found voices:%s' % [str(x) for x in result]) 0241 Voice.__availableVoices = result 0242 return Voice.__availableVoices 0243 0244 @staticmethod 0245 def locate(name): 0246 """return Voice or None if no foreign or local voice matches. 0247 In other words never return a predefined voice""" 0248 for voice in Voice.availableVoices(): 0249 dirname = os.path.split(voice.directory)[-1] 0250 if name == voice.md5sum: 0251 if Debug.sound: 0252 logDebug( 0253 'locate found %s by md5sum in %s' % 0254 (name, voice.directory)) 0255 return voice 0256 if name == dirname and voice.language() == 'local': 0257 if Debug.sound: 0258 logDebug( 0259 'locate found %s by name in %s' % 0260 (name, voice.directory)) 0261 return voice 0262 if Debug.sound: 0263 logDebug('Personal sound for %s not found' % (name)) 0264 return None 0265 0266 def buildSubvoice(self, oggName, side): 0267 """side is 'left' or 'right'.""" 0268 angleDirectory = os.path.join( 0269 cacheDir(), 0270 'angleVoices', 0271 self.md5sum, 0272 side) 0273 stdName = os.path.join(self.directory, oggName) 0274 angleName = os.path.join(angleDirectory, oggName) 0275 if os.path.exists(stdName) and not os.path.exists(angleName): 0276 sox = which('sox') 0277 if not sox: 0278 return stdName 0279 if not os.path.exists(angleDirectory): 0280 os.makedirs(angleDirectory) 0281 args = [sox, stdName, angleName, 'remix'] 0282 if side == 'left': 0283 args.extend(['1,2', '0']) 0284 elif side == 'right': 0285 args.extend(['0', '1,2']) 0286 callResult = subprocess.call(args) 0287 if callResult: 0288 if Debug.sound: 0289 logDebug( 0290 'failed to build subvoice %s: return code=%s' % 0291 (angleName, callResult)) 0292 return stdName 0293 if Debug.sound: 0294 logDebug('built subvoice %s' % angleName) 0295 return angleName 0296 0297 def localTextName(self, text, angle): 0298 """build the name of the wanted sound file""" 0299 oggName = text.lower().replace(' ', '') + '.ogg' 0300 if angle == 90: 0301 return self.buildSubvoice(oggName, 'left') 0302 if angle == 270: 0303 return self.buildSubvoice(oggName, 'right') 0304 return os.path.join(self.directory, oggName) 0305 0306 def speak(self, text, angle): 0307 """text must be a sound filename without extension""" 0308 if isinstance(text, Tile): 0309 text = str(text.exposed) 0310 fileName = self.localTextName(text, angle) 0311 if not os.path.exists(fileName): 0312 if Debug.sound: 0313 logDebug('Voice.speak: fileName %s not found' % fileName) 0314 Sound.speak(fileName) 0315 0316 def oggFiles(self): 0317 """a list of all found ogg files""" 0318 if os.path.exists(self.directory): 0319 return sorted(x for x in os.listdir(self.directory) if x.endswith('.ogg')) 0320 return [] 0321 0322 def __computeMd5sum(self): 0323 """update md5sum file. If it changed, return True. 0324 If unchanged or no ogg files exist, remove archive and md5sum and return False. 0325 If ogg files exist but no archive, return True.""" 0326 if self.__md5sum: 0327 # we already checked 0328 return 0329 md5FileName = os.path.join(self.directory, 'md5sum') 0330 archiveExists = os.path.exists(self.archiveName()) 0331 ogg = self.oggFiles() 0332 if not ogg: 0333 removeIfExists(self.archiveName()) 0334 removeIfExists(md5FileName) 0335 self.__md5sum = None 0336 logDebug('no ogg files in %s' % self) 0337 return 0338 md5sum = md5() 0339 for oggFile in ogg: 0340 md5sum.update( 0341 open(os.path.join(self.directory, oggFile), 'rb').read()) 0342 # the md5 stamp goes into the old archive directory 'username' 0343 self.__md5sum = md5sum.hexdigest() 0344 existingMd5sum = self.savedmd5Sum() 0345 md5Name = self.md5FileName() 0346 if self.__md5sum != existingMd5sum: 0347 if Debug.sound: 0348 if not os.path.exists(md5Name): 0349 logDebug('creating new %s' % md5Name) 0350 else: 0351 logDebug( 0352 'md5sum %s changed, rewriting %s with %s' % 0353 (existingMd5sum, md5Name, self.__md5sum)) 0354 try: 0355 open(md5Name, 'w').write('%s\n' % self.__md5sum) 0356 except OSError as exception: 0357 logException( 0358 '\n'.join([i18n('cannot write <filename>%1</filename>: %2', 0359 md5Name, 0360 str(exception)), 0361 i18n('The voice files have changed, their checksum has changed.'), 0362 i18n('Please reinstall kajongg or do, with sufficient permissions:'), 0363 'cd {} ; cat *.ogg | md5sum > md5sum'.format(self.directory)])) 0364 if archiveExists: 0365 archiveIsOlder = os.path.getmtime( 0366 md5Name) > os.path.getmtime(self.archiveName()) 0367 if self.__md5sum != existingMd5sum or archiveIsOlder: 0368 os.remove(self.archiveName()) 0369 0370 def __buildArchive(self): 0371 """write the archive file and set self.__md5sum""" 0372 self.__computeMd5sum() 0373 if not os.path.exists(self.archiveName()): 0374 tarFile = tarfile.open(self.archiveName(), mode='w:bz2') 0375 for oggFile in self.oggFiles(): 0376 tarFile.add( 0377 os.path.join( 0378 self.directory, 0379 oggFile), 0380 arcname=oggFile) 0381 tarFile.close() 0382 0383 def archiveName(self): 0384 """ the full path of the archive file""" 0385 return os.path.join(self.directory, 'content.tbz') 0386 0387 def md5FileName(self): 0388 """the name of the md5sum file""" 0389 return os.path.join(self.directory, 'md5sum') 0390 0391 def savedmd5Sum(self): 0392 """return the current value of the md5sum file""" 0393 if os.path.exists(self.md5FileName()): 0394 try: 0395 line = open(self.md5FileName(), 'r').readlines()[0].replace(' -', '').strip() 0396 if len(line) == self.md5sumLength: 0397 return line 0398 logWarning('{} has wrong content: {}'.format(self.md5FileName(), line)) 0399 except OSError as exc: 0400 logWarning('{} has wrong content: {}'.format(self.md5FileName(), exc)) 0401 return None 0402 0403 @property 0404 def md5sum(self): 0405 """the current checksum over all ogg files""" 0406 self.__computeMd5sum() 0407 return self.__md5sum 0408 0409 def __setArchiveContent(self, content): 0410 """fill the Voice with ogg files""" 0411 if not content: 0412 return 0413 if not os.path.exists(self.directory): 0414 os.makedirs(self.directory) 0415 filelike = BytesIO(content) 0416 tarFile = tarfile.open(mode='r|bz2', fileobj=filelike) 0417 tarFile.extractall(path=self.directory) 0418 if Debug.sound: 0419 logDebug('extracted archive into %s' % self.directory) 0420 tarFile.close() 0421 filelike.close() 0422 0423 @property 0424 def archiveContent(self): 0425 """the content of the tarfile""" 0426 self.__buildArchive() 0427 if os.path.exists(self.archiveName()): 0428 return open(self.archiveName(), 'rb').read() 0429 return None 0430 0431 @archiveContent.setter 0432 def archiveContent(self, content): 0433 """new archive content""" 0434 self.__setArchiveContent(content)