Browse Source

Make noisicaa.audioproc.engine mypy clean.

Mostly by adding stubs for various cython modules.
looper
Ben Niemann 5 years ago
parent
commit
29e0b30523
  1. 3
      3rdparty/typeshed/PyQt5/QtCore.pyi
  2. 4
      3rdparty/typeshed/README
  3. 35
      3rdparty/typeshed/posix_ipc.pyi
  4. 2
      NOTES.org
  5. 10
      noisicaa/audioproc/engine/CMakeLists.txt
  6. 52
      noisicaa/audioproc/engine/backend.pyi
  7. 2
      noisicaa/audioproc/engine/backend_test.py
  8. 39
      noisicaa/audioproc/engine/block_context.pyi
  9. 23
      noisicaa/audioproc/engine/buffer_arena.pyi
  10. 2
      noisicaa/audioproc/engine/engine_test.py
  11. 170
      noisicaa/audioproc/engine/graph.py
  12. 27
      noisicaa/audioproc/engine/message_queue.pyi
  13. 29
      noisicaa/audioproc/engine/player.pyi
  14. 52
      noisicaa/audioproc/engine/plugin_host.pyi
  15. 13
      noisicaa/audioproc/engine/plugin_host_process.py
  16. 8
      noisicaa/audioproc/engine/plugin_host_process_test.py
  17. 5
      noisicaa/audioproc/engine/plugin_host_test.py
  18. 30
      noisicaa/audioproc/engine/plugin_ui_host.pyi
  19. 53
      noisicaa/audioproc/engine/processor.pyi
  20. 2
      noisicaa/audioproc/engine/processor_csound_test.py
  21. 86
      noisicaa/audioproc/engine/realm.pyi
  22. 2
      noisicaa/audioproc/engine/realm_test.py
  23. 43
      noisicaa/audioproc/engine/spec.pyi
  24. 2
      noisicaa/ui/instrument_library_test.py
  25. 27
      noisidev/unittest_engine_utils.pyi

3
3rdparty/typeshed/PyQt5/QtCore.pyi vendored

@ -29,6 +29,9 @@ import datetime
# Support for new-style signals and slots.
class pyqtSignal:
def __init__(self, *types, name: str = ...) -> None: ...
def connect(self, slot: typing.Callable) -> None: ...
def disconnect(self, slot: typing.Optional[typing.Callable] = None) -> None: ...
def emit(self, *args: typing.Any) -> None: ...
class pyqtBoundSignal:
def emit(self, *args) -> None: ...

4
3rdparty/typeshed/README vendored

@ -10,3 +10,7 @@ The PyQt5 package, when installed with pip, does not include the stub files, whi
are created when building PyQt5 from source. These files are just copied from a manual
build.
Also tweaked the files a bit, because they don't work with mypy as is...
- posix_ipc.pyi
Manually crafted, with just the subset of class that I actually use (i.e. incomplete).

35
3rdparty/typeshed/posix_ipc.pyi vendored

@ -0,0 +1,35 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Optional
class SharedMemory(object):
def __init__(
self, name: Optional[str], flags: int = 0, mode: int = 0o600, size: int = 0,
read_only: bool = False) -> None: ...
@property
def name(self) -> str: ...
@property
def fd(self) -> int: ...
@property
def size(self) -> int: ...
def close_fd(self) -> None: ...
def unlink(self) -> None: ...

2
NOTES.org

@ -260,7 +260,7 @@ Probably related to unittest.UITestCase
- until grep finds no more files.
* clean up mypy issues in mypy-unclean files :CLEANUP:TESTING:
- grep -r mypy-unclean noisicaa/
- wc -l $(grep -l -r mypy-unclean noisicaa/ | grep -Ev '~$') | sort -nr
- pick some file and clean it up.
- until grep finds no more files.

10
noisicaa/audioproc/engine/CMakeLists.txt

@ -19,15 +19,23 @@
# @end:license
add_python_package(
backend.pyi
backend_test.py
block_context.pyi
buffer_arena.pyi
engine_perftest.py
engine_test.py
graph.py
message_queue.pyi
player.pyi
plugin_host.pyi
plugin_host_process.py
plugin_host_process_test.py
plugin_host_test.py
plugin_host_ladspa_test.py
plugin_host_lv2_test.py
plugin_ui_host.pyi
processor.pyi
processor_test.py
processor_csound_test.py
processor_cvgenerator_test.py
@ -38,7 +46,9 @@ add_python_package(
processor_sample_script_test.py
processor_sound_file_test.py
processor_track_mixer_test.py
realm.pyi
realm_test.py
spec.pyi
)
set(LIB_SRCS

52
noisicaa/audioproc/engine/backend.pyi

@ -0,0 +1,52 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import List, Optional, Union
from noisicaa import host_system as host_system_lib
from . import block_context
from . import buffers
from . import realm as realm_lib
class PyBackendSettings(object):
datastream_address = ... # type: str
time_scale = ... # type: float
def __init__(
self, *, datastream_address: Optional[float] = None, time_scale: Optional[float] = None
) -> None: ...
class PyBackend(object):
def __init__(
self, host_system: host_system_lib.HostSystem, name: Union[str, bytes],
settings: PyBackendSettings) -> None: ...
def setup(self, realm: realm_lib.PyRealm) -> None: ...
def cleanup(self) -> None: ...
def stop(self) -> None: ...
def stopped(self) -> bool: ...
def release(self) -> None: ...
def released(self) -> bool: ...
def send_message(self, msg: bytes) -> None: ...
def begin_block(self, ctxt: block_context.PyBlockContext) -> None: ...
def end_block(self, ctxt: block_context.PyBlockContext) -> None: ...
def output(
self, ctxt: block_context.PyBlockContext, channel: str, samples: List) -> None: ...

2
noisicaa/audioproc/engine/backend_test.py

@ -18,8 +18,6 @@
#
# @end:license
# TODO: mypy-unclean
from noisidev import unittest
from noisidev import unittest_engine_mixins
from noisidev import unittest_engine_utils

39
noisicaa/audioproc/engine/block_context.pyi

@ -0,0 +1,39 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Iterator, Optional
from noisicaa import core
from noisicaa import audioproc
from . import buffer_arena
from . import message_queue
class PyBlockContext(object):
sample_pos = ... # type: int
def __init__(self, buffer_arena: Optional[buffer_arena.PyBufferArena] = None) -> None: ...
def clear_time_map(self) -> None: ...
def append_sample_time(
self, start_time: audioproc.MusicalTime, end_time: audioproc.MusicalTime) -> None: ...
@property
def perf(self) -> core.PerfStats: ...
@property
def out_messages(self) -> Iterator[message_queue.PyMessage]: ...

23
noisicaa/audioproc/engine/buffer_arena.pyi

@ -0,0 +1,23 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
class PyBufferArena(object):
def __init__(self, size: int) -> None: ...

2
noisicaa/audioproc/engine/engine_test.py

@ -20,8 +20,6 @@
#
# @end:license
# TODO: mypy-unclean
import logging
import async_generator

170
noisicaa/audioproc/engine/graph.py

@ -20,22 +20,24 @@
#
# @end:license
# TODO: mypy-unclean
import logging
import os
from typing import Any, Dict, List, Optional, Set # pylint: disable=unused-import
import toposort
from noisicaa import core
from noisicaa.core import ipc # pylint: disable=unused-import
from noisicaa import audioproc
from noisicaa import node_db
from noisicaa import host_system as host_system_lib
from . import control_value
from . import processor
from . import processor as processor_lib
from . import processor_pb2
from . import plugin_host_pb2
from . import buffers
from . import spec as spec_lib
from . import realm as realm_lib
logger = logging.getLogger(__name__)
@ -45,51 +47,51 @@ class GraphError(Exception):
class Port(object):
def __init__(self, *, description):
def __init__(self, *, description: node_db.PortDescription) -> None:
self.__description = description
self.owner = None
self.owner = None # type: Node
def __str__(self):
def __str__(self) -> str:
return '<%s %s:%s>' % (
type(self).__name__,
self.owner.id if self.owner is not None else 'None',
self.name)
@property
def description(self):
def description(self) -> node_db.PortDescription:
return self.__description
@property
def name(self):
def name(self) -> str:
return self.__description.name
@property
def buf_name(self):
def buf_name(self) -> str:
return '%s:%s' % (self.owner.id, self.__description.name)
def get_buf_type(self):
def get_buf_type(self) -> buffers.PyBufferType:
raise NotImplementedError(type(self).__name__)
def set_prop(self, **kwargs):
def set_prop(self, **kwargs: Any) -> None:
assert not kwargs
class InputPortMixin(Port):
# pylint: disable=abstract-method
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.inputs = []
self.inputs = [] # type: List[Port]
def connect(self, port):
def connect(self, port: Port) -> None:
self.check_port(port)
self.inputs.append(port)
def disconnect(self, port):
def disconnect(self, port: Port) -> None:
assert port in self.inputs, port
self.inputs.remove(port)
def check_port(self, port):
def check_port(self, port: Port) -> None:
if not isinstance(port, OutputPortMixin):
raise GraphError("Can only connect to output ports")
@ -97,87 +99,91 @@ class InputPortMixin(Port):
class OutputPortMixin(Port):
# pylint: disable=abstract-method
def __init__(self, *, bypass_port=None, **kwargs):
def __init__(self, *, bypass_port: Optional[str] = None, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._bypass = False
self._bypass_port = bypass_port
@property
def bypass_port(self):
def bypass_port(self) -> Optional[str]:
return self._bypass_port
@property
def bypass(self):
def bypass(self) -> bool:
return self._bypass
@bypass.setter
def bypass(self, value):
def bypass(self, value: bool) -> None:
assert self._bypass_port is not None
self._bypass = bool(value)
def set_prop(self, *, bypass=None, **kwargs): # pylint: disable=arguments-differ
def set_prop( # pylint: disable=arguments-differ
self, *, bypass: Optional[bool] = None, **kwargs: Any
) -> None:
super().set_prop(**kwargs)
if bypass is not None:
self.bypass = bypass
class AudioPortMixin(Port):
def get_buf_type(self):
def get_buf_type(self) -> buffers.PyBufferType:
return buffers.PyFloatAudioBlockBuffer()
class ARateControlPortMixin(Port):
def get_buf_type(self):
def get_buf_type(self) -> buffers.PyBufferType:
return buffers.PyFloatAudioBlockBuffer()
class KRateControlPortMixin(Port):
def get_buf_type(self):
def get_buf_type(self) -> buffers.PyBufferType:
return buffers.PyFloatControlValueBuffer()
class EventPortMixin(Port):
def get_buf_type(self):
def get_buf_type(self) -> buffers.PyBufferType:
return buffers.PyAtomDataBuffer()
class AudioInputPort(AudioPortMixin, InputPortMixin, Port):
def check_port(self, port):
def check_port(self, port: Port) -> None:
super().check_port(port)
if not isinstance(port, AudioOutputPort):
raise GraphError("Can only connect to AudioOutputPort")
class AudioOutputPort(AudioPortMixin, OutputPortMixin, Port):
def __init__(self, *, drywet_port=None, **kwargs):
def __init__(self, *, drywet_port: Optional[str] = None, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._drywet = 0.0
self._drywet_port = drywet_port
@property
def drywet_port(self):
def drywet_port(self) -> str:
return self._drywet_port
@property
def drywet(self):
def drywet(self) -> float:
return self._drywet
@drywet.setter
def drywet(self, value):
def drywet(self, value: float) -> None:
value = float(value)
if value < -100.0 or value > 100.0:
raise ValueError("Invalid dry/wet value.")
self._drywet = float(value)
def set_prop(self, *, drywet=None, **kwargs): # pylint: disable=arguments-differ
def set_prop( # pylint: disable=arguments-differ
self, *, drywet: Optional[float] = None, **kwargs: Any
) -> None:
super().set_prop(**kwargs)
if drywet is not None:
self.drywet = drywet
class ARateControlInputPort(ARateControlPortMixin, InputPortMixin, Port):
def check_port(self, port):
def check_port(self, port: Port) -> None:
super().check_port(port)
if not isinstance(port, ARateControlOutputPort):
raise GraphError("Can only connect to ARateControlOutputPort")
@ -188,7 +194,7 @@ class ARateControlOutputPort(ARateControlPortMixin, OutputPortMixin, Port):
class KRateControlInputPort(KRateControlPortMixin, InputPortMixin, Port):
def check_port(self, port):
def check_port(self, port: Port) -> None:
super().check_port(port)
if not isinstance(port, KRateControlOutputPort):
raise GraphError("Can only connect to KRateControlOutputPort")
@ -199,11 +205,11 @@ class KRateControlOutputPort(KRateControlPortMixin, OutputPortMixin, Port):
class EventInputPort(EventPortMixin, InputPortMixin, Port):
def __init__(self, *, csound_instr='1', **kwargs):
def __init__(self, *, csound_instr: str = '1', **kwargs: Any) -> None:
super().__init__(**kwargs)
self.csound_instr = csound_instr
def check_port(self, port):
def check_port(self, port: Port) -> None:
super().check_port(port)
if not isinstance(port, EventOutputPort):
raise GraphError("Can only connect to EventOutputPort")
@ -218,8 +224,10 @@ class Node(object):
def __init__(
self, *,
host_system, description, id, # pylint: disable=redefined-builtin
name=None, initial_state=None):
host_system: host_system_lib.HostSystem, description: node_db.NodeDescription,
id: str, # pylint: disable=redefined-builtin
name: Optional[str] = None, initial_state: Optional[audioproc.PluginState] = None
) -> None:
assert isinstance(description, node_db.NodeDescription), description
self._host_system = host_system
@ -228,19 +236,19 @@ class Node(object):
self.id = id
self.initial_state = initial_state
self.__realm = None
self.__realm = None # type: realm_lib.PyRealm
self.broken = False
self.ports = []
self.inputs = {}
self.outputs = {}
self.ports = [] # type: List[Port]
self.inputs = {} # type: Dict[str, InputPortMixin]
self.outputs = {} # type: Dict[str, OutputPortMixin]
self.__control_values = {}
self.__control_values = {} # type: Dict[str, control_value.PyControlValue]
if self.init_ports_from_description:
self.init_ports()
@classmethod
def create(cls, *, description: node_db.NodeDescription, **kwargs) -> 'Node':
def create(cls, *, description: node_db.NodeDescription, **kwargs: Any) -> 'Node':
cls_map = {
node_db.NodeDescription.PROCESSOR: ProcessorNode,
node_db.NodeDescription.PLUGIN: PluginNode,
@ -257,21 +265,21 @@ class Node(object):
return node_cls(description=description, **kwargs)
@property
def realm(self):
def realm(self) -> realm_lib.PyRealm:
assert self.__realm is not None
return self.__realm
def set_realm(self, realm):
def set_realm(self, realm: realm_lib.PyRealm) -> None:
assert self.__realm is None
self.__realm = realm
def clear_realm(self):
def clear_realm(self) -> None:
self.__realm = None
def is_owned_by(self, realm):
def is_owned_by(self, realm: realm_lib.PyRealm) -> bool:
return self.__realm is realm
def init_ports(self):
def init_ports(self) -> None:
port_cls_map = {
(node_db.PortDescription.AUDIO,
node_db.PortDescription.INPUT): AudioInputPort,
@ -313,14 +321,14 @@ class Node(object):
self.outputs[port.name] = port
@property
def parent_nodes(self):
parents = []
def parent_nodes(self) -> List['Node']:
parents = [] # type: List[Node]
for port in self.inputs.values():
for upstream_port in port.inputs:
parents.append(upstream_port.owner)
return parents
async def setup(self):
async def setup(self) -> None:
"""Set up the node.
Any expensive initialization should go here.
@ -335,7 +343,7 @@ class Node(object):
self.__control_values[port.buf_name] = cv
self.realm.add_active_control_value(cv)
async def cleanup(self, deref=False):
async def cleanup(self, deref: bool = False) -> None:
"""Clean up the node.
The counterpart of setup().
@ -343,10 +351,10 @@ class Node(object):
logger.info("%s: cleanup()", self.name)
@property
def control_values(self):
def control_values(self) -> List[control_value.PyControlValue]:
return [v for _, v in sorted(self.__control_values.items())]
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
for cv in self.control_values:
spec.append_control_value(cv)
@ -361,7 +369,7 @@ class Node(object):
for upstream_port in port.inputs:
spec.append_opcode('MIX', upstream_port.buf_name, port.buf_name)
def add_to_spec_post(self, spec):
def add_to_spec_post(self, spec: spec_lib.PySpec) -> None:
for port_idx, port in enumerate(self.ports):
if isinstance(port, AudioOutputPort):
spec.append_opcode('POST_RMS', self.id, port_idx, port.buf_name)
@ -370,23 +378,23 @@ class Node(object):
class ProcessorNode(Node):
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__processor = processor.PyProcessor(self.id, self._host_system, self.description)
self.__processor = processor_lib.PyProcessor(self.id, self._host_system, self.description)
@property
def processor(self):
def processor(self) -> processor_lib.PyProcessor:
assert self.__processor is not None
return self.__processor
async def setup(self):
async def setup(self) -> None:
await super().setup()
self.__processor.setup()
self.realm.add_active_processor(self.__processor)
async def cleanup(self, deref=False):
async def cleanup(self, deref: bool = False) -> None:
if self.__processor is not None:
if deref:
self.__processor = None
@ -395,7 +403,7 @@ class ProcessorNode(Node):
await super().cleanup(deref)
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
super().add_to_spec_pre(spec)
spec.append_processor(self.__processor)
@ -407,13 +415,13 @@ class ProcessorNode(Node):
class PluginNode(ProcessorNode):
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__plugin_host = None
self.__plugin_pipe_path = None
self.__plugin_host = None # type: ipc.Stub
self.__plugin_pipe_path = None # type: str
async def setup(self):
async def setup(self) -> None:
await super().setup()
self.__plugin_host = await self.realm.get_plugin_host()
@ -431,7 +439,7 @@ class PluginNode(ProcessorNode):
processor_pb2.ProcessorParameters(
plugin_pipe_path=os.fsencode(self.__plugin_pipe_path)))
async def cleanup(self, deref=False):
async def cleanup(self, deref: bool = False) -> None:
await super().cleanup(deref)
if self.__plugin_pipe_path is not None:
@ -440,7 +448,7 @@ class PluginNode(ProcessorNode):
self.__plugin_host = None
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
super().add_to_spec_pre(spec)
spec.append_buffer('%s:plugin_cond' % self.id, buffers.PyPluginCondBuffer())
spec.append_opcode(
@ -448,28 +456,28 @@ class PluginNode(ProcessorNode):
class RealmSinkNode(Node):
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(id='sink', **kwargs)
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
super().add_to_spec_pre(spec)
spec.append_opcode('POST_RMS', self.id, 0, self.inputs['in:left'].buf_name)
spec.append_opcode('POST_RMS', self.id, 1, self.inputs['in:right'].buf_name)
class ChildRealmNode(Node):
def __init__(self, *, child_realm, **kwargs):
def __init__(self, *, child_realm: str, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__child_realm_name = child_realm
self.__child_realm = None
self.__child_realm = None # type: realm_lib.PyRealm
async def setup(self):
async def setup(self) -> None:
await super().setup()
self.__child_realm = self.realm.child_realms[self.__child_realm_name]
self.realm.add_active_child_realm(self.__child_realm)
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
super().add_to_spec_pre(spec)
spec.append_child_realm(self.__child_realm)
@ -480,12 +488,12 @@ class ChildRealmNode(Node):
class EventSourceNode(Node):
def __init__(self, *, track_id, **kwargs):
def __init__(self, *, track_id: str, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__track_id = track_id
def add_to_spec_pre(self, spec):
def add_to_spec_pre(self, spec: spec_lib.PySpec) -> None:
super().add_to_spec_pre(spec)
spec.append_opcode(
@ -495,25 +503,25 @@ class EventSourceNode(Node):
class Graph(object):
def __init__(self, realm):
def __init__(self, realm: realm_lib.PyRealm) -> None:
self.__realm = realm
self.__nodes = {}
self.__nodes = {} # type: Dict[str, Node]
@property
def nodes(self):
def nodes(self) -> Set[Node]:
return set(self.__nodes.values())
def find_node(self, node_id):
def find_node(self, node_id: str) -> Node:
return self.__nodes[node_id]
def add_node(self, node):
def add_node(self, node: Node) -> None:
if node.id in self.__nodes:
raise GraphError("Duplicate node ID '%s'" % node.id)
node.set_realm(self.__realm)
self.__nodes[node.id] = node
def remove_node(self, node):
def remove_node(self, node: Node) -> None:
if not node.is_owned_by(self.__realm):
raise GraphError("Node has not been added to this realm")
node.clear_realm()

27
noisicaa/audioproc/engine/message_queue.pyi

@ -0,0 +1,27 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
class PyMessage(object):
def __init__(self) -> None: ...
@property
def type(self) -> str: ...
@property
def size(self) -> int: ...

29
noisicaa/audioproc/engine/player.pyi

@ -0,0 +1,29 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from noisicaa import audioproc
from noisicaa import host_system as host_system_lib
class PyPlayer(object):
def __init__(self, host_system: host_system_lib.HostSystem) -> None: ...
def setup(self) -> None: ...
def cleanup(self) -> None: ...
def update_state(self, state: audioproc.PlayerState) -> None: ...

52
noisicaa/audioproc/engine/plugin_host.pyi

@ -0,0 +1,52 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Callable, List, Tuple
from noisicaa import host_system as host_system_lib
from noisicaa import audioproc
from . import plugin_ui_host
from . import plugin_host_pb2
def build_memory_mapping(
shmem_path: str, cond_offset: int, block_size: int, buffers: List[Tuple[int, int]]
) -> bytearray: ...
def init_cond(buf: memoryview, offset: int) -> int: ...
def cond_wait(buf: memoryview, offset: int) -> None: ...
def cond_clear(buf: memoryview, offset: int) -> None: ...
class PyPluginHost(object):
def __init__(
self, spec: plugin_host_pb2.PluginInstanceSpec, host_system: host_system_lib.HostSystem
) -> None: ...
def setup(self) -> None: ...
def cleanup(self) -> None: ...
def create_ui(
self, control_value_change_cb: Callable[[int, float, int], None]
) -> plugin_ui_host.PyPluginUIHost: ...
def main_loop(self, pipe_fd: int) -> None: ...
def exit_loop(self) -> None: ...
def connect_port(self, port: int, data: bytearray) -> None: ...
def process_block(self, block_size: int) -> None: ...
def has_state(self) -> bool: ...
def get_state(self) -> audioproc.PluginState: ...
def set_state(self, state: audioproc.PluginState) -> None: ...

13
noisicaa/audioproc/engine/plugin_host_process.py

@ -20,8 +20,6 @@
#
# @end:license
# TODO: mypy-unclean
import asyncio
import concurrent.futures
import functools
@ -41,7 +39,8 @@ with warnings.catch_warnings():
# So let's cross fingers that is keeps working and suppress that warning.
warnings.filterwarnings('ignore', r"You have imported the Gtk 2\.0 module")
gi.require_version("Gtk", "2.0")
from gi.repository import Gtk # pylint: disable=unused-import
# pylint: disable=unused-import
from gi.repository import Gtk # type: ignore
from noisicaa import core
from noisicaa.core import ipc
@ -56,6 +55,7 @@ from . import plugin_ui_host # pylint: disable=unused-import
logger = logging.getLogger(__name__)
# TODO: this should not extend PyPluginHost, but be a proxy class.
class PluginHost(plugin_host.PyPluginHost):
def __init__(
self, *,
@ -107,7 +107,8 @@ class PluginHost(plugin_host.PyPluginHost):
super().set_state(state)
self.__state = state
async def setup(self) -> None:
# Parent class doesn't use async setup/cleanup.
async def setup(self) -> None: # type: ignore
super().setup()
if self.__callback_address is not None:
@ -132,7 +133,7 @@ class PluginHost(plugin_host.PyPluginHost):
self.__state_fetcher_task = self.__event_loop.create_task(
self.__state_fetcher_main())
async def cleanup(self) -> None:
async def cleanup(self) -> None: # type: ignore
if self.__state_fetcher_task:
if self.__state_fetcher_task.done():
self.__state_fetcher_task.result()
@ -172,7 +173,7 @@ class PluginHost(plugin_host.PyPluginHost):
else:
self.__thread_result.set_result(True)
async def __state_fetcher_main(self):
async def __state_fetcher_main(self) -> None:
while True:
await asyncio.sleep(1.0, loop=self.__event_loop)

8
noisicaa/audioproc/engine/plugin_host_process_test.py

@ -20,8 +20,6 @@
#
# @end:license
# TODO: mypy-unclean
import asyncio
import logging
import mmap
@ -142,9 +140,11 @@ class PluginHostProcessTest(
p = shm_data[offset:offset+bufsize]
if port_spec.type == node_db.PortDescription.AUDIO:
bufp[port_spec.name] = p.cast('f')
# TODO: mypy doesn't know memoryview.cast
bufp[port_spec.name] = p.cast('f') # type: ignore
elif port_spec.type == node_db.PortDescription.KRATE_CONTROL:
bufp[port_spec.name] = p.cast('f')
# TODO: mypy doesn't know memoryview.cast
bufp[port_spec.name] = p.cast('f') # type: ignore
else:
raise ValueError(port_spec.type)

5
noisicaa/audioproc/engine/plugin_host_test.py

@ -18,8 +18,6 @@
#
# @end:license
# TODO: mypy-unclean
import contextlib
from noisidev import unittest_mixins
@ -58,7 +56,8 @@ class PluginHostMixin(unittest_mixins.NodeDBMixin, unittest_engine_mixins.HostSy
buf = bytearray(bufsize)
plugin.connect_port(idx, buf)
bufp[port.name] = memoryview(buf).cast(valuetype)
# TODO: mypy doesn't know memoryview.cast
bufp[port.name] = memoryview(buf).cast(valuetype) # type: ignore
yield plugin, bufp

30
noisicaa/audioproc/engine/plugin_ui_host.pyi

@ -0,0 +1,30 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Tuple
class PyPluginUIHost(object):
def setup(self) -> None: ...
def cleanup(self) -> None: ...
@property
def wid(self) -> int: ...
@property
def size(self) -> Tuple[int, int]: ...

53
noisicaa/audioproc/engine/processor.pyi

@ -0,0 +1,53 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
import enum
from noisicaa import node_db
from noisicaa import audioproc
from noisicaa import host_system as host_system_lib
from . import block_context
from . import processor_pb2
class State(enum.Enum):
INACTIVE = ... # type: State
SETUP = ... # type: State
RUNNING = ... # type: State
BROKEN = ... # type: State
CLEANUP = ... # type: State
class PyProcessor(object):
def __init__(
self, node_id: str, host_system: host_system_lib.HostSystem,
node_description: node_db.NodeDescription) -> None: ...
@property
def id(self) -> int: ...
@property
def state(self) -> State: ...
def setup(self) -> None: ...
def cleanup(self) -> None: ...
def connect_port(
self, ctxt: block_context.PyBlockContext, port_index: int, data: memoryview) -> None: ...
def process_block(
self, ctxt: block_context.PyBlockContext, time_mapper: audioproc.TimeMapper) -> None: ...
def handle_message(self, msg: audioproc.ProcessorMessage) -> None: ...
def set_parameters(self, parameters: processor_pb2.ProcessorParameters) -> None: ...

2
noisicaa/audioproc/engine/processor_csound_test.py

@ -18,8 +18,6 @@
#
# @end:license
# TODO: mypy-unclean
import textwrap
from noisidev import unittest

86
noisicaa/audioproc/engine/realm.pyi

@ -0,0 +1,86 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Dict, List, Optional
from noisicaa import core
from noisicaa.core import ipc
from noisicaa import audioproc
from noisicaa import node_db
from noisicaa import host_system as host_system_lib
from . import player as player_lib
from . import spec as spec_lib
from . import processor
from . import control_value as control_value_lib
from . import block_context as block_context_lib
from . import buffers
from . import processor
from . import graph as graph_lib
from . import engine as engine_lib
# We actually use memoryviews, but mypy doesn't know that a memoryview can also behave like an
# array of floats.
BufferView = List
class PyProgram(object):
pass
class PyRealm(object):
listeners = ... # type: core.CallbackRegistry
child_realms = ... # type: Dict[str, PyRealm]
def __init__(
self, *, engine: engine_lib.Engine, name: str, parent: PyRealm,
host_system: host_system_lib.HostSystem, player: player_lib.PyPlayer,
callback_address: str) -> None: ...
@property
def name(self) -> str: ...
@property
def parent(self) -> PyRealm: ...
@property
def graph(self) -> graph_lib.Graph: ...
@property
def player(self) -> player_lib.PyPlayer: ...
@property
def callback_address(self) -> str: ...
@property
def block_context(self) -> block_context_lib.PyBlockContext: ...
async def setup(self) -> None: ...
async def cleanup(self) -> None: ...
def clear_programs(self) -> None: ...
def get_buffer(self, name: str, type: buffers.PyBufferType) -> BufferView: ...
async def get_plugin_host(self) -> ipc.Stub: ...
def update_spec(self) -> None: ...
def set_spec(self, spec: spec_lib.PySpec) -> None: ...
async def setup_node(self, node: graph_lib.Node) -> None: ...
def add_active_processor(self, proc: processor.PyProcessor) -> None: ...
def add_active_control_value(self, control_value: control_value_lib.PyControlValue) -> None: ...
def add_active_child_realm(self, child: PyRealm) -> None: ...
def set_control_value(self, name: str, value: float, generation: int) -> None: ...
async def set_plugin_state(self, node: str, state: audioproc.PluginState) -> None: ...
def send_node_message(self, msg: audioproc.ProcessorMessage) -> None: ...
def update_project_properties(
self, *, bpm: Optional[int] = None,
duration: Optional[audioproc.MusicalDuration] = None) -> None: ...
def get_active_program(self) -> Optional[PyProgram]: ...
def process_block(self, program: PyProgram) -> None: ...
def run_maintenance(self) -> None: ...

2
noisicaa/audioproc/engine/realm_test.py

@ -18,8 +18,6 @@
#
# @end:license
# TODO: mypy-unclean
import os
import os.path

43
noisicaa/audioproc/engine/spec.pyi

@ -0,0 +1,43 @@
# @begin:license
#
# Copyright (c) 2015-2018, Benjamin Niemann <pink@odahoda.de>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# @end:license
from typing import Any, Dict
from noisicaa import audioproc
from . import buffers
from . import control_value
from . import processor as processor_lib
from . import realm
opcode_map = ... # type: Dict[str, int]
opname = ... # type: Dict[int, str]
class PySpec(object):
bpm = ... # type: int
duration = ... # type: audioproc.MusicalTime
def __init__(self) -> None: ...
def dump(self) -> str: ...
def append_buffer(self, name: str, buf_type: buffers.PyBufferType) -> None: ...
def append_control_value(self, cv: control_value.PyControlValue) -> None: ...
def append_processor(self, processor: processor_lib.PyProcessor) -> None: ...
def append_child_realm(self, child_realm: realm.PyRealm) -> None: ...
def append_opcode(self, opcode: str, *args: Any) -> None: ...

2
noisicaa/ui/instrument_library_test.py

@ -20,8 +20,6 @@
#
# @end:license
# TODO: mypy-unclean
import os.path
from PyQt5.QtCore import Qt

27
noisidev/unittest_engine_utils.pyi

@ -18,30 +18,37 @@
#
# @end:license
from typing import List
from noisicaa import node_db
from noisicaa.host_system.host_system import PyHostSystem
from noisicaa import host_system as host_system_lib
from noisicaa.audioproc.engine import buffers
from noisicaa.audioproc.engine.buffer_arena import PyBufferArena
from noisicaa.audioproc.engine.block_context import PyBlockContext
from noisicaa.audioproc.engine.processor import PyProcessor
from noisicaa.audioproc.engine import buffer_arena
from noisicaa.audioproc.engine import block_context
from noisicaa.audioproc.engine import processor
# We actually use memoryviews, but mypy doesn't know that a memoryview can also behave like an
# array of floats.
BufferView = List
class BufferManager(object):
def __init__(
self,
host_system: PyHostSystem ,
arena: PyBufferArena = None,
host_system: host_system_lib.HostSystem ,
arena: buffer_arena.PyBufferArena = None,
size: int = 2**20) -> None:
...
def allocate_from_node_description(
self, node_description: node_db.NodeDescription, prefix: str = '') -> None: ...
def connect_ports(
self,
proc: PyProcessor,
ctxt: PyBlockContext,
proc: processor.PyProcessor,
ctxt: block_context.PyBlockContext,
node_description: node_db.NodeDescription,
prefix: str = '') -> None: ...
def allocate(self, name: str, type: buffers.PyBufferType) -> memoryview: ...
def __getitem__(self, name: str) -> memoryview: ...
def allocate(self, name: str, type: buffers.PyBufferType) -> BufferView: ...
def __getitem__(self, name: str) -> BufferView: ...
def type(self, name: str) -> buffers.PyBufferType: ...
def data(self, name: str) -> memoryview: ...

Loading…
Cancel
Save