Creating custom blocks

Estimated time: 15–25 minutes

The nio Block Library contains many useful blocks for all types of nio projects. However, sometimes you may discover that a custom block would be more suitable for your project. This tutorial walks you through the creation and testing of a custom block.

This tutorial shows you how to create and test a custom block for your nio projects. The block that you will develop is simple; however, you will learn several important concepts that will aid you in creating your own blocks. The custom block in this example will receive signals through its input terminal, multiply the values by a number specified in the block properties, and emit the new value as a signal through the output terminal.

[info] Prerequisites


Create Multiplier block

From the blocks directory of your project, add the new block as a submodule named multiplier using the nio CLI newblock command.

nio newblock multiplier

Find your block in the System Designer

  1. From the root of the project directory, type niod to start the instance.
  2. Open the System Designer in your web browser.
  3. Open the system you previously created and open your connected local instance.
  4. Add a new service, name it, and click accept.
  5. In the installed tab of the block library, search for the Multiplier block.
  6. Drag the block to the canvas, name it Double, and click accept.

At this point, the block has no functionality. It will pass incoming signals directly through to the output terminal unchanged. The next step will add functionality to the block. To do this properly, we will implement test-driven development.


Write a block test

  1. In the project directory, type Ctrl + C to shut down the instance.
  2. Type cd blocks/multiplier to navigate back to the block directory.
  3. Type pip3 install -U pytest to download the pytest.
  4. Type py.test from the block directory. This test will pass because the block simply passes unmodified signals.
  5. Update the test file to match the following code:

    
     from nio.block.terminals import DEFAULT_TERMINAL
     from nio.signal.base import Signal
     from nio.testing.block_test_case import NIOBlockTestCase
     from ..multiplier_block import Multiplier
    
     class TestMultiplier(NIOBlockTestCase):
    
     def test_process_signals(self):
         """Signals are multiplied."""
         blk = Multiplier()
         self.configure_block(blk, {'multiplier': 2, 'number': ""})
         blk.start()
         blk.process_signals([Signal({"num": 4})])
         blk.stop()
         self.assert_num_signals_notified(1)
         self.assertDictEqual(
             self.last_notified[DEFAULT_TERMINAL][0].to_dict(),
             {"num": 4, "multiplied": 8})
    

    This test expects the block to be configured with two integer properties that set the multiplier to 2 and the number to multiply to the num attribute of the incoming signal. When the block starts, a signal will be simulated that has a num attribute with a value of 4. After the block stops, the test will check that 1 signal passed through the block and it contained a new multiplied attribute with a value of 8 (2*4).

  6. Type py.test The test fails because the block returns what was passed into it {"num": 4}, but the test expects a new multiplied value to be added to the outgoing signal. Now, you need to update the block code to implement this functionality.

[info] Test Driven Development

Next, you need to write the test for how you expect the block to behave, see the test fail with the current block implementation, and then update the block code to make the test pass.


Write block code

  1. From the root of the project directory, open the multiplier_block.py file.
  2. You need to make several additions to the code:
    • To create the configurable properties, the block needs to import IntProperty after VersionProperty.
    • The Multiplier class needs to create variables for these properties.
    • The process_signals method needs to multiply the num attribute of each signal by the multiplier value and append it to the outgoing signal.
  3. Update the block file to match the following code:

    
     from nio.block.base import Block
     from nio.properties import VersionProperty, IntProperty
    
     class Multiplier(Block):
    
       version = VersionProperty('0.1.0')
       multiplier = IntProperty(default=1, title="Multiplier Value")
       number = IntProperty(title="Number to Multiply")
    
       def process_signals(self, signals):
           for signal in signals:
               multiplied = self.multiplier() * self.number(signal)
               setattr(signal, "multiplied", multiplied)
           self.notify_signals(signals)
    
  4. Run py.test again to confirm that the block works as expected.


Test block

  1. From the root of the project, type niod to start the instance.
  2. In the System Designer, open the local instance within your service.
  3. Open the service you created earlier in this tutorial.
  4. The Multiplier block contains two configurable properties: Multiplier Value and Number to Multiply.
  5. Add a CounterIntervalSimulator and Logger block into the service and save the default configurations.
  6. Configure the Double block:
    • Multiplier Value: 2
    • Number to Multiply: {{ $sim }}
  7. Configure the service to match the diagram at right:
  8. Start the service.
  9. Check the logger panel to see how the block behaves.

 


Add signal enrichment

This method for modifying a signal is great, but there may be times where you would rather replace the attributes of the incoming signal with the new or modified attributes created by your block. This is where the enrich mixin comes in handy!

Update the test file to match the following code:


from nio.block.terminals import DEFAULT_TERMINAL
from nio.signal.base import Signal
from nio.testing.block_test_case import NIOBlockTestCase
from ..multiplier_block import Multiplier

class TestMultiplier(NIOBlockTestCase):

  def test_process_signals(self):
      """Signals are multiplied."""
      blk = Multiplier()
      self.configure_block(blk, {
          'enrich': {
              'exclude_existing': False,
          },
          'multiplier': 2,
          'number': "{{ $num }}",
      })
      blk.start()
      blk.process_signals([Signal({"num": 4})])
      blk.stop()
      self.assert_num_signals_notified(1)
      self.assertDictEqual(
          self.last_notified[DEFAULT_TERMINAL][0].to_dict(),
          {"num": 4, "multiplied": 8})

  def test_signal_enrichment(self):
      """Output signals do not contain original number."""
      blk = Multiplier()
      self.configure_block(blk, {
          'enrich': {
              'exclude_existing': True,
          },
          'multiplier': 2,
          'number': "{{ $num }}"
      })
      blk.start()
      blk.process_signals([Signal({"num": 4})])
      blk.stop()
      self.assert_num_signals_notified(1)
      self.assertDictEqual(
          self.last_notified[DEFAULT_TERMINAL][0].to_dict(),{"multiplied": 8})

The test_process_signals function now has a new exclude property set to False so the incoming signal is also part of the outgoing signal. The new test, test_signal_enrichment, has an exclude property set to True so the incoming signal is not included in the outgoing signal as shown in the last line of the test. This test only returns the multiplied attribute.

Run py.test to see the test fail.

Update multiplier_block.py with the new signal enrichment changes shown below:


  from nio.block.base import Block
  from nio.properties import VersionProperty, IntProperty
  from nio.block.mixins import EnrichSignals

  class Multiplier(Block, EnrichSignals):

      version = VersionProperty('0.1.0')
      multiplier = IntProperty(default=1, title="Multiplier Value")
      number = IntProperty(title="Number to Multiply")

      def process_signals(self, signals):
          out_sigs = []
          for sig in signals:
              multiplied = {"multiplied": self.multiplier() * self.number(sig)}
              out_sigs.append(self.get_output_signal(multiplied, sig))
          self.notify_signals(out_sigs)

This new block file has a few changes.

  • There is a new import for the EnrichSignals mixin which is added as a class that Multiplier inherits from.
  • A new array for outgoing signals (out_sigs) is created for all modified signals to be appended to.
  • Finally, the new multiplied attribute is created as a JSON object and passed through the get_output_signal method. This method handles the logic for appending the new signal to the existing one or excluding the incoming signals' attributes.

Start the instance with niod and check the System Designer to see the new configuration option and how it affects the service.

results matching ""

    No results matching ""