Add packet creation methods to AddressNode

src/oscpython/address.py

@@ -7,6 +7,17 @@
 
 from oscpython.arguments import StringArgument
 
+messages, Message, Bundle = None, None, None
+def _import_message_classes():
+    global messages, Message, Bundle
+    if messages is not None:
+        return
+    from oscpython import messages as _msg_module
+    messages = _msg_module
+    Message = _msg_module.Message
+    Bundle = _msg_module.Bundle
+
+
 StrOrAddress = Union[str, 'Address']
 
 __all__ = ('Address', 'AddressPart', 'AddressSpace', 'AddressNode')
@@ -243,6 +254,12 @@ def __len__(self):
     def __iter__(self):
         yield from self.parts
 
+    def copy(self) -> 'Address':
+        """Create a copy of the instance
+        """
+        cls = self.__class__
+        return cls.from_parts(self.parts)
+
     def join(self, other) -> 'Address':
         """Join the address with either a str or :class:`Address` instance,
         separating the :attr:`pattern` with ``"/"``
@@ -642,6 +659,28 @@ def walk(self) -> Iterator['AddressNode']:
         for child in self:
             yield from child.walk()
 
+    def create_message(self, *args, **kwargs) -> 'oscpython.messages.Message':
+        """Create a :class:`~.messages.Message` using this node's :attr:`address`
+
+        Positional and keyword arguments (``*args`` and ``**kwargs``)
+        will be passed directly to :meth:`.messages.Message.create`
+        """
+        _import_message_classes()
+        addr = self.address.copy()
+        return Message.create(addr, *args, **kwargs)
+
+    def create_bundled_message(self, *args, **kwargs) -> 'oscpython.messages.Bundle':
+        """Create a :class:`~.messages.Bundle` containing a :class:`~.messages.Message`
+        using this node's :attr:`address`
+
+        Keyword arguments (``**kwargs``) will be passed to the
+        Bundle's ``__init__`` method and positional arguments (``*args``)
+        will be passed to :meth:`.messages.Message.create`
+        """
+        msg = self.create_message(*args)
+        bun = Bundle(**kwargs)
+        bun.add_packet(msg)
+        return bun
 
     def add_child(self, name: str, cls: Optional[type] = None) -> 'AddressNode':
         """Add a child node to this point in the tree

tests/test_address_node.py

@@ -1,5 +1,6 @@
 import pytest
 from oscpython.address import Address, AddressSpace, AddressNode
+from oscpython import Client, TimeTag
 
 def test_add_and_reparent():
     sp = AddressSpace()
@@ -212,3 +213,30 @@ def test_node_tree(message_addresses):
 
 
     assert len(message_addresses) ==  len(all_addr_nodes)
+
+def test_packet_creation(message_addresses, random_arguments, faker):
+    sp = AddressSpace()
+
+    base_tt = TimeTag.now()
+
+    for i, addr in enumerate(message_addresses):
+        root, node = sp.create_from_address(addr)
+        client = Client(address=faker.ipv4(), port=faker.port_number())
+        args = tuple([next(random_arguments) for _ in range(3)])
+        arg_vals = tuple([arg.value for arg in args])
+
+        msg = node.create_message(*args, remote_client=client)
+        assert msg.remote_client == client
+        assert msg.address == node.address
+        assert msg.address is not node.address
+        assert tuple([arg.value for arg in msg]) == arg_vals
+
+        tt = TimeTag(seconds=base_tt.seconds + i, fraction=base_tt.fraction)
+        bun = node.create_bundled_message(*args, remote_client=client, timetag=tt)
+        assert bun.timetag == tt
+        assert bun.remote_client == client
+        assert len(bun) == 1
+        bun_msg = bun[0]
+        assert bun_msg.address == node.address
+        assert bun_msg.address is not node.address
+        assert tuple([arg.value for arg in bun_msg]) == arg_vals