Skip to content

Commit

Permalink
Fix types
Browse files Browse the repository at this point in the history
  • Loading branch information
rhelmot committed Dec 6, 2023
1 parent 3e5856c commit 8b91a1f
Show file tree
Hide file tree
Showing 3 changed files with 29 additions and 28 deletions.
2 changes: 1 addition & 1 deletion sound/asyncplayer.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ def _play(self, note):
end = self.frame + int(note.duration)
bisect.insort(self.playing, (-end, self.frame, note))

def play(self, note):
def play(self, note): # type: ignore[reportIncompatibleMethodOverride]
"""
:param note: A Signal object to play, starting now.
"""
Expand Down
12 changes: 6 additions & 6 deletions sound/envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,12 @@ def __init__(self, duration):
self.duration = duration * SAMPLE_RATE
self.pure = True

def amplitude(self, frame):
def amplitude(self, frame: int) -> float:
if frame < 0:
return 0
return 0.
if frame < self.duration:
return 1
return 0
return 1.
return 0.

def apply_adsr(self, attack, decay, release, attack_level=1, sustain_level=0.5):
"""
Expand Down Expand Up @@ -71,7 +71,7 @@ def __init__(self, attack, decay, sustain, release, attack_level: Union[int, flo

def amplitude(self, frame):
if frame < 0:
return 0
return 0.
if frame < self.attack:
return float(frame) / self.attack * self.attack_level
frame -= self.attack
Expand All @@ -83,7 +83,7 @@ def amplitude(self, frame):
frame -= self.sustain
if frame < self.release:
return float(frame) / self.release * -self.sustain_level + self.sustain_level
return 0
return 0.

class Decay(Envelope):
"""
Expand Down
43 changes: 22 additions & 21 deletions sound/signal.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# pylint: disable=superfluous-parens
from typing import Optional
import numpy
import struct
import wave
Expand Down Expand Up @@ -41,7 +42,7 @@ class Signal(object):
"""
# pylint: disable=unused-argument,no-self-use

def play(self, length=None, progress=False):
def play(self, length: Optional[float]=None, progress=False):
"""
Play this signal. Block until playback is complete.
If the given signal is infinitely long, default to three seconds of playback.
Expand Down Expand Up @@ -121,47 +122,47 @@ def render(self, length=None, progress=False, clip_warn=True):
print('Warning: clipping! max val %s' % clipped)
return out

def amplitude(self, frame):
def amplitude(self, frame: int) -> float:
"""
The main interface for accessing sample data. This is the primary method that should
be overridden by subclasses.
:param frame: The frame whose amplitude should be returned.
"""
return 0
return 0.

duration = 0
pure = True

def __add__(self, other):
def __add__(self, other) -> "Signal":
if other == 0:
return self
if type(other) is DelaySignal:
return SequenceSignal((self, 0), (other.src, other.delay))
return MixSignal(self, ConstantSignal.wrap(other))

def __radd__(self, other):
def __radd__(self, other) -> "Signal":
return self + other

def __sub__(self, other):
def __sub__(self, other) -> "Signal":
if other == 0:
return self
return self + -ConstantSignal.wrap(other)

def __rsub__(self, other):
def __rsub__(self, other) -> "Signal":
return -self + ConstantSignal.wrap(other)

def __mul__(self, other):
def __mul__(self, other) -> "Signal":
if other == 1:
return self
if other == 0 or (type(other) == ConstantSignal and other._amplitude == 0):
return ConstantSignal(0)
return EnvelopeSignal(self, ConstantSignal.wrap(other))

def __rmul__(self, other):
def __rmul__(self, other) -> "Signal":
return self * other

def __div__(self, other):
def __div__(self, other) -> "Signal":
if type(other) not in numty:
raise TypeError("Can't divide by %s" % repr(other))
return self * (1./other)
Expand All @@ -172,21 +173,21 @@ def __div__(self, other):
def __rdiv__(self, other):
raise TypeError("Can't divide by %s" % repr(self))

def __neg__(self):
def __neg__(self) -> "Signal":
return InvertSignal(self)

def __rshift__(self, other):
def __rshift__(self, other) -> "Signal":
if type(other) not in numty:
raise TypeError("Can't shift by %s" % repr(other))
return DelaySignal(self, int(other*SAMPLE_RATE))

def __lshift__(self, other):
def __lshift__(self, other) -> "Signal":
return self >> -other

def __and__(self, other):
def __and__(self, other) -> "Signal":
return self + (other >> (float(self.duration) / SAMPLE_RATE))

def __mod__(self, other):
def __mod__(self, other) -> "Signal":
if type(other) not in numty:
raise TypeError("Can't loop by %s" % repr(other))
return LoopSignal(self, other)
Expand All @@ -201,7 +202,7 @@ def __getitem__(self, key):
else:
raise KeyError(key)

def purify(self, preprocess=False):
def purify(self, preprocess=False) -> "Signal":
"""
Return a pure version of this signal. This is a no-op for pure signals, but for
impure signals it installs a caching layer on top of the signal.
Expand All @@ -213,7 +214,7 @@ def purify(self, preprocess=False):
return self
return Purifier(self, preprocess=preprocess)

def reverse(self):
def reverse(self) -> "Signal":
"""
Return a reversed version of this signal.
"""
Expand Down Expand Up @@ -387,19 +388,19 @@ def __init__(self, src, length=None, preprocess=False):
length = int(length * SAMPLE_RATE)
self.nextf = 0
self.duration = length
self.storage = [None]*(100000 if self.duration == float('inf') else int(self.duration))
self.storage = [0.]*(100000 if self.duration == float('inf') else int(self.duration))
self.pure = True
self.src = src

if preprocess:
self.amplitude(self.duration - 1)

def amplitude(self, frame):
if frame < 0: return 0
if frame >= self.duration: return 0
if frame < 0: return 0.
if frame >= self.duration: return 0.
while frame >= self.nextf:
if self.nextf >= len(self.storage):
self.storage += [None]*100000
self.storage += [0.]*100000
self.storage[self.nextf] = self.src.amplitude(self.nextf)
self.nextf += 1
return self.storage[frame]
Expand Down

0 comments on commit 8b91a1f

Please sign in to comment.