diff --git a/csoundengine/csoundlib.py b/csoundengine/csoundlib.py index ec46e2a..9a96768 100755 --- a/csoundengine/csoundlib.py +++ b/csoundengine/csoundlib.py @@ -37,6 +37,7 @@ import emlib.textlib import emlib.dialogs import emlib.mathlib +from emlib.common import runonce import numpy as np from typing import TYPE_CHECKING @@ -408,6 +409,7 @@ def __init__(self, kind='callback'): def getSystemSr(self) -> int | None: if sys.platform == 'linux' and linuxaudio.isPipewireRunning(): info = linuxaudio.pipewireInfo() + assert info is not None return info.sr return super().getSystemSr() @@ -507,7 +509,7 @@ def nextpow2(n:int) -> int: return int(2 ** _math.ceil(_math.log(n, 2))) -@emlib.misc.runonce +@runonce def findCsound() -> str | None: """ Find the csound binary or None if not found @@ -735,14 +737,14 @@ def userPluginsFolder(float64=True, apiversion='6.0') -> str: def runCsd(csdfile:str, - outdev="", - indev="", - backend="", + outdev='', + indev='', + backend='', nodisplay=False, nomessages=False, comment='', piped=False, - extra: list[str] = None + extra: list[str] | None = None ) -> _subprocess.Popen: """ Run the given .csd as a csound subprocess @@ -791,14 +793,15 @@ def runCsd(csdfile:str, if nomessages: args.append('-m16') if comment and offline: - args.append(f'-+id_comment="{comment}"') + args.append(f'-+id_comment="{comment}"' + ) if extra: args.extend(extra) args.append(csdfile) return csoundSubproc(args, piped=piped) -def joinCsd(orc: str, sco="", options: list[str] | None = None) -> str: +def joinCsd(orc: str, sco='', options: list[str] | None = None) -> str: """ Joins an orc and a score (both as str), returns a csd as string @@ -1044,6 +1047,7 @@ def _csoundGetInfoViaAPI(opcodedir='') -> dict: if opcodedir: cs.setOption(f'--opcode-dir={opcodedir}') opcodes, n = cs.newOpcodeList() + assert opcodes is not None opcodeNames = [opc.opname.decode('utf-8') for opc in opcodes] cs.disposeOpcodeList(opcodes) version = cs.version() @@ -1079,7 +1083,7 @@ def _opcodesList(opcodedir='') -> list[str]: def saveAsGen23(data: Sequence[float] | np.ndarray, outfile: str, fmt="%.12f", - header="" + header='' ) -> None: """ Saves the data to a gen23 table @@ -2428,13 +2432,14 @@ def _writeCsd(self, stream, datadir='') -> None: def run(self, output: str, - csdfile: str = None, - inputdev: str = None, - backend: str = None, + csdfile='', + inputdev='', + backend='', suppressdisplay=True, nomessages=False, piped=False, - extraOptions: list[str] = None) -> _subprocess.Popen: + extraOptions: list[str] | None = None + ) -> _subprocess.Popen: """ Run this csd. @@ -2552,9 +2557,9 @@ def mincer(sndfile:str, ts = np.arange(t0, t1+dt, dt) fmt = "%.12f" _, time_gen23 = _tempfile.mkstemp(prefix='time-', suffix='.gen23') - np.savetxt(time_gen23, timebpf.map(ts), fmt=fmt, header=str(dt), comments="") + np.savetxt(time_gen23, timebpf.map(ts), fmt=fmt, header=str(dt), comments='') _, pitch_gen23 = _tempfile.mkstemp(prefix='pitch-', suffix='.gen23') - np.savetxt(pitch_gen23, pitchbpf.map(ts), fmt=fmt, header=str(dt), comments="") + np.savetxt(pitch_gen23, pitchbpf.map(ts), fmt=fmt, header=str(dt), comments='') csd = f""" @@ -2645,7 +2650,7 @@ def _instr_as_orc(instrid, body, initstr, sr, ksmps, nchnls): return orc -def recInstr(body:str, events:list, init="", outfile:str=None, +def recInstr(body: str, events: list, init='', outfile='', sr=44100, ksmps=64, nchnls=2, a4=442, samplefmt='float', dur=None ) -> tuple[str, _subprocess.Popen]: @@ -3285,7 +3290,7 @@ def splitDocstring(body: str | list[str]) -> tuple[str, str]: rest = '\n'.join(lines[docend:]) else: docstring = '' - rest = body + rest = body if isinstance(body, str) else '\n'.join(lines) return docstring, rest @@ -3323,11 +3328,11 @@ def instrParseBody(body: str) -> ParsedInstrBody: pfieldsNameToIndex={'ibus': 4, 'kfreq': 5}) """ if not body.strip(): - return ParsedInstrBody(pfieldIndexToValue=EMPTYDICT, + return ParsedInstrBody(pfieldIndexToValue={}, pfieldLines=(), body='', lines=(), - pfieldIndexToName=EMPTYDICT) + pfieldIndexToName={}) pfieldLines = [] bodyLines = [] @@ -3412,7 +3417,7 @@ def instrParseBody(body: str) -> ParsedInstrBody: lines=lines) -def bestSampleEncodingForExtension(ext: str) -> str | None: +def bestSampleEncodingForExtension(ext: str) -> str: """ Given an extension, return the best sample encoding. @@ -3541,9 +3546,10 @@ def _soundfontInstrumentsAndPresets(sfpath: str sf = Sf2File(f) instruments: list[tuple[int, str]] = [(num, instr.name.strip()) for num, instr in enumerate(sf.instruments) - if instr.name != 'EOI'] + if instr.name and instr.name != 'EOI'] presets: list[tuple[int, int, str]] = [(p.bank, p.preset, p.name.strip()) - for p in sf.presets if p.name != 'EOP'] + for p in sf.presets + if p.name and p.name != 'EOP'] presets.sort() return instruments, presets diff --git a/csoundengine/engine.py b/csoundengine/engine.py index bb5330a..c42b58d 100755 --- a/csoundengine/engine.py +++ b/csoundengine/engine.py @@ -335,8 +335,8 @@ def __init__(self, globalcode: str = "", numAudioBuses: int | None = None, numControlBuses: int | None = None, - quiet: bool = None, - udpserver: bool = None, + quiet: bool | None = None, + udpserver: bool | None = None, udpport: int = 0, commandlineOptions: list[str] | None = None, includes: list[str] | None = None, @@ -370,8 +370,10 @@ def __init__(self, f"{availableBackends}") cascadingBackends = [b.strip() for b in backend.split(",")] - resolvedBackend = internal.resolveOption(cascadingBackends, - availableBackends) + resolvedBackend = internal.resolveOption(cascadingBackends, availableBackends) + if resolvedBackend is None: + raise RuntimeError(f"No audio backends available") + backendDef = csoundlib.getAudioBackend(resolvedBackend) if not backendDef: @@ -403,7 +405,7 @@ def __init__(self, nchnls = selected.numchannels elif isinstance(outdev, int) or _re.search(r"\bdac[0-9]+\b", outdev): # dac1, dac8 - if resolvedBackend == 'jack': + if backendDef.name == 'jack': logger.warning( "This way of setting the audio device is discouraged with jack" ". Use a regex to select a specific client or None to connect" @@ -411,7 +413,7 @@ def __init__(self, if isinstance(outdev, int): outdev = f"dac{outdev}" else: - if resolvedBackend == 'jack': + if backendDef.name == 'jack': outdevName = outdev else: selected = next((d for d in outdevs if _fnmatch.fnmatch(d.name, outdev)), None) @@ -471,10 +473,12 @@ def __init__(self, logger.error(f"Asked for system sr, but backend '{resolvedBackend}', does not" f"have a fixed sr. Using sr={sr}") - if a4 is None: a4 = cfg['A4'] + if a4 is None: + a4 = cfg['A4'] if numthreads == 0: numthreads = config['numthreads'] - if ksmps is None: ksmps = cfg['ksmps'] + if ksmps is None: + ksmps = cfg['ksmps'] if nchnls_i is None: nchnls_i = cfg['nchnls_i'] if nchnls is None: @@ -484,30 +488,32 @@ def __init__(self, outpattern=outdev, inpattern=indev) nchnls = nchnls or outchnls nchnls_i = nchnls_i or inchnls - assert nchnls > 0 - assert nchnls_i >= 0 + assert nchnls is not None and nchnls > 0 + assert nchnls_i is not None and nchnls_i >= 0 + + if quiet is None: + quiet = cfg['suppress_output'] - if quiet is None: quiet = cfg['suppress_output'] if quiet: commandlineOptions.append('-m0') commandlineOptions.append('-d') self.name = name "Name of this Engine" - assert sr > 0 - self.sr = sr + assert sr is not None and sr > 0 + self.sr: int = sr "Sample rate" self.backend = resolvedBackend "Name of the backend used (jack, portaudio, etc)" - self.a4 = a4 + self.a4: int | float = a4 "Reference frequency for A4" - self.ksmps = ksmps + self.ksmps: int = ksmps "Number of samples per cycle" - self.onecycle = ksmps / sr + self.onecycle: float = ksmps / sr "Duration of one performance cycle (ksmps/sr)" self.outdev = outdev @@ -522,19 +528,19 @@ def __init__(self, self.indevName = indevName "Long name of the input device" - self.nchnls = nchnls + self.nchnls: int = nchnls "Number of output channels" - self.nchnls_i = nchnls_i + self.nchnls_i: int = nchnls_i "Number of input channels" - self.globalCode = globalcode + self.globalCode: str = globalcode "Global (init) code to execute at the start of the Engine" self.started = False "Is this engine started?" - self.extraOptions = commandlineOptions + self.extraOptions: list[str] = commandlineOptions "Extra options passed to csound" self.commandlineOptions: list[str] = [] @@ -543,13 +549,13 @@ def __init__(self, self.includes: list[str] = includes if includes is not None else [] "List of include files" - self.extraLatency = latency if latency is not None else config['sched_latency'] + self.extraLatency: float = latency if latency is not None else config['sched_latency'] "Added latency for better synch" - self.numAudioBuses = numAudioBuses if numAudioBuses is not None else config['num_audio_buses'] + self.numAudioBuses: int = numAudioBuses if numAudioBuses is not None else config['num_audio_buses'] "Number of audio buses" - self.numControlBuses = numControlBuses if numControlBuses is not None else config['num_control_buses'] + self.numControlBuses: int = numControlBuses if numControlBuses is not None else config['num_control_buses'] "Number of control buses" self.udpPort = 0 @@ -558,20 +564,20 @@ def __init__(self, self.csound: None | ctcsound.Csound = None "The csound object" - self.autosync = autosync + self.autosync: bool = autosync """If True, call .sync whenever is needed""" backendBufferSize, backendNumBuffers = backendDef.bufferSizeAndNum() buffersize = (buffersize or backendBufferSize or config['buffersize'] or 256) - buffersize = max(ksmps * 2, buffersize) + buffersize = max(self.ksmps * 2, buffersize) numbuffers = (numbuffers or backendNumBuffers or config['numbuffers'] or internal.determineNumbuffers(self.backend or "portaudio", buffersize=buffersize)) - self.bufferSize = buffersize + self.bufferSize: int = buffersize "Buffer size" - self.numBuffers = numbuffers + self.numBuffers: int = numbuffers "Number of buffers to fill" self.midiBackend: None | str = midibackend @@ -580,7 +586,7 @@ def __init__(self, self.started = False """Has this engine already started?""" - self.numthreads = numthreads + self.numthreads: int = numthreads """Number of threads to use in performance (corresponds to csound -j N)""" self._builtinInstrs: dict[str, int] = {} @@ -767,7 +773,7 @@ def _getSyncToken(self) -> int: self._responsesTable[token] = _UNSET return token - def _waitOnToken(self, token: int, sleepfunc=time.sleep, period=0.001, timeout: float = None + def _waitOnToken(self, token: int, sleepfunc=time.sleep, period=0.001, timeout: float | None = None ) -> float | None: if timeout is None: timeout = config['timeout'] @@ -960,10 +966,10 @@ def stop(self): """ if not hasattr(self, "name"): return - logger.info(f"stopping Engine {self.name}") - if not self.started or self._exited: + if self.csound is None or not self.started or self._exited: logger.debug(f"Engine {self.name} was not running, so can't stop it") return + logger.info(f"stopping Engine {self.name}") logger.info("... stopping thread") self._perfThread.stop() time.sleep(0.1) @@ -1095,7 +1101,7 @@ def controlLatency(self) -> float: """ return self.ksmps/self.sr * 2 - def sync(self, timeout: float = None, force=False, threshold=2.) -> bool: + def sync(self, timeout: float | None = None, force=False, threshold=2.) -> bool: """ Block until csound has processed its immediate events @@ -1459,6 +1465,7 @@ def lockClock(self, lock=True): if self.isClockLocked(): logger.debug("The elapsed time clock is already locked") else: + assert self.csound is not None self._lockedElapsedTime = self.csound.currentTimeSamples()/self.sr else: if not self._lockedElapsedTime: @@ -1878,9 +1885,9 @@ def unschedAll(self) -> None: self._setupGlobalInstrs() def session(self, - priorities: int = None, - maxControlsPerInstr: int = None, - numControlSlots: int = None + priorities: int | None = None, + maxControlsPerInstr: int | None = None, + numControlSlots: int | None = None ) -> _session.Session: """ Return the Session corresponding to this Engine @@ -2182,11 +2189,12 @@ def callLater(self, delay: float, callback: Callable) -> None: pargs = [self._builtinInstrs['pingback'], delay, 0.01, token] self._eventWithCallback(token, pargs, lambda token: callback()) - def _eventWait(self, token: int, pargs: Sequence[float], timeout: float = None + def _eventWait(self, token: int, pargs: Sequence[float], timeout: float | None = None ) -> float | None: if timeout is None: timeout = config['timeout'] - assert timeout > 0 + else: + assert timeout > 0 q = self._registerSync(token) self._perfThread.scoreEvent(0, "i", pargs) try: @@ -2200,7 +2208,7 @@ def plotTableSpectrogram(self, tabnum: int, fftsize=2048, mindb=-90, - maxfreq: int = None, + maxfreq: int | None = None, overlap: int = 4, minfreq: int = 0, sr: int = 44100, @@ -2597,7 +2605,7 @@ def channelPointer(self, channel: str, kind='control', mode='rw') -> np.ndarray: return ptr def setChannel(self, channel: str, value: float | str | np.ndarray, - method: str = None, delay=0. + method='', delay=0. ) -> None: """ Set the value of a software channel @@ -2607,7 +2615,7 @@ def setChannel(self, channel: str, value: float | str | np.ndarray, value: the new value, should match the type of the channel (a float for a control channel, a string for a string channel or a numpy array for an audio channel) - method: one of ``'api'``, ``'score'``, ``'udp'``. None will choose the most appropriate + method: one of ``'api'``, ``'score'``, ``'udp'``. An empty str will choose the most appropriate method for the current engine/args delay: a delay to set the channel @@ -2633,7 +2641,7 @@ def setChannel(self, channel: str, value: float | str | np.ndarray, method = "api" elif delay > 0: method = "score" - elif method is None: + elif not method: if self.udpPort and config['prefer_udp']: method = 'udp' else: @@ -2905,7 +2913,7 @@ def includeFile(self, include: str) -> None: return self.includes.append(abspath) - def readSoundfile(self, path="?", tabnum: int = None, chan=0, + def readSoundfile(self, path="?", tabnum: int | None = None, chan=0, callback=None, block=False, skiptime=0.) -> int: """ Read a soundfile into a table (via GEN1), returns the table number @@ -2973,7 +2981,7 @@ def readSoundfile(self, path="?", tabnum: int = None, chan=0, return tabnum def soundfontPlay(self, index: int, pitch: float, amp=0.7, delay=0., - dur=-1., vel: int = None, chan=1 + dur=-1., vel: int | None = None, chan=1 ) -> float: """ Play a note of a previously loaded soundfont @@ -3035,14 +3043,15 @@ def soundfontPlay(self, index: int, pitch: float, amp=0.7, delay=0., """ assert index in self._soundfontPresets.values() if vel is None: - vel = amp/127 + vel = amp*127 args = [pitch, amp, index, vel, chan] return self.sched(self._builtinInstrs['soundfontPlay'], delay=delay, dur=dur, args=args) def soundfontPreparePreset(self, sf2path: str, - preset: tuple[int, int] = None) -> int: + preset: tuple[int, int] | None = None + ) -> int: """ Prepare a soundfont's preset to be used @@ -3091,7 +3100,7 @@ def soundfontPreparePreset(self, def _readSoundfileAsync(self, path: str, - tabnum: int = None, + tabnum: int | None = None, chan=0) -> int: assert self.started if tabnum is None: @@ -3734,7 +3743,7 @@ def automateBus(self, self.automateBus(bus=bus, pairs=subgroup, delay=delay+subdelay, mode=mode, overtake=overtake) - def readBus(self, bus: int, default=0.) -> float: + def readBus(self, bus: int, default: float | None = None) -> float | None: """ Read the current value of a control bus @@ -3861,7 +3870,7 @@ def _dumpbus(self, bus: int): pargs = [self._builtinInstrs['busdump'], 0, 0, int(bus)] self._perfThread.scoreEvent(0, "i", pargs) - def assignBus(self, kind='', value: float = None, persist=False + def assignBus(self, kind='', value: float | None = None, persist=False ) -> int: """ Assign one audio/control bus, returns the bus number. diff --git a/csoundengine/event.py b/csoundengine/event.py index 2de70df..478e776 100644 --- a/csoundengine/event.py +++ b/csoundengine/event.py @@ -70,9 +70,8 @@ def automate(self, param: str, pairs: Sequence[float], delay=0., interpolation=interpolation, overtake=overtake) self.automations.append(autom) - def set(self, param: str = '', value: float = None, delay=0., **kws): + def set(self, param='', value=0., delay=0., **kws): if param: - assert value is not None self.automate(param=param, pairs=(0, value), delay=delay) if kws: for param, value in kws.items(): diff --git a/csoundengine/instr.py b/csoundengine/instr.py index 98c4613..4c01286 100644 --- a/csoundengine/instr.py +++ b/csoundengine/instr.py @@ -230,14 +230,14 @@ def __init__(self, name: str, body: str, args: dict[str, float | str] | None = None, - init: str = '', - numchans: int = 1, + init='', + numchans=1, preschedCallback=None, - doc: str = '', + doc='', includes: list[str] | None = None, - aliases: dict[str, str] = None, - maxNamedArgs: int = 0, - useDynamicPfields: bool = None + aliases: dict[str, str] | None = None, + maxNamedArgs=0, + useDynamicPfields: bool | None = None ) -> None: assert isinstance(name, str) @@ -692,7 +692,7 @@ def paramValue(self, param: str) -> float | str | None: defaults = self.paramDefaultValues(aliased=True) if param2 not in defaults: raise KeyError(f"Unknown parameter '{param}'. " - f"Possible parameters: {default.keys()}") + f"Possible parameters: {defaults.keys()}") return defaults[param2] @cache @@ -932,6 +932,8 @@ def pfieldsTranslate(self, if kws: kwsindexes = [k if isinstance(k, int) else self.pfieldIndex(k) for k in kws] maxidx = max(maxidx, max(kwsindexes) - 5) + else: + kwsindexes = [] numpfields = maxidx + 1 if not args: @@ -960,12 +962,12 @@ def pfieldsTranslate(self, def rec(self, dur: float, - outfile: str | None = None, + outfile='', args: list[float] | dict[str, float] | None = None, sr: int | None = None, ksmps: int | None = None, encoding: str | None = None, - nchnls: int = 2, + nchnls=2, wait=True, a4: int | None = None, delay=0., diff --git a/csoundengine/offline.py b/csoundengine/offline.py index 4bd2677..f450fa9 100644 --- a/csoundengine/offline.py +++ b/csoundengine/offline.py @@ -233,15 +233,17 @@ def __init__(self, nchnls: int = 2, ksmps: int | None = None, a4: float | None = None, - priorities: int = None, + priorities: int | None = None, numAudioBuses=1000, numControlBuses=10000, dynamicArgsPerInstr: int = 16, - dynamicArgsSlots: int = None): + dynamicArgsSlots: int | None = None): if priorities is None: priorities = config['session_priorities'] + assert isinstance(priorities, int) + self.sr = sr or config['rec_sr'] """Samplerate""" @@ -276,7 +278,7 @@ def __init__(self, self._idCounter = 0 self._nameAndPriorityToInstrnum: dict[tuple[str, int], int] = {} self._instrnumToNameAndPriority: dict[int, tuple[str, int]] = {} - self._numbuckets = priorities + self._numbuckets: int = priorities self._bucketCounters = [0] * priorities self._startUserInstrs = 50 self._instanceCounters: dict[int, int] = {} @@ -516,13 +518,13 @@ def registerInstr(self, instr: Instr) -> bool: def defInstr(self, name: str, body: str, - args: dict[str, float|str] = None, + args: dict[str, float|str] | None = None, init: str = '', - priority: int = None, + priority: int | None = None, doc: str = '', includes: list[str] | None = None, - aliases: dict[str, str] = None, - useDynamicPfields: bool = None, + aliases: dict[str, str] | None = None, + useDynamicPfields: bool | None = None, **kws) -> Instr: """ Create an :class:`~csoundengine.instr.Instr` and register it with this renderer @@ -657,7 +659,7 @@ def sched(self, dur=-1., priority=1, args: Sequence[float | str] | dict[str, float] | None = None, - whenfinished: Callable = None, + whenfinished: Callable | None = None, relative=True, **kwargs ) -> SchedEvent: @@ -975,12 +977,12 @@ def render(self, verbose: bool | None = None, openWhenDone=False, starttime=0., - compressionBitrate: int = None, - sr: int = None, - ksmps: int = None, + compressionBitrate: int | None = None, + sr: int | None = None, + ksmps: int | None = None, tail=0., numthreads=0, - csoundoptions: list[str] = None + csoundoptions: list[str] | None = None ) -> RenderJob: """ Render to a soundfile @@ -1194,7 +1196,7 @@ def getEventsByP1(self, p1: float) -> list[SchedEvent]: """ return [ev for ev in self.scheduledEvents.values() if ev.p1 == p1] - def strSet(self, s: str, index: int = None) -> int: + def strSet(self, s: str, index: int | None = None) -> int: """ Set a string in this renderer. diff --git a/csoundengine/session.py b/csoundengine/session.py index 176f722..1ab153f 100644 --- a/csoundengine/session.py +++ b/csoundengine/session.py @@ -249,7 +249,7 @@ class _RenderingSessionHandler(SessionHandler): def __init__(self, renderer: offline.Renderer): self.renderer = renderer - def schedEvent(self, event: csoundengine.event.Event): + def schedEvent(self, event: Event): return self.renderer.schedEvent(event) def makeTable(self, @@ -257,7 +257,7 @@ def makeTable(self, size: int | tuple[int, int] = 0, sr: int = 0, ) -> TableProxy: - self.renderer.makeTable(data=data, size=size, sr=sr) + return self.renderer.makeTable(data=data, size=size, sr=sr) def readSoundfile(self, path: str, @@ -266,7 +266,7 @@ def readSoundfile(self, delay=0., force=False, ) -> TableProxy: - self.renderer.readSoundfile(path) + return self.renderer.readSoundfile(path) class Session(AbstractRenderer): @@ -350,9 +350,9 @@ class Session(AbstractRenderer): """ def __new__(cls, engine: str | Engine | None = None, - priorities: int = None, - dynamicArgsPerInstr: int = None, - dynamicArgsSlots: int = None, + priorities: int | None = None, + dynamicArgsPerInstr: int | None = None, + dynamicArgsSlots: int | None = None, **enginekws): if isinstance(engine, str): @@ -367,9 +367,9 @@ def __new__(cls, def __init__(self, engine: str | Engine | None = None, - priorities: int = None, - maxControlsPerInstr: int = None, - numControlSlots: int = None, + priorities: int | None = None, + maxControlsPerInstr: int | None = None, + numControlSlots: int | None = None, **enginekws ) -> None: """ @@ -782,13 +782,13 @@ def setHandler(self, handler: SessionHandler | None def defInstr(self, name: str, body: str, - args: dict[str, float|str] = None, - init: str = '', - priority: int = None, - doc: str = '', + args: dict[str, float|str] | None = None, + init='', + priority: int | None = None, + doc='', includes: list[str] | None = None, - aliases: dict[str, str] = None, - useDynamicPfields: bool = None, + aliases: dict[str, str] | None = None, + useDynamicPfields: bool | None = None, **kws) -> Instr: """ Create an :class:`~csoundengine.instr.Instr` and register it at this session @@ -986,6 +986,8 @@ def getInstr(self, instrname: str) -> Instr | None: """ if instrname == "?": instrname = _dialogs.selectItem(list(self.instrs.keys())) + if instrname is None: + return None return self.instrs.get(instrname) def _getReifiedInstr(self, name: str, priority: int) -> _ReifiedInstr | None: @@ -1073,7 +1075,7 @@ def instrnum(self, instrname: str, priority: int = 1) -> int: rinstr, needssync = self.prepareSched(instrname, priority) return rinstr.instrnum - def assignBus(self, kind='', value: float = None, persist=False + def assignBus(self, kind='', value: float | None = None, persist=False ) -> busproxy.Bus: """ Creates a bus in the engine @@ -1232,7 +1234,7 @@ def schedEvent(self, event: Event) -> Synth: for automation in event.automations: synth.automate(param=automation.param, pairs=automation.pairs, - delay=automation.delay, + delay=automation.delay or 0., mode=automation.interpolation, overtake=automation.overtake) return synth @@ -1260,16 +1262,17 @@ def __exit__(self, exc_type, exc_val, exc_tb): self._lockedLatency = None def rendering(self, - outfile: str = '', - sr: int | None = None, + outfile='', + sr=0, nchnls: int | None = None, - ksmps: int | None = None, + ksmps=0, encoding='', starttime=0., endtime=0., tail=0., openWhenDone=False, - verbose: bool = None) -> Renderer: + verbose: bool | None = None + ) -> offline.Renderer: """ A context-manager for offline rendering @@ -1296,8 +1299,6 @@ def rendering(self, the rendering. tail: extra render time at the end, to accomodate extended releases openWhenDone: open the file in the default application after rendering. - redirect: if True, within the context any call to .sched will be - redirected to the offline Renderer verbose: if True, output rendering information. If None uses the value specified in the config (``config['rec_suppress_output']``) @@ -1330,7 +1331,7 @@ def rendering(self, handler = _RenderingSessionHandler(renderer=renderer) self.setHandler(handler) - def atexit(r: Renderer, _outfile=outfile): + def atexit(r: offline.Renderer, _outfile=outfile): r.render(outfile=_outfile, endtime=endtime, encoding=encoding, starttime=starttime, openWhenDone=openWhenDone, tail=tail, verbose=verbose) @@ -1513,9 +1514,11 @@ def sched(self, abstime = delay if not relative else self.engine.elapsedTime() + delay + self.engine.extraLatency if instrname == "?": - instrname = _dialogs.selectItem(list(self.instrs.keys()), - title="Select Instr", - ensureSelection=True) + selected = _dialogs.selectItem(list(self.instrs.keys()), + title="Select Instr", + ensureSelection=True) + assert selected is not None + instrname = selected instr = self.getInstr(instrname) if instr is None: @@ -1645,7 +1648,7 @@ def unsched(self, event: int | float | SchedEvent, delay=0.) -> None: event: the event to stop, either a Synth or the p1 delay: how long to wait before stopping them """ - if self.isRendering() and self._blockWhileRendering: + if self.isRendering(): raise RuntimeError("This Session is blocked while in rendering mode. Call .unsched on the renderer instead") synthid = event if isinstance(event, (int, float)) else event.p1 @@ -1742,7 +1745,7 @@ def readSoundfile(self, >>> session.playSample(table) """ - if self.isRendering() and self._blockWhileRendering: + if self.isRendering(): raise RuntimeError("This Session is blocked during rendering. Call .readSoundFile on the offline " "renderer instead") if path == "?": @@ -1801,7 +1804,7 @@ def makeTable(self, a TableProxy object """ - if self.isRendering() and self._blockWhileRendering: + if self.isRendering(): raise RuntimeError("This Session is in rendering mode. Call .makeTable on the renderer instead " "(with session.rendering() as r: ... r.makeTable(...)") @@ -1976,7 +1979,7 @@ def playPartials(self, >>> session.playPartials(source='packed.mtx', speed=0.5) """ - if self.isRendering() and self._blockWhileRendering: + if self.isRendering(): raise RuntimeError("This Session is blocked during rendering") iskip, inumrows, inumcols = -1, 0, 0 @@ -2086,7 +2089,7 @@ def playSample(self, A Synth with the following mutable parameters: kgain, kspeed, kchan, kpan """ - if self.isRendering() and self._blockWhileRendering: + if self.isRendering(): raise RuntimeError("This Session is in rendering mode. Call .playSample on the renderer instead") if isinstance(source, int): @@ -2127,8 +2130,8 @@ def playSample(self, kgain=gain, ixfade=crossfade)) - def makeRenderer(self, sr: int = None, nchnls: int = None, ksmps: int = None, - ) -> Renderer: + def makeRenderer(self, sr=0, nchnls: int | None = None, ksmps=0, + ) -> offline.Renderer: """ Create a :class:`~csoundengine.offline.Renderer` (to render offline) with the instruments defined in this Session @@ -2138,7 +2141,8 @@ def makeRenderer(self, sr: int = None, nchnls: int = None, ksmps: int = None, Args: sr: the samplerate (see config['rec_sr']) - ksmps: ksmps used for rendering (see also config['rec_ksmps']) + ksmps: ksmps used for rendering (see also config['rec_ksmps']). 0 uses + the default defined in the config nchnls: the number of output channels. If not given, nchnls is taken from the session diff --git a/csoundengine/synth.py b/csoundengine/synth.py index 61edefd..a8a4b6a 100644 --- a/csoundengine/synth.py +++ b/csoundengine/synth.py @@ -339,7 +339,7 @@ def __repr__(self): if self.instr.hasControls(): ctrlparts = [] for k, v in self.instr.controls.items(): - if k in self.controls: + if self.controls is not None and k in self.controls: v = self.controls[k] ctrlparts.append(f'{k}={v}') parts.append(f"|{' '.join(ctrlparts)}|") @@ -535,7 +535,7 @@ def stop(self, delay=0.) -> None: self.session.unsched(self.p1, delay=delay) -def _synthsCreateHtmlTable(synths: list[Synth], maxrows: int = None, tablestyle='', +def _synthsCreateHtmlTable(synths: list[Synth], maxrows: int | None = None, tablestyle='', ) -> str: synth0 = synths[0] instr0 = synth0.instr @@ -774,7 +774,7 @@ def _automatePfield(self, f"parameters: {self.dynamicParamNames()}") return eventids - def _htmlTable(self, style='', maxrows: int = None) -> str: + def _htmlTable(self, style='', maxrows: int | None = None) -> str: subgroups = _iterlib.classify(self.synths, lambda synth: synth.instr.name) lines = [] instrcol = jupytertools.defaultPalette["name.color"]