Skip to content
This repository was archived by the owner on Apr 26, 2023. It is now read-only.

facilities tutorial developing a monitor

Jump to bottom
Pablo Conesa edited this page Jun 18, 2018 · 25 revisions

Table of contents generated with markdown-toc

What is a monitor

A monitor is a non-GUI class which main action would be to do something, periodically and notify any possible "listeners".

Monitors anatomy

The base monitor class looks like this:

class Monitor():
    def __init__(self, **kwargs):
        # Where to store any data from this monitor
        self.workingDir = kwargs['workingDir']
        self.samplingInterval = kwargs.get('samplingInterval', None)
        self.monitorTime = kwargs.get('monitorTime', None)

        self._notifiers = []

        if kwargs.get('email', None) is not None:
            self._notifiers.append(kwargs['email'])

        if 'stdout' in kwargs:
            self._notifiers.append(PrintNotifier())

    def notify(self, title, message):
        for n in self._notifiers:
            if n: 
                n.notify(title, message)

    def info(self, message):
        self.notify("INFO", message)

    def initLoop(self):
        """ To be defined in subclasses. """
        pass

    def loop(self):
        self.initLoop()
        timeout = time.time() + 60. * self.monitorTime   # interval minutes from now

        while True:
            finished = self.step()
            if (time.time() > timeout) or finished:
                break
            time.sleep(self.samplingInterval)

    def step(self):
        """ To be defined in subclasses. """
        pass

    def addNotifier(self, notifier):
        self._notifiers.append(notifier)

The most important method of a monitor is

    def step(self):
        """ To be defined in subclasses. """
        pass

In this base class it is not implemented, but if you want to develop a monitor, better implement this method. The step() method will be called from the loop() in a regular manner until a self.monitorTime is reached or it is intentionally stoped.

Optionally, you can code and initLoop() method that will be called before the loops start, suitable for some initializations.

Additionally, monitors follow and observer pattern where any interested "listener/observer" could register and receive notifications from the monitor.

Developing a monitor to observe free space left in the HD where the project is placed.

Let's start coding something. Let's create a python file in our tutorial package to have and create there our new monitor class.

First iteration

  1. Add a mymonitor.py file to pyworkflow/em/packages/myfacility/
  2. Import monitor base class, and define your new class based on base monitor
from  pyworkflow.em.protocol.monitors import Monitor


class SpaceMonitor(Monitor):
    """
    Monitor to monitor free space on the HD where scipion project is placed
    """
  1. Implement the step method. This method should find the HD where the project is placed.

Import some modules at the top of the file (import section)

import os
import collections

Add the following to the step method:

    def step(self):
        """ Using the workingdir attribute has to find the HD and then get the
        available free space."""

        usage = disk_usage(self.workingDir)

        print(usage)



# Taken from http://code.activestate.com/recipes/577972-disk-usage/
def disk_usage(path):
    _ntuple_diskusage = collections.namedtuple('usage', 'total used free')

    st = os.statvfs(path)
    free = st.f_bavail * st.f_frsize
    total = st.f_blocks * st.f_frsize
    used = (st.f_blocks - st.f_bfree) * st.f_frsize
    return _ntuple_diskusage(total, used, free)

This is adding a new method disk_usage which receives a path and uses it in the step method, printing it (temporarily).

Testing first iteration

We are going now to add some test to test our progress.

  1. Add a tests folder to the test folder of the package (myfacility/tests)
  2. Add an empty __init__.py to make it a module
  3. Add test_monitor.py to the tests folder
  4. Add the code below
from pyworkflow.tests import *
from pyworkflow.em.packages.myfacility import SpaceMonitor

# Test monitor functionality
class TestMonitor(BaseTest):

    def test_monitor(self):

        # Instantiate the monitor
        spaceMonitor = SpaceMonitor(workingDir=os.getcwd())

        spaceMonitor.step()
  1. Run the test: scipion test em.packages.myfacility.tests.test_monitor.TestMonitor

Output should look like this:

Scipion  (2018-06-12)  ((HEAD detached at june_2018_course) b9d0e37)

>>>>> python  scripts/run_tests.py em.packages.myfacility.tests.test_monitor.TestMonitor
Running tests....
usage(total=245527773184, used=127268225024, free=105763827712)
[ RUN   OK ] TestMonitor.test_monitor (0.001 secs)

[==========] run 1 tests (0.001 secs)
[  PASSED  ] 1 tests

The disk stats are printed on the screen. This is not a proper test, since it will never fail and there are no checks(assertions) but soon we will add them.

2nd iteration: persist the values in a txt file

We need to persist/store the values, and the easiest approach here is to write the values in a text file.

Test first

Following a TDD approach we will first modify the test to expect the new behaviour. Since we are going to generate files, let's tell the monitor to use a temporary folder instead of the os.getcwd(). For this we need to first import at the top mkdtemp function from tempfile from tempfile import mkdtemp and then use it when creating the monitor:

    def test_monitor(self):

        # Instantiate the monitor
        spaceMonitor = SpaceMonitor(workingDir=mkdtemp())

        spaceMonitor.step()

        # Check storage file exists
        fnStorageFile = spaceMonitor.getStorageFilePath()

        # Check that the file exists
        self.assertTrue(os.path.exists(fnStorageFile),
                        "Storage file %s not created." % fnStorageFile)

        # Check there are 2 lines (headers and first data line)
        num_lines = sum(1 for line in open(fnStorageFile,'r'))

        # Assert lines are 2
        self.assertEqual(2, num_lines,
                         "First step of the monitor does not "
                         "have the expected lines: %s" % 2)

Additionally we have added some lines to get the name of the storage file and assert that it exists and to test that is has 2 lines. Since we havent modified our Monitor yet this test should fail.

[   FAILED ] TestMonitor.test_monitor

Traceback (most recent call last):
  File "/home/pablo/desarrollo/scipion/software/lib/python2.7/unittest/case.py", line 329, in run
    testMethod()
  File "/home/pablo/desarrollo/scipion/pyworkflow/em/packages/myfacility/tests/test_monitor.py", line 16, in test_monitor
    fnStorageFile = spaceMonitor.getStorageFilePath()
AttributeError: SpaceMonitor instance has no attribute 'getStorageFilePath'

[==========] run 1 tests (0.003 secs)
[  FAILED  ] 1 tests
[  PASSED  ] 0 tests

Our monitor does not have the getStorageFilePath() function.

Monitor second

Let's implement what the test is expecting:

    def step(self):
        """ Using the workingdir attribute has to find the HD and then get the
        available free space."""

        usage = disk_usage(self.workingDir)

        self.storeUsageData(usage)

    def getStorageFilePath(self):
        return os.path.join(self.workingDir, 'space_usage.txt')

    def storeUsageData(self, usageData):

        fnStorageFile = self.getStorageFilePath()

        if not os.path.exists(fnStorageFile):
            fhStorage = open(fnStorageFile, "w")
            fhStorage.write("total\tused\tfree\n")
        else:
            fhStorage = open(fnStorageFile, "a")

        fhStorage.write("%s\t%s\t%s\n" % (usageData.total, usageData.used, usageData.free))

        fhStorage.close()

Note that we have:

  1. Implemented the getStorageFilePath() method.
  2. Implemented a storeUsageData() to store our usage data
  3. In the step() function we have called the storage function: self.storeUsageData(usage)
Clone this wiki locally