Merge pull request #1610 from phiros/feature_pytermcontroller

dist/pyterm: control server incl. support for multiple testbeds

dist/tools/pyterm/pyterm

@@ -75,7 +75,8 @@ class SerCmd(cmd.Cmd):
     """
 
     def __init__(self, port=None, baudrate=None, tcp_serial=None,
-                 confdir=None, conffile=None, host=None, run_name=None):
+                 confdir=None, conffile=None, host=None, run_name=None, 
+                 log_dir_name=None):
         """Constructor.
 
         Args:
@@ -98,6 +99,7 @@ class SerCmd(cmd.Cmd):
         self.configfile = conffile
         self.host = host
         self.run_name = run_name
+        self.log_dir_name = log_dir_name
 
         if not self.host:
             self.host = defaulthostname
@@ -105,6 +107,9 @@ class SerCmd(cmd.Cmd):
         if not self.run_name:
             self.run_name = defaultrunname
 
+        if not self.log_dir_name:
+            self.log_dir_name = self.hostname
+
         if not os.path.exists(self.configdir):
             os.makedirs(self.configdir)
 
@@ -132,7 +137,7 @@ class SerCmd(cmd.Cmd):
         # create formatter
         formatter = logging.Formatter(self.fmt_str)
 
-        directory = self.configdir + os.path.sep + self.host
+        directory = self.configdir + os.path.sep + self.log_dir_name
         if not os.path.exists(directory):
             os.makedirs(directory)
         logging.basicConfig(filename = directory + os.path.sep +
@@ -579,21 +584,33 @@ class SerCmd(cmd.Cmd):
             #sys.stdout.flush()
 
 class PytermProt(Protocol):
+    def __init__(self, factory):
+        self.factory = factory
+        
+    def connectionMade(self):
+        print("writing to transport")
+        self.transport.write("hostname: %s\n" % (self.factory.shell.host))
+    
     def dataReceived(self, data):
         sys.stdout.write(data)
+        if(data.strip() == "/exit"):
+            reactor.callLater(2, self.factory.shell.do_PYTERM_exit, data)
+        else:
+            self.factory.shell.ser.write(data + "\n")               
 
     def sendMessage(self, msg):
         self.transport.writeSomeData("%d#%s\n" % (len(msg), msg))
 
 class PytermClientFactory(ReconnectingClientFactory):
 
-    def __init__(self):
+    def __init__(self, shell = None):
         self.myproto = None
+        self.shell = shell
 
     def buildProtocol(self, addr):
         print('Connected.')
         self.resetDelay()
-        self.myproto = PytermProt()
+        self.myproto = PytermProt(self)
         return self.myproto
 
     def clientConnectionLost(self, connector, reason):
@@ -655,14 +672,19 @@ if __name__ == "__main__":
             help="Hostname of this maschine")
     parser.add_argument("-rn", "--run-name",
             help="Run name, used for logfile")
+    parser.add_argument("-ln", "--log-dir-name",
+            help="Log directory name (default is hostname e.g. %s/<hostname>)"
+                 %defaultdir,
+                 default=defaultdir)
     args = parser.parse_args()
 
     myshell = SerCmd(args.port, args.baudrate, args.tcp_serial,
-                     args.directory, args.config, args.host, args.run_name)
+                     args.directory, args.config, args.host, args.run_name,
+                     args.log_dir_name)
     myshell.prompt = ''
 
     if args.server and args.tcp_port:
-        myfactory = PytermClientFactory()
+        myfactory = PytermClientFactory(myshell)
         reactor.connectTCP(args.server, args.tcp_port, myfactory)
         myshell.factory = myfactory
         reactor.callInThread(myshell.cmdloop, "Welcome to pyterm!\n"

dist/tools/pyterm/pytermcontroller/__init__.py

+__all__ = ["pytermcontroller"]
+

dist/tools/pyterm/pytermcontroller/pytermcontroller.py

+#!/usr/bin/env python2
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2014  Philipp Rosenkranz <philipp.rosenkranz@fu-berlin.de>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301 USA
+
+
+import sys, signal, threading
+
+from twisted.internet import reactor
+from twisted.internet.protocol import Protocol, Factory
+
+
+class PubProtocol(Protocol):
+    def __init__(self, factory):
+        self.factory = factory
+
+    def connectionMade(self):
+        print("new connection made")   
+
+    def connectionLost(self, reason):
+        self.factory.numProtocols = self.factory.numProtocols - 1
+
+    def connectionLost(self, reason):
+        self.factory.clients = {key: value for key, value in self.factory.clients.items() 
+             if value is not self.transport}
+
+    def dataReceived(self, data):
+        if data.startswith("hostname: "):
+            remoteHostname = data.split()[1].strip()
+            print("dataReceived added: " + remoteHostname)
+            self.factory.clients[remoteHostname] = self.transport
+        else:
+            print("received some useless data...")
+
+class PubFactory(Factory):
+    def __init__(self):
+        self.clients = dict()
+
+    def buildProtocol(self, addr):
+        return PubProtocol(self)
+  
+    
+class ExperimentRunner():
+    def __init__(self, experiment, testbed):
+        self.publisher = PubFactory()
+        self.port = int(testbed.serverPort)
+        self.experiment = experiment(self.publisher, self)
+        self.testbed = testbed
+    
+    def run(self):
+        signal.signal(signal.SIGINT, self.handle_sigint)
+        self.experiment.run()
+        reactor.listenTCP(self.port, self.publisher)
+        # clean logfiles and start nodes but don't flash nodes
+        self.testbed.initClean() 
+        reactor.run() 
+        
+    def stop(self):
+        self.testbed.stop()
+        reactor.callFromThread(self.safeReactorStop)
+            
+    def safeReactorStop(self):
+        if reactor.running:
+            try:
+                reactor.stop()
+            except:
+                print("tried to shutdown reactor twice!")
+                
+        
+    def handle_sigint(self, signal, frame):        
+        self.experiment.stop()
+        self.testbed.stop()
+        self.stop() # shutdown if experiment didn't already
+         
+
+class Experiment():
+    def __init__(self, factory, runner):
+        self.factory = factory
+        self.runner = runner
+        self.hostid = dict()
+        self.sumDelay = 0
+        
+    def run(self):
+        print("Running preHook")
+        self.preHook()
+        print("Running experiment")
+        self.start()        
+
+    def start(self):
+        raise NotImplementedError("Inherit from Experiment and implement start")  
+    
+    def stop(self): 
+        print("Running postHook")       
+        self.postHook()  
+        self.runner.stop()       
+      
+    def preHook(self):
+        pass   
+    
+    def postHook(self):
+        pass 
+ 
+    def readHostFile(self, path):
+        id = 1
+        with open(path) as f:
+            for line in f:
+                self.hostid[line.strip()] = id
+                id += 1
+    
+    def sendToHost(self, host=None, cmd=""):
+        if host in self.factory.clients:
+            self.factory.clients[host].write(cmd + "\n")
+        else:
+            print("sendToHost: no such host known: " + host + " !")
+    
+    def sendToAll(self, cmd=""):
+       for host, transport in self.factory.clients.items():
+            self.sendToHost(host, cmd)        
+         
+    def pauseInSeconds(self, seconds=0):
+        from time import time, sleep
+        start = time()
+        while (time() - start < seconds):
+            sleep(seconds - (time() - start))
+            
+    def callLater(self, absoluteDelay = 0.0, function = None):
+        reactor.callLater(absoluteDelay, function)
+        
+    def waitAndCall(self, relativeDelay = 0.0, function = None):
+        self.sumDelay += relativeDelay
+        self.callLater(self.sumDelay, function)
+        
+    def clientIterator(self):
+        return self.factory.clients.items()
+    
+    def connectionByHostname(self, host=None):
+        if host in self.factory.clients:
+            return self.factory.clients[host] 
+    
+    @staticmethod 
+    def sendToConnection(connection, line):
+        connection.write(line + "\n")   
+  

dist/tools/pyterm/testbeds/__init__.py

+__all__ = ["testbeds"]
+

dist/tools/pyterm/testbeds/testbeds.py

+#!/usr/bin/python2
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2014  Philipp Rosenkranz <philipp.rosenkranz@fu-berlin.de>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301 USA
+
+
+import os, re, datetime
+from subprocess import call, Popen, PIPE
+
+
+class Testbed():
+    log_dir_name = 'log'
+    
+    def __init__(self):
+        pass
+    
+    def initCleanWithFlash(self):
+        self.stop()
+        self.cleanLogs()
+        self.flashNodes()
+        self.start()
+        
+    def initClean(self):
+        self.cleanLogs()
+        self.start()
+    
+    def flashNodes(self):
+        raise NotImplementedError("Inherit from Testbed and implement flashNodes") 
+    
+    def cleanLogs(self):
+        raise NotImplementedError("Inherit from Testbed and implement flashNodes") 
+    
+    def archiveLogs(self, experiment = None):
+        raise NotImplementedError("Inherit from Testbed and implement flashNodes") 
+    
+    def start(self):
+        raise NotImplementedError("Inherit from Testbed and implement flashNodes")  
+    
+    def stop(self):
+        raise NotImplementedError("Inherit from Testbed and implement flashNodes") 
+    
+    def defaultArchivePostfix(self, experimentName = None):
+        if not experimentName:
+            experimentName = "unknown"
+        time = datetime.datetime.now().strftime("%Y-%m-%d_%H_%M_%S")
+        postfix = "-" + experimentName +"_" + time  
+        return postfix     
+    
+    def printAndCall(self, cmdString):
+        print(cmdString) 
+        call(cmdString, shell=True)         
+    
+    
+class DESTestbed(Testbed):    
+    def __init__(self, serverHost = None, serverPort=None, userName = None, flasher = None, 
+                 hexfilePath = None, pyterm = None, logFilePath = None, hostFile = None):
+        self.serverHost = serverHost
+        self.serverPort = str(serverPort)
+        self.userName = userName
+        self.flasher = flasher
+        self.hexFilePath = hexfilePath
+        self.pyterm = pyterm
+        self.logFilePath = logFilePath
+        self.hostFile = hostFile
+        
+    def flashNodes(self):       
+        self.printAndCall("parallel-ssh -h %s -l %s 'python %s'" % (self.hostFile, self.userName, self.flasher))
+        
+    def cleanLogs(self):        
+        self.printAndCall("rm -rf %s/*.log" % (self.logFilePath))
+        
+    def archiveLogs(self, postfix = None): 
+        postfix = self.defaultArchivePostfix(postfix)
+        logDir = self.logFilePath.split("/")[-1]
+        self.printAndCall("cd %s/..; tar -cjf archived_logs%s.tar.bz2 %s/*.log" % (self.logFilePath, postfix, logDir))
+        
+    def start(self):        
+        self.printAndCall("parallel-ssh -h %s -l %s 'screen -S pyterm -d -m python %s -ln %s'" % (self.hostFile, self.userName, self.pyterm, self.log_dir_name))
+        
+    def stop(self):        
+        self.printAndCall("parallel-ssh -h %s -l %s 'screen -X -S pyterm quit'" % (self.hostFile, self.userName))
+        
+class LocalTestbed(Testbed):    
+    def __init__(self, serverHost = None, serverPort=None, flasher = None, hexfilePath = None, pyterm = None, logFilePath = None):
+        self.serverHost = serverHost
+        self.serverPort = str(serverPort)
+        self.flasher = flasher
+        self.hexFilePath = hexfilePath
+        self.pyterm = pyterm
+        self.logFilePath = logFilePath
+        
+    def findPorts(self):
+        devlist = os.listdir("/dev/")        
+        regex = re.compile('^ttyUSB')        
+        return sorted([port for port in devlist if regex.match(port)])
+
+    def flashNodes(self):       
+        self.printAndCall("python %s %s" % (self.flasher, self.hexFilePath))
+        
+    def cleanLogs(self):      
+        self.printAndCall("rm -rf %s/*.log" % (self.logFilePath))
+        
+    def archiveLogs(self, postfix = None): 
+        postfix = self.defaultArchivePostfix(postfix)
+        logDir = self.logFilePath.split("/")[-1]
+        self.printAndCall("cd %s/..; tar -cjf archived_logs%s.tar.bz2 %s/*.log" % (self.logFilePath, postfix, logDir))
+        
+    def start(self):
+        portList = self.findPorts()
+        for port in portList:           
+            self.printAndCall("screen -S pyterm-%s -d -m python %s -H %s -rn %s -p /dev/%s -ln %s" % (port, self.pyterm, port, port, port, self.log_dir_name))
+            
+    def stop(self):
+        portList = self.findPorts()
+        for port in portList:         
+            self.printAndCall("screen -X -S pyterm-%s quit" % (port))
+            
+class DesVirtTestbed(Testbed):       
+    def __init__(self, serverHost = None, serverPort=None, desvirtPath = None, topologyName = None, pyterm = None, logFilePath = None):
+        self.serverHost = serverHost
+        self.serverPort = str(serverPort)
+        self.desvirtPath = desvirtPath
+        self.topologyName = topologyName
+        self.pyterm = pyterm
+        self.logFilePath = logFilePath
+        self.namePortList = []
+        
+    def findPorts(self):        
+        return self.namePortList
+    
+    def startDesVirtNetwork(self):
+        print "executing: " + "./vnet --start --name " + self.topologyName + " in: " + self.desvirtPath
+        call("sh -c \"./vnet --define --name " + self.topologyName + "\"", cwd=self.desvirtPath, shell=True)
+        stream = Popen("sh -c \"./vnet --start --name " + self.topologyName + "\"", cwd=self.desvirtPath, shell=True, stderr=PIPE).stderr
+        pats = r'.*riotnative.*\.elf (\S+) -t (\S+)'
+        pattern = re.compile(pats)
+        for line in stream:
+            match = pattern.match(line)
+            if(match):
+                tuple = match.groups()
+                self.namePortList.append((tuple[0], int(tuple[1])))
+        self.namePortList = sorted(self.namePortList)
+        for tuple in self.namePortList:
+            print "name: " + tuple[0] + " port: " + str(tuple[1])
+            
+    def stopDesVirtNetwork(self):
+        call("sh -c \"./vnet --stop --name " + self.topologyName + "\"", cwd=self.desvirtPath, shell=True)                
+
+    def flashNodes(self):       
+        pass
+        
+    def cleanLogs(self): 
+        self.printAndCall("rm -rf %s/*.log" % (self.logFilePath))
+        
+    def archiveLogs(self, postfix = None): 
+        postfix = self.defaultArchivePostfix(postfix)
+        logDir = self.logFilePath.split("/")[-1]
+        self.printAndCall("cd %s/..; tar -cjf archived_logs%s.tar.bz2 %s/*.log" % (self.logFilePath, postfix, logDir))
+        
+    def start(self):
+        for node in self.namePortList:           
+            self.printAndCall("screen -S pyterm-%s -d -m python %s -H %s -rn %s -ts %s -ln %s" % (node[0], self.pyterm, node[0], node[0], node[1], self.log_dir_name))
+            
+    def stop(self):
+        print "stop called"
+        for node in self.namePortList:         
+            self.printAndCall("screen -X -S pyterm-%s quit" % (node[0]))
+        self.stopDesVirtNetwork()
+
+
+