<commit_before>import pyaudio
import numpy
import math
import time
from functools import reduce

def numpyMapOneDimensional( func, values ):
	output = numpy.empty_like( values )
	for i in range( values.size ):
		output[ i ] = func( values[ i ] )
	return output
	
def scale( x, xMin, xMax ):
	return min( 1, max( 0, ( x - xMin ) ) / ( xMax - xMin ) )

class Int16Settings:
	def __init__( self ):
		self.npArrayDtype = numpy.int16
		self.npSumDtype = numpy.uint64
		self.npDifferenceDtype = numpy.int64
		self.paFormat = pyaudio.paInt16
		self.maxValue = 1 << 15		

class Float32Settings:
	def __init__( self ):
		self.npArrayDtype = numpy.float32
		self.npSumDtype = numpy.float32
		self.npDifferenceDtype = numpy.float32
		self.paFormat = pyaudio.paFloat32
		self.maxValue = numpy.float32( 1 )		

class RollingPowerWindowEvaluator:
	def __init__( self, audioBufferLength, smoothingWindowHalfSize, typeSettings ):
		self.smoothingWindowHalfSize = smoothingWindowHalfSize
		self.typeSettings = typeSettings

		self.totalPower = numpy.float32( 0 )
		self.totalSquaresOfPower = numpy.float32( 0 )
		self.powerStdDevEstimateTimesBufferSize = numpy.float32( 0 )
		self.outlyingPowerThreshhold = numpy.float32( 0 )
		self.totalOutlyingPower = numpy.float32( 0 )
		
		self.amplitudeBuffer = numpy.zeros( audioBufferLength, dtype = self.typeSettings.npArrayDtype )
		self.powerWindowBuffer = numpy.zeros( audioBufferLength, dtype = numpy.float32 )
		self.outlyingPowerBuffer = numpy.zeros( audioBufferLength, dtype = numpy.float32 )

		# make odd length smoothing window so it can be centered on an integer index
		smoothingWindowSize = ( self.smoothingWindowHalfSize * 2 ) + 1
		# simple average of neigbors... each weighted 1 / window size
		self.smoothingWindowFunction = numpy.full( smoothingWindowSize, 1 / float( smoothingWindowSize ) )
		
	def addData( self, data ):
		newAmplitudes = numpy.abs( data )

		self.amplitudeBuffer = numpy.roll( self.amplitudeBuffer, -newAmplitudes.size )
		self.amplitudeBuffer[ -newAmplitudes.size: ] = newAmplitudes
		
		self.powerWindowBuffer = numpy.roll( self.powerWindowBuffer, -newAmplitudes.size )
		
		# Convolve applies the smoothing function to the amplitudes.
		# Boundary effects exist for output[:halfWindowSize] and output[-halfWindowSize:]
		# Boundary effects from the last update have been rolled down the buffer to powerWindowBuffer[ -(newAmplitudes.size + halfWindowSize):-newAmplitudes.size ]
		# To correct those (now that there are enough values), convlove must run on input values starting an additional halfWindowSize earlier ( -(newAmplitudes.size + halfWindowSize + halfWindowSize) ), discarding the front-end boundary errors: output[:halfWindowSize]
		replacementIndex = -( newAmplitudes.size + self.smoothingWindowHalfSize )
		outgoingPowers = self.powerWindowBuffer[ replacementIndex: ]
		# Note that for all of the cumulative values, terrible sins of floatinng point arithmetic are committed resulting in some floating point error accumulation over time.
		# Since this application isn't remotely #SCIENCE, I'm leaving the Knuthification for future work. For long periods of time, it might be helpful to "reset". (For now that means restarting the script.)
		self.totalPower -= outgoingPowers.sum()
		self.totalSquaresOfPower -= numpy.dot( outgoingPowers, outgoingPowers )
		incomingPowers = numpy.convolve(
			self.amplitudeBuffer[ ( replacementIndex - self.smoothingWindowHalfSize - 5): ],
			self.smoothingWindowFunction,
			'same' )[ replacementIndex: ]
		self.powerWindowBuffer[ replacementIndex: ] = incomingPowers
			
		self.totalPower += incomingPowers.sum()
		self.totalSquaresOfPower += numpy.dot( incomingPowers, incomingPowers )
		self.powerStdDevEstimateTimesBufferSize = math.sqrt( max( 0, ( self.powerWindowBuffer.size * self.totalSquaresOfPower ) - ( self.totalPower * self.totalPower ) ) )
		self.outlyingPowerThreshhold = ( self.totalPower + self.powerStdDevEstimateTimesBufferSize ) / self.powerWindowBuffer.size

		# Need a different way of determining a good threshhold and keeping it steady for this to work, but it showed some promise.
		self.outlyingPowerBuffer = numpy.roll( self.outlyingPowerBuffer, -newAmplitudes.size )		
		self.totalOutlyingPower -= self.outlyingPowerBuffer[ replacementIndex: ].sum()
		self.outlyingPowerBuffer[ replacementIndex: ] = numpy.clip( self.powerWindowBuffer[ replacementIndex: ] - self.outlyingPowerThreshhold, 0, self.typeSettings.maxValue )
		self.totalOutlyingPower += self.outlyingPowerBuffer[ replacementIndex: ].sum()
		
		# Unsurprisingly too slow most of the time, calculating power over current overall threshhold for each incremental data instead, which isn't the same thing, but still useful in its own way.
		#self.totalOutlyingPower = reduce( lambda sum, value: sum + max( 0, value - self.outlyingPowerThreshhold ), self.powerWindowBuffer )
	
	def getTotalPowerPercent( self ):
		return self.totalPower / ( self.amplitudeBuffer.size * float( self.typeSettings.maxValue ) )
	
	def getOutlyingPowerPercent( self ):
		return self.totalOutlyingPower / ( self.amplitudeBuffer.size * ( self.typeSettings.maxValue - self.outlyingPowerThreshhold ) )
		
	def getPowerCoefficientOfVariation( self ):
		if self.totalPower == 0:
			return 0
		return self.powerStdDevEstimateTimesBufferSize / self.totalPower
		
	def getTalkLevelEstimate( self, cvMin = 0.5, cvMax = 0.9, pMin = 0.005, pMax = 0.02 ):
		return scale( self.getPowerCoefficientOfVariation(), cvMin, cvMax ) * scale( self.getTotalPowerPercent(), pMin, pMax )
	
class AudioMonitor:
	def __init__( self, audioInterface, deviceIndex = 0, bufferSeconds = 3 ):
		self.audioInterface = audioInterface
		self.deviceIndex = deviceIndex
		self.bufferSeconds = bufferSeconds
		self.active = False
		self.muteDetected = False
		self.audioInputStream = None
		self.typeSettings = None
		self.audioInputFrameCount = 4096
		
	def initializeStream( self ):
		deviceInfo = audioInterface.get_device_info_by_index( micDeviceIndex )
		# HACK: arbitrary, but modern computer input sample rates are much higher than necessary for this application.
		sampleRate = round( deviceInfo[ 'defaultSampleRate' ] / 2 )
		audioBufferLength = math.ceil( sampleRate * self.bufferSeconds )
		
		# Half of 85Hz wave, the low end of human voice. Half since taking abs value flips the bottom end of the wave up to another peak.
		# Then half again because knowing half of the window size is helpful below.
		smoothingWindowHalfSize = round( sampleRate / ( 85 * 2 * 2 ) )

		self.typeSettings = Int16Settings()
#		if not self.audioInterface.is_format_supported( sampleRate, input_device = self.deviceIndex, input_channels=1, input_format=None, output_device=None, output_channels=None, output_format=None)
#		self.typeSettings = Float32Settings()

		self.rollingPowerWindowEvaluator = RollingPowerWindowEvaluator( audioBufferLength, smoothingWindowHalfSize, self.typeSettings )
		
		self.audioInputStream = audioInterface.open( 
			input_device_index = self.deviceIndex,
			format = self.typeSettings.paFormat,
			channels = 1,
			rate = sampleRate,
			input = True,
			frames_per_buffer = self.audioInputFrameCount,
			stream_callback = lambda data, frameCount, timeInfo, statusFlags: self.receiveData( data, frameCount, timeInfo, statusFlags ) )
	
	def start( self ):
		if self.audioInputStream is None:
			self.initializeStream()<commit_msg>Attempt at auto-detecting device input format (could use improvement.<commit_after>import pyaudio
import numpy
import math
import time
from functools import reduce

def numpyMapOneDimensional( func, values ):
	output = numpy.empty_like( values )
	for i in range( values.size ):
		output[ i ] = func( values[ i ] )
	return output
	
def scale( x, xMin, xMax ):
	return min( 1, max( 0, ( x - xMin ) ) / ( xMax - xMin ) )

class Int16Settings:
	def __init__( self ):
		self.npArrayDtype = numpy.int16
		self.npSumDtype = numpy.uint64
		self.npDifferenceDtype = numpy.int64
		self.paFormat = pyaudio.paInt16
		self.maxValue = 1 << 15		

class Float32Settings:
	def __init__( self ):
		self.npArrayDtype = numpy.float32
		self.npSumDtype = numpy.float32
		self.npDifferenceDtype = numpy.float32
		self.paFormat = pyaudio.paFloat32
		self.maxValue = numpy.float32( 1 )		

class RollingPowerWindowEvaluator:
	def __init__( self, audioBufferLength, smoothingWindowHalfSize, typeSettings ):
		self.smoothingWindowHalfSize = smoothingWindowHalfSize
		self.typeSettings = typeSettings

		self.totalPower = numpy.float32( 0 )
		self.totalSquaresOfPower = numpy.float32( 0 )
		self.powerStdDevEstimateTimesBufferSize = numpy.float32( 0 )
		self.outlyingPowerThreshhold = numpy.float32( 0 )
		self.totalOutlyingPower = numpy.float32( 0 )
		
		self.amplitudeBuffer = numpy.zeros( audioBufferLength, dtype = self.typeSettings.npArrayDtype )
		self.powerWindowBuffer = numpy.zeros( audioBufferLength, dtype = numpy.float32 )
		self.outlyingPowerBuffer = numpy.zeros( audioBufferLength, dtype = numpy.float32 )

		# make odd length smoothing window so it can be centered on an integer index
		smoothingWindowSize = ( self.smoothingWindowHalfSize * 2 ) + 1
		# simple average of neigbors... each weighted 1 / window size
		self.smoothingWindowFunction = numpy.full( smoothingWindowSize, 1 / float( smoothingWindowSize ) )
		
	def addData( self, data ):
		newAmplitudes = numpy.abs( data )

		self.amplitudeBuffer = numpy.roll( self.amplitudeBuffer, -newAmplitudes.size )
		self.amplitudeBuffer[ -newAmplitudes.size: ] = newAmplitudes
		
		self.powerWindowBuffer = numpy.roll( self.powerWindowBuffer, -newAmplitudes.size )
		
		# Convolve applies the smoothing function to the amplitudes.
		# Boundary effects exist for output[:halfWindowSize] and output[-halfWindowSize:]
		# Boundary effects from the last update have been rolled down the buffer to powerWindowBuffer[ -(newAmplitudes.size + halfWindowSize):-newAmplitudes.size ]
		# To correct those (now that there are enough values), convlove must run on input values starting an additional halfWindowSize earlier ( -(newAmplitudes.size + halfWindowSize + halfWindowSize) ), discarding the front-end boundary errors: output[:halfWindowSize]
		replacementIndex = -( newAmplitudes.size + self.smoothingWindowHalfSize )
		outgoingPowers = self.powerWindowBuffer[ replacementIndex: ]
		# Note that for all of the cumulative values, terrible sins of floatinng point arithmetic are committed resulting in some floating point error accumulation over time.
		# Since this application isn't remotely #SCIENCE, I'm leaving the Knuthification for future work. For long periods of time, it might be helpful to "reset". (For now that means restarting the script.)
		self.totalPower -= outgoingPowers.sum()
		self.totalSquaresOfPower -= numpy.dot( outgoingPowers, outgoingPowers )
		incomingPowers = numpy.convolve(
			self.amplitudeBuffer[ ( replacementIndex - self.smoothingWindowHalfSize - 5): ],
			self.smoothingWindowFunction,
			'same' )[ replacementIndex: ]
		self.powerWindowBuffer[ replacementIndex: ] = incomingPowers
			
		self.totalPower += incomingPowers.sum()
		self.totalSquaresOfPower += numpy.dot( incomingPowers, incomingPowers )
		self.powerStdDevEstimateTimesBufferSize = math.sqrt( max( 0, ( self.powerWindowBuffer.size * self.totalSquaresOfPower ) - ( self.totalPower * self.totalPower ) ) )
		self.outlyingPowerThreshhold = ( self.totalPower + self.powerStdDevEstimateTimesBufferSize ) / self.powerWindowBuffer.size

		# Need a different way of determining a good threshhold and keeping it steady for this to work, but it showed some promise.
		self.outlyingPowerBuffer = numpy.roll( self.outlyingPowerBuffer, -newAmplitudes.size )		
		self.totalOutlyingPower -= self.outlyingPowerBuffer[ replacementIndex: ].sum()
		self.outlyingPowerBuffer[ replacementIndex: ] = numpy.clip( self.powerWindowBuffer[ replacementIndex: ] - self.outlyingPowerThreshhold, 0, self.typeSettings.maxValue )
		self.totalOutlyingPower += self.outlyingPowerBuffer[ replacementIndex: ].sum()
		
		# Unsurprisingly too slow most of the time, calculating power over current overall threshhold for each incremental data instead, which isn't the same thing, but still useful in its own way.
		#self.totalOutlyingPower = reduce( lambda sum, value: sum + max( 0, value - self.outlyingPowerThreshhold ), self.powerWindowBuffer )
	
	def getTotalPowerPercent( self ):
		return self.totalPower / ( self.amplitudeBuffer.size * float( self.typeSettings.maxValue ) )
	
	def getOutlyingPowerPercent( self ):
		return self.totalOutlyingPower / ( self.amplitudeBuffer.size * ( self.typeSettings.maxValue - self.outlyingPowerThreshhold ) )
		
	def getPowerCoefficientOfVariation( self ):
		if self.totalPower == 0:
			return 0
		return self.powerStdDevEstimateTimesBufferSize / self.totalPower
		
	def getTalkLevelEstimate( self, cvMin = 0.5, cvMax = 0.9, pMin = 0.005, pMax = 0.02 ):
		return scale( self.getPowerCoefficientOfVariation(), cvMin, cvMax ) * scale( self.getTotalPowerPercent(), pMin, pMax )
	
class AudioMonitor:
	def __init__( self, audioInterface, deviceIndex = 0, bufferSeconds = 3 ):
		self.audioInterface = audioInterface
		self.deviceIndex = deviceIndex
		self.bufferSeconds = bufferSeconds
		self.active = False
		self.muteDetected = False
		self.audioInputStream = None
		self.typeSettings = None
		self.audioInputFrameCount = 4096
		
	def initializeStream( self ):
		deviceInfo = audioInterface.get_device_info_by_index( micDeviceIndex )
		# HACK: arbitrary, but modern computer input sample rates are much higher than necessary for this application.
		sampleRate = round( deviceInfo[ 'defaultSampleRate' ] / 2 )
		audioBufferLength = math.ceil( sampleRate * self.bufferSeconds )
		
		# Half of 85Hz wave, the low end of human voice. Half since taking abs value flips the bottom end of the wave up to another peak.
		# Then half again because knowing half of the window size is helpful below.
		smoothingWindowHalfSize = round( sampleRate / ( 85 * 2 * 2 ) )

		self.typeSettings = Int16Settings()
		if not self.audioInterface.is_format_supported( sampleRate, input_device = self.deviceIndex, input_channels=1, input_format=self.typeSettings.paFormat, output_device=None, output_channels=None, output_format=None ):
			self.typeSettings = Float32Settings()
		if not self.audioInterface.is_format_supported( sampleRate, input_device = self.deviceIndex, input_channels=1, input_format=self.typeSettings.paFormat, output_device=None, output_channels=None, output_format=None ):
			raise Exception( 'Could not find a compatible device format for {0}'.format( self.audioInterface.get_device_info_by_index( self.deviceIndex )[ 'name' ] ) )
			
		self.rollingPowerWindowEvaluator = RollingPowerWindowEvaluator( audioBufferLength, smoothingWindowHalfSize, self.typeSettings )
		
		self.audioInputStream = audioInterface.open( 
			input_device_index = self.deviceIndex,
			format = self.typeSettings.paFormat,
			channels = 1,
			rate = sampleRate,
			input = True,
			frames_per_buffer = self.audioInputFrameCount,
			stream_callback = lambda data, frameCount, timeInfo, statusFlags: self.receiveData( data, frameCount, timeInfo, statusFlags ) )
	
	def start( self ):
		if self.audioInputStream is None:
			self.initializeStream()
{'fraction_non_alphanumeric': 0.0846956864289935, 'fraction_numerical': 0.011555380474033221, 'mean_word_length': 4.7675123059447175, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3767231', 'n_tokens_mistral': 4953, 'n_tokens_neox': 4337, 'n_words': 1258}