Python Examples

The following examples illustrate some ideas on how one could use MSL-Network.

  1. Digital Multimeter

  2. Additional (Runnable) Examples

  3. RPi-SmartGadget – Uses a Raspberry Pi to communicate with a Sensirion SHTxx sensor.

Digital Multimeter

This example shows how a digital multimeter that has a non-Ethernet interface, e.g., GPIB or RS232, can be controlled from any computer that is on the network. It uses the MSL-Equipment package to connect to the digital multimeter and MSL-Network to enable the digital multimeter as a Service on the network. This example is included with MSL-Network when it is installed, but since it requires additional hardware (a digital multimeter) it can only be run if the hardware is attached to the computer.

The first task to do is to Start the Network Manager on the same computer that the digital multimeter is physically connected to (via a GPIB cable or a DB9 cable). Next, on the same computer, copy and paste the following script to a file, edit the equipment record used by MSL-Equipment for the information relevant to your DMM (e.g., the COM#, GPIB address) and then run the script to start the digital multimeter Service.

"""
Example showing how a digital multimeter that has a non-Ethernet interface
(e.g., GPIB or RS232) can be controlled from any computer that is on the network.
"""
from msl.equipment import ConnectionRecord
from msl.equipment import EquipmentRecord

from msl.network import Service


class DigitalMultimeter(Service):

    def __init__(self):
        """Initialize the communication with the digital multimeter.

        This script must be run on a computer that the multimeter is
        physically connected to.
        """

        # Initialize the Service. Set the name of the DigitalMultimeter Service,
        # as it will appear on the Network Manager, to be 'Hewlett Packard 34401A'
        # and specify that only 1 Client on the network can control the digital
        # multimeter at any instance in time. Once the Client disconnects from
        # the Network Manager another Client would then be able to link with the
        # DigitalMultimeter Service to control the digital multimeter.
        super().__init__(name='Hewlett Packard 34401A', max_clients=1)

        # Connect to the digital multimeter
        # (see MSL-Equipment for more details)
        record = EquipmentRecord(
            manufacturer='HP',
            model='34401A',
            connection=ConnectionRecord(
                address='COM4',  # RS232 interface
                backend='MSL',
            )
        )
        self._dmm = record.connect()

    def write(self, command: str) -> None:
        """Write a command to the digital multimeter.

        Parameters
        ----------
        command : str
            The command to write.
        """
        self._dmm.write(command)

    def read(self) -> str:
        """Read the response from the digital multimeter.

        Returns
        -------
        str
            The response.
        """
        return self._dmm.read().rstrip()

    def query(self, command: str) -> str:
        """Query the digital multimeter.

        Performs a write then a read.

        Parameters
        ----------
        command : str
            The command to write.

        Returns
        -------
        str
            The response.
        """
        return self._dmm.query(command).rstrip()


if __name__ == '__main__':
    # Initialize and start the DigitalMultimeter Service
    dmm_service = DigitalMultimeter()
    dmm_service.start()

With the DigitalMultimeter Service running you can execute the following commands on another computer that is on the same network as the Manager in order to interact with the digital multimeter from the remote computer.

Connect to the Manager by specifying the hostname (or IP address) of the computer that the Manager is running on

>>> from msl.network import connect
>>> cxn = connect(host='the hostname or IP address of the computer that the Manager is running on')

Since the name of the DigitalMultimeter Service was specified to be 'Hewlett Packard 34401A', we must link with the correct name of the Service

>>> dmm = cxn.link('Hewlett Packard 34401A')

Tip

The process of establishing a connection to a Manager and linking with a Service can also be done in a single line. A LinkedClient exists for this purpose. This can be useful if you only want to link with a single Service.

>>> from msl.network import LinkedClient
>>> dmm = LinkedClient('Hewlett Packard 34401A', host='hostname or IP address of the Manager')

Now we can send write, read or query commands to the digital multimeter

>>> dmm.query('MEASURE:VOLTAGE:DC?')
'-6.23954727E-02'

When you are finished sending requests to the Manager you should disconnect from the Manager. This will allow other Client's to be able to control the digital multimeter.

>>> cxn.disconnect()

Additional (Runnable) Examples

The following Service's are included with MSL-Network. To start any of these Service's, first make sure that you Start the Network Manager, and then run the following command in a terminal. For this example, the Echo Service is running

python -c "from msl.examples.network import Echo; Echo().start()"

You can then send requests to the Echo Service

>>> from msl.network import connect
>>> cxn = connect()
>>> e = cxn.link('Echo')
>>> e.echo('hello')
[['hello'], {}]
>>> e.echo('world!', x=7, array=list(range(10)))
[['world!'], {'x': 7, 'array': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]}]
>>> cxn.disconnect()

Echo Service

"""
Example echo Service.

Before running this module ensure that the Network Manager is running on the
same computer, i.e., run the following command in a terminal

msl-network start

then run this module to connect to the Manager as a Service.

After the Echo Service starts you can connect to the Manager as a Client,
link with the Echo Service and then send requests, e.g.,

from msl.network import connect
cxn = connect()
e = cxn.link('Echo')
args, kwargs = e.echo(1, 2, x='hello', y='world')
"""
from msl.network import Service


class Echo(Service):

    @staticmethod
    def echo(*args, **kwargs):
        return args, kwargs


if __name__ == '__main__':
    service = Echo()
    service.start()

BasicMath Service

"""
Example Service for illustrating the difference between synchronous and
asynchronous requests.

Before running this module ensure that the Network Manager is running on the
same computer, i.e., run the following command in a terminal

msl-network start

then run this module to connect to the Manager as a Service.

After the BasicMath Service starts you can connect to the Manager as a Client,
link with the BasicMath Service and then send requests, e.g.,

from msl.network import connect
cxn = connect()
bm = cxn.link('BasicMath')
value = bm.add(1, 2)
"""
import time
from typing import Union

from msl.network import Service

number = Union[int, float]


class BasicMath(Service):

    euler = 2.718281828459045

    @property
    def pi(self) -> float:
        return 3.141592653589793

    def add(self, x: number, y: number) -> number:
        time.sleep(1)
        return x + y

    def subtract(self, x: number, y: number) -> number:
        time.sleep(2)
        return x - y

    def multiply(self, x: number, y: number) -> number:
        time.sleep(3)
        return x * y

    def divide(self, x: number, y: number) -> number:
        time.sleep(4)
        return x / float(y)

    def ensure_positive(self, x: number) -> bool:
        time.sleep(5)
        if x < 0:
            raise ValueError('The value is < 0')
        return True

    def power(self, x: number, n=2) -> number:
        time.sleep(6)
        return x ** n


if __name__ == '__main__':
    import logging

    # Optional: allows for "info" log messages to be visible on the Service
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(levelname)-5s] %(message)s',
    )

    service = BasicMath()
    service.start()

MyArray Service

"""
Example Service for generating and manipulating arrays. This example
illustrates how to interface a LabVIEW program with MSL-Network.

Before running this module ensure that the Network Manager is running on the
same computer, i.e., run the following command in a terminal

msl-network start

then run this module to connect to the Manager as a Service.

After the MyArray Service starts you can connect to the Manager as a Client,
link with the MyArray Service and then send requests, e.g.,

from msl.network import connect
cxn = connect()
my_array = cxn.link('MyArray')
linspace = my_array.linspace(0, 1)
"""
from typing import List, Union

from msl.network import Service

number = Union[int, float]
Vector = List[float]


class MyArray(Service):

    @staticmethod
    def linspace(start: number, stop: number, n=100) -> List[float]:
        """Return evenly-spaced numbers over a specified interval."""
        dx = (stop-start)/float(n-1)
        return [start+i*dx for i in range(int(n))]

    @staticmethod
    def scalar_multiply(scalar: number, data: Vector) -> Vector:
        """Multiply every element in `data` by a number."""
        return [element*scalar for element in data]


if __name__ == '__main__':
    service = MyArray()
    service.start()

Heartbeat Service

"""
Example Service that emits notifications to all linked Clients. This example
also shows how to add a task to the event loop of the Service.

Before running this module ensure that the Network Manager is running on the
same computer, i.e., run the following command in a terminal

msl-network start

then run this module to connect to the Manager as a Service.

After the Heartbeat Service starts you can connect to the Manager as a Client,
link with the Heartbeat Service, handle notifications from the Service and also
send requests, e.g.,

import types
from msl.network import connect

def print_notification(self, *args, **kwargs):
    print(f'The {self.service_name} Service emitted', args, kwargs)

cxn = connect()
heartbeat = cxn.link('Heartbeat')
heartbeat.notification_handler = types.MethodType(print_notification, heartbeat)

# some time later

heartbeat.reset()
"""
import asyncio

from msl.network import Service


class Heartbeat(Service):

    def __init__(self):
        """A Service that emits a counter value."""
        super(Heartbeat, self).__init__()
        self._sleep = 1.0
        self._counter = 0
        self._alive = True

    def kill(self) -> None:
        """Stop emitting the heartbeat."""
        self._alive = False

    def reset(self) -> None:
        """Reset the heartbeat counter."""
        self._counter = 0

    def set_heart_rate(self, beats_per_second: int) -> None:
        """Change the rate that the value of the counter is emitted."""
        self._sleep = 1.0 / float(beats_per_second)

    def shutdown_handler(self) -> None:
        """Called when the connection to the Manager is closed."""
        self._alive = False

    async def emit(self) -> None:
        """This coroutine is also run in the event loop."""
        while self._alive:
            self.emit_notification(self._counter)
            self._counter += 1
            await asyncio.sleep(self._sleep)


if __name__ == '__main__':
    # Initialize the Service
    service = Heartbeat()

    # Add a task to the event loop of the Service
    service.add_tasks(service.emit())

    # Start the Service
    service.start()