The great model merge itself.

Create a single class hierarchy for model classes.
model-merge
Ben Niemann 2019-05-03 23:53:03 +02:00
parent 1ab72dba20
commit 4d072f83f5
148 changed files with 3915 additions and 5157 deletions

View File

@ -38,7 +38,8 @@ add_subdirectory(host_system)
add_subdirectory(instr)
add_subdirectory(instrument_db)
add_subdirectory(lv2)
add_subdirectory(model)
add_subdirectory(model_base)
add_subdirectory(value_types)
add_subdirectory(music)
add_subdirectory(node_db)
add_subdirectory(ui)

View File

@ -21,8 +21,7 @@
add_python_package(
node_description_registry.py
ui_registry.py
client_registry.py
server_registry.py
model_registry.py
)
py_proto(model_registry.proto)

View File

@ -22,9 +22,7 @@ add_python_package(
node_description.py
model.py
commands.py
client_impl.py
client_impl_test.py
server_impl.py
server_impl_test.py
node_ui.py
track_ui.py

View File

@ -1,64 +0,0 @@
#!/usr/bin/python3
# @begin:license
#
# Copyright (c) 2015-2019, 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 Sequence
from noisicaa.core.typing_extra import down_cast
from noisicaa import audioproc
from noisicaa import model
from noisicaa.music import project_client_model
from . import model as beat_track_model
class Beat(
project_client_model.ProjectChild,
beat_track_model.Beat,
project_client_model.ObjectBase):
@property
def time(self) -> audioproc.MusicalDuration:
return audioproc.MusicalDuration.from_proto(self.get_property_value('time'))
@property
def velocity(self) -> int:
return self.get_property_value('velocity')
@property
def measure(self) -> 'BeatMeasure':
return down_cast(BeatMeasure, super().measure)
class BeatMeasure(
project_client_model.Measure,
beat_track_model.BeatMeasure,
project_client_model.ObjectBase):
@property
def beats(self) -> Sequence[Beat]:
return self.get_property_value('beats')
class BeatTrack(
project_client_model.MeasuredTrack,
beat_track_model.BeatTrack,
project_client_model.ObjectBase):
@property
def pitch(self) -> model.Pitch:
return self.get_property_value('pitch')

View File

@ -22,16 +22,16 @@
from noisidev import unittest
from noisicaa import audioproc
from noisicaa import model
from noisicaa import value_types
from noisicaa import music
from noisicaa.music import base_track_test
from . import server_impl
from . import model
from . import commands
class BeatTrackTest(base_track_test.TrackTestMixin, unittest.AsyncTestCase):
node_uri = 'builtin://beat-track'
track_cls = server_impl.BeatTrack
track_cls = model.BeatTrack
async def test_create_measure(self):
track = await self._add_track()
@ -64,8 +64,8 @@ class BeatTrackTest(base_track_test.TrackTestMixin, unittest.AsyncTestCase):
await self.client.send_command(commands.update(
track,
set_pitch=model.Pitch('C2')))
self.assertEqual(track.pitch, model.Pitch('C2'))
set_pitch=value_types.Pitch('C2')))
self.assertEqual(track.pitch, value_types.Pitch('C2'))
async def test_add_beat(self):
track = await self._add_track()

View File

@ -23,7 +23,8 @@
syntax = "proto2";
import "noisicaa/audioproc/public/musical_time.proto";
import "noisicaa/model/project.proto";
import "noisicaa/value_types/value_types.proto";
import "noisicaa/music/model.proto";
package noisicaa.pb;

View File

@ -21,14 +21,14 @@
# @end:license
from noisicaa import audioproc
from noisicaa import model
from noisicaa import value_types
from noisicaa import music
from noisicaa.builtin_nodes import commands_registry_pb2
from . import client_impl
from . import model
def update(
track: client_impl.BeatTrack, *, set_pitch: model.Pitch = None) -> music.Command:
track: model.BeatTrack, *, set_pitch: value_types.Pitch = None) -> music.Command:
cmd = music.Command(command='update_beat_track')
pb = cmd.Extensions[commands_registry_pb2.update_beat_track]
pb.track_id = track.id
@ -37,7 +37,7 @@ def update(
return cmd
def create_beat(
measure: client_impl.BeatMeasure, *,
measure: model.BeatMeasure, *,
time: audioproc.MusicalTime,
velocity: int = None
) -> music.Command:
@ -50,7 +50,7 @@ def create_beat(
return cmd
def update_beat(
beat: client_impl.Beat, *, set_velocity: int = None) -> music.Command:
beat: model.Beat, *, set_velocity: int = None) -> music.Command:
cmd = music.Command(command='update_beat')
pb = cmd.Extensions[commands_registry_pb2.update_beat]
pb.beat_id = beat.id
@ -58,7 +58,7 @@ def update_beat(
pb.set_velocity = set_velocity
return cmd
def delete_beat(beat: client_impl.Beat) -> music.Command:
def delete_beat(beat: model.Beat) -> music.Command:
cmd = music.Command(command='delete_beat')
pb = cmd.Extensions[commands_registry_pb2.delete_beat]
pb.beat_id = beat.id

View File

@ -23,7 +23,8 @@
syntax = "proto2";
import "noisicaa/audioproc/public/musical_time.proto";
import "noisicaa/model/project.proto";
import "noisicaa/value_types/value_types.proto";
import "noisicaa/music/model.proto";
package noisicaa.pb;

View File

@ -20,52 +20,177 @@
#
# @end:license
from typing import cast, Any
from typing import cast, Any, Optional, MutableSequence, Callable, Iterator
from noisicaa.core.typing_extra import down_cast
from noisicaa import core
from noisicaa import node_db
from noisicaa import model
from noisicaa import audioproc
from noisicaa import model_base
from noisicaa import value_types
from noisicaa.music import base_track
from noisicaa.music import model
from noisicaa.music import commands
from noisicaa.audioproc.public import musical_time_pb2
from noisicaa.builtin_nodes import commands_registry_pb2
from noisicaa.builtin_nodes import model_registry_pb2
from . import node_description
from . import commands_pb2
class UpdateBeatTrack(commands.Command):
proto_type = 'update_beat_track'
proto_ext = commands_registry_pb2.update_beat_track
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateBeatTrack, self.pb)
track = down_cast(BeatTrack, self.pool[pb.track_id])
if pb.HasField('set_pitch'):
track.pitch = value_types.Pitch.from_proto(pb.set_pitch)
class CreateBeat(commands.Command):
proto_type = 'create_beat'
proto_ext = commands_registry_pb2.create_beat
def run(self) -> None:
pb = down_cast(commands_pb2.CreateBeat, self.pb)
measure = down_cast(BeatMeasure, self.pool[pb.measure_id])
time = audioproc.MusicalDuration.from_proto(pb.time)
assert audioproc.MusicalDuration(0, 1) <= time < measure.duration
if pb.HasField('velocity'):
velocity = pb.velocity
else:
velocity = 100
beat = self.pool.create(
Beat,
time=time,
velocity=velocity)
measure.beats.append(beat)
class UpdateBeat(commands.Command):
proto_type = 'update_beat'
proto_ext = commands_registry_pb2.update_beat
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateBeat, self.pb)
beat = down_cast(Beat, self.pool[pb.beat_id])
if pb.HasField('set_velocity'):
beat.velocity = pb.set_velocity
class DeleteBeat(commands.Command):
proto_type = 'delete_beat'
proto_ext = commands_registry_pb2.delete_beat
def run(self) -> None:
pb = down_cast(commands_pb2.DeleteBeat, self.pb)
beat = down_cast(Beat, self.pool[pb.beat_id])
measure = beat.measure
del measure.beats[beat.index]
class BeatTrackConnector(base_track.MeasuredTrackConnector):
_node = None # type: BeatTrack
def _add_track_listeners(self) -> None:
self._listeners['pitch'] = self._node.pitch_changed.add(self.__pitch_changed)
def _add_measure_listeners(self, mref: base_track.MeasureReference) -> None:
measure = down_cast(BeatMeasure, mref.measure)
self._listeners['measure:%s:beats' % mref.id] = measure.content_changed.add(
lambda _=None: self.__measure_beats_changed(mref)) # type: ignore
def _remove_measure_listeners(self, mref: base_track.MeasureReference) -> None:
self._listeners.pop('measure:%s:beats' % mref.id).remove()
def _create_events(
self, time: audioproc.MusicalTime, measure: base_track.Measure
) -> Iterator[base_track.PianoRollInterval]:
measure = down_cast(BeatMeasure, measure)
for beat in measure.beats:
beat_time = time + beat.time
event = base_track.PianoRollInterval(
beat_time, beat_time + audioproc.MusicalDuration(1, 4),
self._node.pitch, 127)
yield event
def __pitch_changed(self, change: model_base.PropertyChange) -> None:
self._update_measure_range(0, len(self._node.measure_list))
def __measure_beats_changed(self, mref: base_track.MeasureReference) -> None:
self._update_measure_range(mref.index, mref.index + 1)
class Beat(model.ProjectChild):
class BeatSpec(model.ObjectSpec):
class BeatSpec(model_base.ObjectSpec):
proto_type = 'beat'
proto_ext = model_registry_pb2.beat
time = model.ProtoProperty(musical_time_pb2.MusicalDuration)
velocity = model.Property(int)
time = model_base.ProtoProperty(musical_time_pb2.MusicalDuration)
velocity = model_base.Property(int)
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.time_changed = \
core.Callback[model.PropertyChange[musical_time_pb2.MusicalDuration]]()
self.velocity_changed = core.Callback[model.PropertyChange[int]]()
core.Callback[model_base.PropertyChange[musical_time_pb2.MusicalDuration]]()
self.velocity_changed = core.Callback[model_base.PropertyChange[int]]()
def create(
self, *,
time: Optional[audioproc.MusicalDuration] = None,
velocity: Optional[int] = None,
**kwargs: Any) -> None:
super().create(**kwargs)
self.time = time
self.velocity = velocity
@property
def time(self) -> audioproc.MusicalDuration:
return audioproc.MusicalDuration.from_proto(self.get_property_value('time'))
@time.setter
def time(self, value: audioproc.MusicalDuration) -> None:
self.set_property_value('time', value.to_proto())
@property
def velocity(self) -> int:
return self.get_property_value('velocity')
@velocity.setter
def velocity(self, value: int) -> None:
self.set_property_value('velocity', value)
@property
def measure(self) -> 'BeatMeasure':
return cast(BeatMeasure, self.parent)
def property_changed(self, change: model.PropertyChange) -> None:
def property_changed(self, change: model_base.PropertyChange) -> None:
super().property_changed(change)
if self.measure is not None:
self.measure.content_changed.call()
class BeatMeasure(model.Measure):
class BeatMeasureSpec(model.ObjectSpec):
class BeatMeasure(base_track.Measure):
class BeatMeasureSpec(model_base.ObjectSpec):
proto_type = 'beat_measure'
proto_ext = model_registry_pb2.beat_measure
beats = model.ObjectListProperty(Beat)
beats = model_base.ObjectListProperty(Beat)
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.beats_changed = core.Callback[model.PropertyListChange[Beat]]()
self.beats_changed = core.Callback[model_base.PropertyListChange[Beat]]()
self.content_changed = core.Callback[None]()
@ -73,19 +198,59 @@ class BeatMeasure(model.Measure):
super().setup()
self.beats_changed.add(lambda _: self.content_changed.call())
@property
def beats(self) -> MutableSequence[Beat]:
return self.get_property_value('beats')
class BeatTrack(model.MeasuredTrack):
class BeatTrackSpec(model.ObjectSpec):
@property
def empty(self) -> bool:
return len(self.beats) == 0
class BeatTrack(base_track.MeasuredTrack):
class BeatTrackSpec(model_base.ObjectSpec):
proto_type = 'beat_track'
proto_ext = model_registry_pb2.beat_track
pitch = model.WrappedProtoProperty(model.Pitch)
pitch = model_base.WrappedProtoProperty(value_types.Pitch)
measure_cls = BeatMeasure
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.pitch_changed = core.Callback[model.PropertyChange[model.Pitch]]()
self.pitch_changed = core.Callback[model_base.PropertyChange[value_types.Pitch]]()
def create(
self, *,
pitch: Optional[value_types.Pitch] = None,
num_measures: int = 1, **kwargs: Any) -> None:
super().create(**kwargs)
if pitch is None:
self.pitch = value_types.Pitch('B2')
else:
self.pitch = pitch
for _ in range(num_measures):
self.append_measure()
@property
def pitch(self) -> value_types.Pitch:
return self.get_property_value('pitch')
@pitch.setter
def pitch(self, value: value_types.Pitch) -> None:
self.set_property_value('pitch', value)
@property
def description(self) -> node_db.NodeDescription:
return node_description.BeatTrackDescription
def create_node_connector(
self,
message_cb: Callable[[audioproc.ProcessorMessage], None],
audioproc_client: audioproc.AbstractAudioProcClient,
) -> BeatTrackConnector:
return BeatTrackConnector(
node=self, message_cb=message_cb, audioproc_client=audioproc_client)

View File

@ -30,18 +30,19 @@ from PyQt5 import QtSvg
from noisicaa.constants import DATA_DIR
from noisicaa import core
from noisicaa import model
from noisicaa import value_types
from noisicaa import model_base
from noisicaa import music
from noisicaa.ui.graph import track_node
from noisicaa.ui import ui_base
from . import commands
from . import client_impl
from . import model
logger = logging.getLogger(__name__)
class BeatTrackWidget(ui_base.ProjectMixin, QtWidgets.QScrollArea):
def __init__(self, track: client_impl.BeatTrack, **kwargs: Any) -> None:
def __init__(self, track: model.BeatTrack, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__track = track
@ -73,12 +74,12 @@ class BeatTrackWidget(ui_base.ProjectMixin, QtWidgets.QScrollArea):
listener.remove()
self.__listeners.clear()
def __pitchChanged(self, change: model.PropertyValueChange[str]) -> None:
def __pitchChanged(self, change: model_base.PropertyValueChange[str]) -> None:
self.__pitch.setText(str(change.new_value))
def __pitchEdited(self) -> None:
try:
pitch = model.Pitch(self.__pitch.text())
pitch = value_types.Pitch(self.__pitch.text())
except ValueError:
self.__pitch.setText(str(self.__track.pitch))
else:
@ -90,9 +91,9 @@ class BeatTrackWidget(ui_base.ProjectMixin, QtWidgets.QScrollArea):
class BeatTrackNode(track_node.TrackNode):
def __init__(self, node: music.BaseNode, **kwargs: Any) -> None:
assert isinstance(node, client_impl.BeatTrack)
assert isinstance(node, model.BeatTrack)
self.__widget = None # type: BeatTrackWidget
self.__track = node # type: client_impl.BeatTrack
self.__track = node # type: model.BeatTrack
super().__init__(
node=node,

View File

@ -1,203 +0,0 @@
#!/usr/bin/python3
# @begin:license
#
# Copyright (c) 2015-2019, 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 logging
from typing import Any, MutableSequence, Optional, Iterator, Callable
from noisicaa.core.typing_extra import down_cast
from noisicaa import audioproc
from noisicaa import model
from noisicaa.music import commands
from noisicaa.music import pmodel
from noisicaa.music import base_track
from noisicaa.builtin_nodes import commands_registry_pb2
from . import commands_pb2
from . import model as beat_track_model
logger = logging.getLogger(__name__)
class UpdateBeatTrack(commands.Command):
proto_type = 'update_beat_track'
proto_ext = commands_registry_pb2.update_beat_track
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateBeatTrack, self.pb)
track = down_cast(BeatTrack, self.pool[pb.track_id])
if pb.HasField('set_pitch'):
track.pitch = model.Pitch.from_proto(pb.set_pitch)
class CreateBeat(commands.Command):
proto_type = 'create_beat'
proto_ext = commands_registry_pb2.create_beat
def run(self) -> None:
pb = down_cast(commands_pb2.CreateBeat, self.pb)
measure = down_cast(BeatMeasure, self.pool[pb.measure_id])
time = audioproc.MusicalDuration.from_proto(pb.time)
assert audioproc.MusicalDuration(0, 1) <= time < measure.duration
if pb.HasField('velocity'):
velocity = pb.velocity
else:
velocity = 100
beat = self.pool.create(
Beat,
time=time,
velocity=velocity)
measure.beats.append(beat)
class UpdateBeat(commands.Command):
proto_type = 'update_beat'
proto_ext = commands_registry_pb2.update_beat
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateBeat, self.pb)
beat = down_cast(Beat, self.pool[pb.beat_id])
if pb.HasField('set_velocity'):
beat.velocity = pb.set_velocity
class DeleteBeat(commands.Command):
proto_type = 'delete_beat'
proto_ext = commands_registry_pb2.delete_beat
def run(self) -> None:
pb = down_cast(commands_pb2.DeleteBeat, self.pb)
beat = down_cast(Beat, self.pool[pb.beat_id])
measure = beat.measure
del measure.beats[beat.index]
class Beat(pmodel.ProjectChild, beat_track_model.Beat, pmodel.ObjectBase):
def create(
self, *,
time: Optional[audioproc.MusicalDuration] = None,
velocity: Optional[int] = None,
**kwargs: Any) -> None:
super().create(**kwargs)
self.time = time
self.velocity = velocity
@property
def time(self) -> audioproc.MusicalDuration:
return audioproc.MusicalDuration.from_proto(self.get_property_value('time'))
@time.setter
def time(self, value: audioproc.MusicalDuration) -> None:
self.set_property_value('time', value.to_proto())
@property
def velocity(self) -> int:
return self.get_property_value('velocity')
@velocity.setter
def velocity(self, value: int) -> None:
self.set_property_value('velocity', value)
@property
def measure(self) -> 'BeatMeasure':
return down_cast(BeatMeasure, super().measure)
class BeatMeasure(base_track.Measure, beat_track_model.BeatMeasure, pmodel.ObjectBase):
@property
def beats(self) -> MutableSequence[Beat]:
return self.get_property_value('beats')
@property
def empty(self) -> bool:
return len(self.beats) == 0
class BeatTrackConnector(base_track.MeasuredTrackConnector):
_node = None # type: BeatTrack
def _add_track_listeners(self) -> None:
self._listeners['pitch'] = self._node.pitch_changed.add(self.__pitch_changed)
def _add_measure_listeners(self, mref: pmodel.MeasureReference) -> None:
measure = down_cast(BeatMeasure, mref.measure)
self._listeners['measure:%s:beats' % mref.id] = measure.content_changed.add(
lambda _=None: self.__measure_beats_changed(mref)) # type: ignore
def _remove_measure_listeners(self, mref: pmodel.MeasureReference) -> None:
self._listeners.pop('measure:%s:beats' % mref.id).remove()
def _create_events(
self, time: audioproc.MusicalTime, measure: pmodel.Measure
) -> Iterator[base_track.PianoRollInterval]:
measure = down_cast(BeatMeasure, measure)
for beat in measure.beats:
beat_time = time + beat.time
event = base_track.PianoRollInterval(
beat_time, beat_time + audioproc.MusicalDuration(1, 4),
self._node.pitch, 127)
yield event
def __pitch_changed(self, change: model.PropertyChange) -> None:
self._update_measure_range(0, len(self._node.measure_list))
def __measure_beats_changed(self, mref: pmodel.MeasureReference) -> None:
self._update_measure_range(mref.index, mref.index + 1)
class BeatTrack(base_track.MeasuredTrack, beat_track_model.BeatTrack, pmodel.ObjectBase):
measure_cls = BeatMeasure
def create(
self, *,
pitch: Optional[model.Pitch] = None,
num_measures: int = 1, **kwargs: Any) -> None:
super().create(**kwargs)
if pitch is None:
self.pitch = model.Pitch('B2')
else:
self.pitch = pitch
for _ in range(num_measures):
self.append_measure()
@property
def pitch(self) -> model.Pitch:
return self.get_property_value('pitch')
@pitch.setter
def pitch(self, value: model.Pitch) -> None:
self.set_property_value('pitch', value)
def create_node_connector(
self,
message_cb: Callable[[audioproc.ProcessorMessage], None],
audioproc_client: audioproc.AbstractAudioProcClient,
) -> BeatTrackConnector:
return BeatTrackConnector(
node=self, message_cb=message_cb, audioproc_client=audioproc_client)

View File

@ -20,44 +20,44 @@
#
# @end:license
from noisicaa import audioproc
from noisicaa import model
from noisicaa.music import pmodel_test
from . import server_impl
# from noisicaa import audioproc
# from noisicaa import value_types
# from noisicaa.music import pmodel_test
# from . import model
T0 = audioproc.MusicalDuration(0, 4)
# T0 = audioproc.MusicalDuration(0, 4)
class BeatTrackTest(pmodel_test.MeasuredTrackMixin, pmodel_test.ModelTest):
cls = server_impl.BeatTrack
create_args = {'name': 'test'}
measure_cls = server_impl.BeatMeasure
# class BeatTrackTest(pmodel_test.MeasuredTrackMixin, pmodel_test.ModelTest):
# cls = model.BeatTrack
# create_args = {'name': 'test'}
# measure_cls = model.BeatMeasure
def test_pitch(self):
track = self.pool.create(self.cls, **self.create_args)
# def test_pitch(self):
# track = self.pool.create(self.cls, **self.create_args)
track.pitch = model.Pitch('F4')
self.assertEqual(track.pitch, model.Pitch('F4'))
# track.pitch = value_types.Pitch('F4')
# self.assertEqual(track.pitch, value_types.Pitch('F4'))
class BeatMeasureTest(pmodel_test.ModelTest):
def test_beats(self):
measure = self.pool.create(server_impl.BeatMeasure)
# class BeatMeasureTest(pmodel_test.ModelTest):
# def test_beats(self):
# measure = self.pool.create(model.BeatMeasure)
beat = self.pool.create(server_impl.Beat, time=T0, velocity=100)
measure.beats.append(beat)
self.assertIs(measure.beats[0], beat)
# beat = self.pool.create(model.Beat, time=T0, velocity=100)
# measure.beats.append(beat)
# self.assertIs(measure.beats[0], beat)
class BeatTest(pmodel_test.ModelTest):
def test_time(self):
beat = self.pool.create(server_impl.Beat, time=T0, velocity=100)
# class BeatTest(pmodel_test.ModelTest):
# def test_time(self):
# beat = self.pool.create(model.Beat, time=T0, velocity=100)
beat.time = audioproc.MusicalDuration(1, 4)
self.assertEqual(beat.time, audioproc.MusicalDuration(1, 4))
# beat.time = audioproc.MusicalDuration(1, 4)
# self.assertEqual(beat.time, audioproc.MusicalDuration(1, 4))
def test_velocity(self):
beat = self.pool.create(server_impl.Beat, time=T0, velocity=100)
# def test_velocity(self):
# beat = self.pool.create(model.Beat, time=T0, velocity=100)
beat.velocity = 120
self.assertEqual(beat.velocity, 120)
# beat.velocity = 120
# self.assertEqual(beat.velocity, 120)

View File

@ -29,11 +29,11 @@ from PyQt5 import QtGui
from noisicaa.core.typing_extra import down_cast
from noisicaa import audioproc
from noisicaa import model
from noisicaa import value_types
from noisicaa.ui.track_list import measured_track_editor
from noisicaa.ui.track_list import tools
from . import commands
from . import client_impl
from . import model
logger = logging.getLogger(__name__)
@ -122,12 +122,12 @@ class BeatMeasureEditor(measured_track_editor.MeasureEditor):
self.__ghost_time = None # type: audioproc.MusicalDuration
@property
def track(self) -> client_impl.BeatTrack:
return down_cast(client_impl.BeatTrack, super().track)
def track(self) -> model.BeatTrack:
return down_cast(model.BeatTrack, super().track)
@property
def measure(self) -> client_impl.BeatMeasure:
return down_cast(client_impl.BeatMeasure, super().measure)
def measure(self) -> model.BeatMeasure:
return down_cast(model.BeatMeasure, super().measure)
def xToTime(self, x: int) -> audioproc.MusicalTime:
return audioproc.MusicalDuration(
@ -234,15 +234,15 @@ class BeatTrackEditor(measured_track_editor.MeasuredTrackEditor):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__play_last_pitch = None # type: model.Pitch
self.__play_last_pitch = None # type: value_types.Pitch
self.setHeight(60)
@property
def track(self) -> client_impl.BeatTrack:
return down_cast(client_impl.BeatTrack, super().track)
def track(self) -> model.BeatTrack:
return down_cast(model.BeatTrack, super().track)
def playNoteOn(self, pitch: model.Pitch) -> None:
def playNoteOn(self, pitch: value_types.Pitch) -> None:
self.playNoteOff()
# TODO: use messages instead

View File

@ -22,9 +22,7 @@ add_python_package(
node_description.py
model.py
commands.py
client_impl.py
client_impl_test.py
server_impl.py
server_impl_test.py
node_ui.py
track_ui.py

View File

@ -1,44 +0,0 @@
#!/usr/bin/python3
# @begin:license
#
# Copyright (c) 2015-2019, 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 Sequence
from noisicaa import audioproc
from noisicaa.music import project_client_model
from . import model
class ControlPoint(
project_client_model.ProjectChild, model.ControlPoint, project_client_model.ObjectBase):
@property
def time(self) -> audioproc.MusicalTime:
return self.get_property_value('time')
@property
def value(self) -> float:
return self.get_property_value('value')
class ControlTrack(project_client_model.Track, model.ControlTrack):
@property
def points(self) -> Sequence[ControlPoint]:
return self.get_property_value('points')

View File

@ -23,13 +23,13 @@
from noisidev import unittest
from noisicaa import audioproc
from noisicaa.music import base_track_test
from . import server_impl
from . import model
from . import commands
class ControlTrackTest(base_track_test.TrackTestMixin, unittest.AsyncTestCase):
node_uri = 'builtin://control-track'
track_cls = server_impl.ControlTrack
track_cls = model.ControlTrack
async def test_add_control_point(self):
track = await self._add_track()

View File

@ -23,11 +23,11 @@
from noisicaa import audioproc
from noisicaa import music
from noisicaa.builtin_nodes import commands_registry_pb2
from . import client_impl
from . import model
def create_control_point(
track: client_impl.ControlTrack, *,
track: model.ControlTrack, *,
time: audioproc.MusicalTime,
value: float
) -> music.Command:
@ -39,7 +39,7 @@ def create_control_point(
return cmd
def update_control_point(
point: client_impl.ControlPoint, *,
point: model.ControlPoint, *,
set_time: audioproc.MusicalTime = None,
set_value: float = None
) -> music.Command:
@ -52,7 +52,7 @@ def update_control_point(
pb.set_value = set_value
return cmd
def delete_control_point(point: client_impl.ControlPoint) -> music.Command:
def delete_control_point(point: model.ControlPoint) -> music.Command:
cmd = music.Command(command='delete_control_point')
pb = cmd.Extensions[commands_registry_pb2.delete_control_point]
pb.point_id = point.id

View File

@ -20,43 +20,225 @@
#
# @end:license
from typing import Any
import logging
import random
from typing import cast, Any, Dict, MutableSequence, Optional, Callable
from noisicaa.core.typing_extra import down_cast
from noisicaa import core
from noisicaa import audioproc
from noisicaa import node_db
from noisicaa import model
from noisicaa import model_base
from noisicaa.music import base_track
from noisicaa.music import node_connector
from noisicaa.music import commands
from noisicaa.music import model
from noisicaa.builtin_nodes import model_registry_pb2
from noisicaa.builtin_nodes import commands_registry_pb2
from . import commands_pb2
from . import node_description
from . import processor_messages
logger = logging.getLogger(__name__)
class CreateControlPoint(commands.Command):
proto_type = 'create_control_point'
proto_ext = commands_registry_pb2.create_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.CreateControlPoint, self.pb)
track = down_cast(ControlTrack, self.pool[pb.track_id])
insert_time = audioproc.MusicalTime.from_proto(pb.time)
for insert_index, point in enumerate(track.points):
if point.time == insert_time:
raise ValueError("Duplicate control point")
if point.time > insert_time:
break
else:
insert_index = len(track.points)
control_point = self.pool.create(ControlPoint, time=insert_time, value=pb.value)
track.points.insert(insert_index, control_point)
class UpdateControlPoint(commands.Command):
proto_type = 'update_control_point'
proto_ext = commands_registry_pb2.update_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateControlPoint, self.pb)
point = down_cast(ControlPoint, self.pool[pb.point_id])
if pb.HasField('set_time'):
new_time = audioproc.MusicalTime.from_proto(pb.set_time)
if not point.is_first:
if new_time <= cast(ControlPoint, point.prev_sibling).time:
raise ValueError("Control point out of order.")
else:
if new_time < audioproc.MusicalTime(0, 4):
raise ValueError("Control point out of order.")
if not point.is_last:
if new_time >= cast(ControlPoint, point.next_sibling).time:
raise ValueError("Control point out of order.")
point.time = new_time
if pb.HasField('set_value'):
# TODO: check that value is in valid range.
point.value = pb.set_value
class DeleteControlPoint(commands.Command):
proto_type = 'delete_control_point'
proto_ext = commands_registry_pb2.delete_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.DeleteControlPoint, self.pb)
point = down_cast(ControlPoint, self.pool[pb.point_id])
track = down_cast(ControlTrack, point.parent)
del track.points[point.index]
class ControlTrackConnector(node_connector.NodeConnector):
_node = None # type: ControlTrack
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__node_id = self._node.pipeline_node_id
self.__listeners = {} # type: Dict[str, core.Listener]
self.__point_ids = {} # type: Dict[int, int]
def _init_internal(self) -> None:
for point in self._node.points:
self.__add_point(point)
self.__listeners['points'] = self._node.points_changed.add(
self.__points_list_changed)
def close(self) -> None:
for listener in self.__listeners.values():
listener.remove()
self.__listeners.clear()
super().close()
def __points_list_changed(self, change: model_base.PropertyChange) -> None:
if isinstance(change, model_base.PropertyListInsert):
self.__add_point(change.new_value)
elif isinstance(change, model_base.PropertyListDelete):
self.__remove_point(change.old_value)
else:
raise TypeError("Unsupported change type %s" % type(change))
def __add_point(self, point: 'ControlPoint') -> None:
point_id = self.__point_ids[point.id] = random.getrandbits(64)
self._emit_message(processor_messages.add_control_point(
node_id=self.__node_id,
id=point_id,
time=point.time,
value=point.value))
self.__listeners['cp:%s:time' % point.id] = point.time_changed.add(
lambda _: self.__point_changed(point))
self.__listeners['cp:%s:value' % point.id] = point.value_changed.add(
lambda _: self.__point_changed(point))
def __remove_point(self, point: 'ControlPoint') -> None:
point_id = self.__point_ids[point.id]
self._emit_message(processor_messages.remove_control_point(
node_id=self.__node_id,
id=point_id))
self.__listeners.pop('cp:%s:time' % point.id).remove()
self.__listeners.pop('cp:%s:value' % point.id).remove()
def __point_changed(self, point: 'ControlPoint') -> None:
point_id = self.__point_ids[point.id]
self._emit_message(processor_messages.remove_control_point(
node_id=self.__node_id,
id=point_id))
self._emit_message(processor_messages.add_control_point(
node_id=self.__node_id,
id=point_id,
time=point.time,
value=point.value))
class ControlPoint(model.ProjectChild):
class ControlPointSpec(model.ObjectSpec):
class ControlPointSpec(model_base.ObjectSpec):
proto_type = 'control_point'
proto_ext = model_registry_pb2.control_point
time = model.WrappedProtoProperty(audioproc.MusicalTime)
value = model.Property(float)
time = model_base.WrappedProtoProperty(audioproc.MusicalTime)
value = model_base.Property(float)
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.time_changed = core.Callback[model.PropertyChange[audioproc.MusicalTime]]()
self.value_changed = core.Callback[model.PropertyChange[float]]()
self.time_changed = core.Callback[model_base.PropertyChange[audioproc.MusicalTime]]()
self.value_changed = core.Callback[model_base.PropertyChange[float]]()
def create(
self, *,
time: Optional[audioproc.MusicalTime] = None, value: float = None,
**kwargs: Any) -> None:
super().create(**kwargs)
self.time = time
self.value = value
@property
def time(self) -> audioproc.MusicalTime:
return self.get_property_value('time')
@time.setter
def time(self, value: audioproc.MusicalTime) -> None:
self.set_property_value('time', value)
@property
def value(self) -> float:
return self.get_property_value('value')
@value.setter
def value(self, value: float) -> None:
self.set_property_value('value', value)
class ControlTrack(model.Track):
class ControlTrackSpec(model.ObjectSpec):
class ControlTrack(base_track.Track):
class ControlTrackSpec(model_base.ObjectSpec):
proto_type = 'control_track'
proto_ext = model_registry_pb2.control_track
points = model.ObjectListProperty(ControlPoint)
points = model_base.ObjectListProperty(ControlPoint)
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.points_changed = core.Callback[model.PropertyListChange[ControlPoint]]()
self.points_changed = core.Callback[model_base.PropertyListChange[ControlPoint]]()
@property
def description(self) -> node_db.NodeDescription:
return node_description.ControlTrackDescription
@property
def points(self) -> MutableSequence[ControlPoint]:
return self.get_property_value('points')
def create_node_connector(
self,
message_cb: Callable[[audioproc.ProcessorMessage], None],
audioproc_client: audioproc.AbstractAudioProcClient,
) -> ControlTrackConnector:
return ControlTrackConnector(
node=self, message_cb=message_cb, audioproc_client=audioproc_client)

View File

@ -1,214 +0,0 @@
#!/usr/bin/python3
# @begin:license
#
# Copyright (c) 2015-2019, 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 logging
import random
from typing import cast, Any, Dict, MutableSequence, Optional, Callable
from noisicaa.core.typing_extra import down_cast
from noisicaa import audioproc
from noisicaa import model
from noisicaa import core
from noisicaa.music import commands
from noisicaa.music import pmodel
from noisicaa.music import node_connector
from noisicaa.music import base_track
from noisicaa.builtin_nodes import commands_registry_pb2
from . import commands_pb2
from . import model as control_track_model
from . import processor_messages
logger = logging.getLogger(__name__)
class CreateControlPoint(commands.Command):
proto_type = 'create_control_point'
proto_ext = commands_registry_pb2.create_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.CreateControlPoint, self.pb)
track = down_cast(ControlTrack, self.pool[pb.track_id])
insert_time = audioproc.MusicalTime.from_proto(pb.time)
for insert_index, point in enumerate(track.points):
if point.time == insert_time:
raise ValueError("Duplicate control point")
if point.time > insert_time:
break
else:
insert_index = len(track.points)
control_point = self.pool.create(ControlPoint, time=insert_time, value=pb.value)
track.points.insert(insert_index, control_point)
class UpdateControlPoint(commands.Command):
proto_type = 'update_control_point'
proto_ext = commands_registry_pb2.update_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.UpdateControlPoint, self.pb)
point = down_cast(ControlPoint, self.pool[pb.point_id])
if pb.HasField('set_time'):
new_time = audioproc.MusicalTime.from_proto(pb.set_time)
if not point.is_first:
if new_time <= cast(ControlPoint, point.prev_sibling).time:
raise ValueError("Control point out of order.")
else:
if new_time < audioproc.MusicalTime(0, 4):
raise ValueError("Control point out of order.")
if not point.is_last:
if new_time >= cast(ControlPoint, point.next_sibling).time:
raise ValueError("Control point out of order.")
point.time = new_time
if pb.HasField('set_value'):
# TODO: check that value is in valid range.
point.value = pb.set_value
class DeleteControlPoint(commands.Command):
proto_type = 'delete_control_point'
proto_ext = commands_registry_pb2.delete_control_point
def run(self) -> None:
pb = down_cast(commands_pb2.DeleteControlPoint, self.pb)
point = down_cast(ControlPoint, self.pool[pb.point_id])
track = down_cast(ControlTrack, point.parent)
del track.points[point.index]
class ControlTrackConnector(node_connector.NodeConnector):
_node = None # type: ControlTrack
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.__node_id = self._node.pipeline_node_id
self.__listeners = {} # type: Dict[str, core.Listener]
self.__point_ids = {} # type: Dict[int, int]
def _init_internal(self) -> None:
for point in self._node.points:
self.__add_point(point)
self.__listeners['points'] = self._node.points_changed.add(
self.__points_list_changed)
def close(self) -> None:
for listener in self.__listeners.values():
listener.remove()
self.__listeners.clear()
super().close()
def __points_list_changed(self, change: model.PropertyChange) -> None:
if isinstance(change, model.PropertyListInsert):
self.__add_point(change.new_value)
elif isinstance(change, model.PropertyListDelete):
self.__remove_point(change.old_value)
else:
raise TypeError("Unsupported change type %s" % type(change))
def __add_point(self, point: 'ControlPoint') -> None: