build.rsudp.c_alertsound.AlertSound.__init__() - Code Metrics - Inspection of "expanding test coverage" - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Retried Inspection

Pull Request — master (#25)

by Ian

created 2021-10-18 01:41 UTC

build.rsudp.c_alertsound.AlertSound.init() A

↳ Parent: build.rsudp.c_alertsound

Complexity

Conditions

Size

Total Lines	28
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	16
dl	0
loc	28
rs	9.6
c	0
b	0
f	0
cc	2
nop	4

import sys, os
from rsudp.raspberryshake import ConsumerThread
from rsudp import printM, printW, printE
from rsudp.test import TEST
import subprocess
try:
	from pydub.playback import play
	from pydub import AudioSegment, utils
	pydub_exists = True
	# avoids import error that arises between pydub 0.23.1 and 0.25.1
	global PLAYER
	PLAYER = utils.get_player_name()
	TEST['d_pydub'][1] = True
except ImportError as e:
	global ERR

	ERR = e
	pydub_exists = False


class AlertSound(ConsumerThread):
	"""
	.. _pydub.AudioSegment: https://github.com/jiaaro/pydub/blob/master/API.markdown#audiosegment

	A consumer class that plays an alert sound when an ``ALARM`` message arrives on the queue.
	``rsudp.c_alertsound.AlertSound.sound`` is a pydub.AudioSegment_ object and is passed from the client.

	:param sta: short term average (STA) duration in seconds.
	:type sta: bool or pydub.AudioSegment_ 
	:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`.

	"""

	def _load_sound(self):
		'''
		Loads MP3 sound if possible, then writes to wav.
		Catches a ``FileNotFoundError`` when no player can be loaded.
		'''
		try:
			soundloc = self.sound
			self.sound = AudioSegment.from_file(self.sound, format="mp3")
			printM('Loaded %.2f sec alert sound from %s' % (len(self.sound)/1000., soundloc), sender=self.sender)
			self.wavloc = '%s.wav' % os.path.splitext(soundloc)[0]
			if 'ffplay' in PLAYER:
				self._write_wav()
		except FileNotFoundError as e:
			printE('Error loading player - %s' % (e), sender=self.sender)
			printW("You have chosen to play a sound, but don't have ffmpeg or libav installed.", sender=self.sender)
			printW('Sound playback requires one of these dependencies.', sender=self.sender, spaces=True)
			printW("To install either dependency, follow the instructions at:", sender=self.sender, spaces=True)
			printW('https://github.com/jiaaro/pydub#playback', sender=self.sender, spaces=True)
			printW('The program will now continue without sound playback.', sender=self.sender, spaces=True)
			self.sound = False

	def _init_sound(self):
		if pydub_exists:
			if os.path.exists(self.sound):
				self._load_sound()
			else:
				printW("The file %s could not be found." % (self.sound), sender=self.sender)
				printW('The program will now continue without sound playback.', sender=self.sender, spaces=True)
				self.sound = False
		else:
			printE('Error importing pydub - %s' % ERR, sender=self.sender)

			printW("You don't have pydub installed, so no sound will play.", sender=self.sender)
			printW('To install pydub, follow the instructions at:', sender=self.sender, spaces=True)
			printW('https://github.com/jiaaro/pydub#installation', sender=self.sender, spaces=True)
			printW('Sound playback also requires you to install either ffmpeg or libav.', sender=self.sender, spaces=True)

	def _write_wav(self):
		'''
		FFPlay can only play raw wav sounds without verbosity, so to support
		non-verbose mode we must export to .wav prior to playing a sound.
		This function checks for an existing wav file and if it does not
		exist, writes a new one.
		'''
		if not os.path.isfile(self.wavloc):
			self.sound.export(self.wavloc, format="wav")
			printM('Wrote wav version of sound file %s' % (self.wavloc), self.sender)


	def __init__(self, testing=False, soundloc=False, q=False):
		"""
		.. _pydub.AudioSegment: https://github.com/jiaaro/pydub/blob/master/API.markdown#audiosegment

		Initializes the alert sound listener thread.
		Needs a pydub.AudioSegment_ to play and a :class:`queue.Queue` to listen on.

		"""
		super().__init__()
		self.sender = 'AlertSound'
		self.alive = True
		self.testing = testing

		self.sound = soundloc
		self.devnull = open(os.devnull, 'w')
		
		self._init_sound()

		if q:
			self.queue = q
		else:
			printE('no queue passed to the consumer thread! We will exit now!',
				   self.sender)
			sys.stdout.flush()
			self.alive = False
			sys.exit()

		printM('Starting.', self.sender)

	def _play_quiet(self):
		'''
		if FFPlay is the player, suppress printed output.
		'''
		self._write_wav()
		subprocess.call([PLAYER,"-nodisp", "-autoexit", "-hide_banner",
						self.wavloc], stdout=self.devnull, stderr=self.devnull)

	def _play(self):
		printM('Playing alert sound...', sender=self.sender)
		if 'ffplay' in PLAYER:
			self._play_quiet()
		else:
			play(self.sound)
		if self.testing:
			TEST['c_play'][1] = True

	def run(self):
		"""
		Reads data from the queue and plays self.sound if it sees an ``ALARM`` message.
		Quits if it sees a ``TERM`` message.
		"""
		while True:
			d = self.queue.get()
			self.queue.task_done()
			if 'TERM' in str(d):
				self.alive = False
				self.devnull.close()
				printM('Exiting.', self.sender)
				sys.exit()
			elif 'ALARM' in str(d):
				if self.sound and pydub_exists:
					self._play()


1			import sys, os
2			from rsudp.raspberryshake import ConsumerThread
3			from rsudp import printM, printW, printE
4			from rsudp.test import TEST
5			import subprocess
6			try:
7			from pydub.playback import play
8			from pydub import AudioSegment, utils
9			pydub_exists = True
10			# avoids import error that arises between pydub 0.23.1 and 0.25.1
11			global PLAYER
12			PLAYER = utils.get_player_name()
13			TEST['d_pydub'][1] = True
14			except ImportError as e:
15			global ERR
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-10-17 22:54 UTC by Report Bug Copy Issue Report The variable `ERR` does not seem to be defined. Loading history...
16			ERR = e
17			pydub_exists = False
18
19
20			class AlertSound(ConsumerThread):
21			"""
22			.. _pydub.AudioSegment: https://github.com/jiaaro/pydub/blob/master/API.markdown#audiosegment
23
24			A consumer class that plays an alert sound when an ``ALARM`` message arrives on the queue.
25			``rsudp.c_alertsound.AlertSound.sound`` is a pydub.AudioSegment_ object and is passed from the client.
26
27			:param sta: short term average (STA) duration in seconds.
28			:type sta: bool or pydub.AudioSegment_
29			:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`.
30
31			"""
32
33			def _load_sound(self):
34			'''
35			Loads MP3 sound if possible, then writes to wav.
36			Catches a ``FileNotFoundError`` when no player can be loaded.
37			'''
38			try:
39			soundloc = self.sound
40			self.sound = AudioSegment.from_file(self.sound, format="mp3")
41			printM('Loaded %.2f sec alert sound from %s' % (len(self.sound)/1000., soundloc), sender=self.sender)
42			self.wavloc = '%s.wav' % os.path.splitext(soundloc)[0]
43			if 'ffplay' in PLAYER:
44			self._write_wav()
45			except FileNotFoundError as e:
46			printE('Error loading player - %s' % (e), sender=self.sender)
47			printW("You have chosen to play a sound, but don't have ffmpeg or libav installed.", sender=self.sender)
48			printW('Sound playback requires one of these dependencies.', sender=self.sender, spaces=True)
49			printW("To install either dependency, follow the instructions at:", sender=self.sender, spaces=True)
50			printW('https://github.com/jiaaro/pydub#playback', sender=self.sender, spaces=True)
51			printW('The program will now continue without sound playback.', sender=self.sender, spaces=True)
52			self.sound = False
53
54			def _init_sound(self):
55			if pydub_exists:
56			if os.path.exists(self.sound):
57			self._load_sound()
58			else:
59			printW("The file %s could not be found." % (self.sound), sender=self.sender)
60			printW('The program will now continue without sound playback.', sender=self.sender, spaces=True)
61			self.sound = False
62			else:
63			printE('Error importing pydub - %s' % ERR, sender=self.sender)
			0 ignored issues – show introduced 2021-10-17 22:54 UTC by Report Bug Copy Issue Report The variable `ERR` does not seem to be defined for all execution paths. Loading history...
64			printW("You don't have pydub installed, so no sound will play.", sender=self.sender)
65			printW('To install pydub, follow the instructions at:', sender=self.sender, spaces=True)
66			printW('https://github.com/jiaaro/pydub#installation', sender=self.sender, spaces=True)
67			printW('Sound playback also requires you to install either ffmpeg or libav.', sender=self.sender, spaces=True)
68
69			def _write_wav(self):
70			'''
71			FFPlay can only play raw wav sounds without verbosity, so to support
72			non-verbose mode we must export to .wav prior to playing a sound.
73			This function checks for an existing wav file and if it does not
74			exist, writes a new one.
75			'''
76			if not os.path.isfile(self.wavloc):
77			self.sound.export(self.wavloc, format="wav")
78			printM('Wrote wav version of sound file %s' % (self.wavloc), self.sender)
79
80
81			def __init__(self, testing=False, soundloc=False, q=False):
82			"""
83			.. _pydub.AudioSegment: https://github.com/jiaaro/pydub/blob/master/API.markdown#audiosegment
84
85			Initializes the alert sound listener thread.
86			Needs a pydub.AudioSegment_ to play and a :class:`queue.Queue` to listen on.
87
88			"""
89			super().__init__()
90			self.sender = 'AlertSound'
91			self.alive = True
92			self.testing = testing
93
94			self.sound = soundloc
95			self.devnull = open(os.devnull, 'w')
96
97			self._init_sound()
98
99			if q:
100			self.queue = q
101			else:
102			printE('no queue passed to the consumer thread! We will exit now!',
103			self.sender)
104			sys.stdout.flush()
105			self.alive = False
106			sys.exit()
107
108			printM('Starting.', self.sender)
109
110			def _play_quiet(self):
111			'''
112			if FFPlay is the player, suppress printed output.
113			'''
114			self._write_wav()
115			subprocess.call([PLAYER,"-nodisp", "-autoexit", "-hide_banner",
116			self.wavloc], stdout=self.devnull, stderr=self.devnull)
117
118			def _play(self):
119			printM('Playing alert sound...', sender=self.sender)
120			if 'ffplay' in PLAYER:
121			self._play_quiet()
122			else:
123			play(self.sound)
124			if self.testing:
125			TEST['c_play'][1] = True
126
127			def run(self):
128			"""
129			Reads data from the queue and plays self.sound if it sees an ``ALARM`` message.
130			Quits if it sees a ``TERM`` message.
131			"""
132			while True:
133			d = self.queue.get()
134			self.queue.task_done()
135			if 'TERM' in str(d):
136			self.alive = False
137			self.devnull.close()
138			printM('Exiting.', self.sender)
139			sys.exit()
140			elif 'ALARM' in str(d):
141			if self.sound and pydub_exists:
142			self._play()
143

raspishake / rsudp

Pull Request — master (#25)

build.rsudp.c_alertsound.AlertSound.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

build.rsudp.c_alertsound.AlertSound.init() A