Welcome to DRAMA Python Interface’s documentation!¶

This is the documentation DRAMA interface to Python, via the DRAMA2 API. It allows you to create python scripts and GUIs in Python which will communicate with DRAMA Tasks.

I’m afraid the layout of this documentation is rather poor - it is automatically generated from the source but there are some restrictions on what I can do - I hope to work out how to improve it over time.

The interface is generated using the SWIG tool. The table belows shows the main entry point classes and procedures, but there are various other support classes listed in the Code Documentation and Index.

Having started DRAMA in the normal way, the DPYTHON_DIR is put in PYTHONPATH and you should be able to import dpython to access these classes.

Class Name	Description
dpython.Task	Implements a DRAMA task.
dpython.Path	Implements a Path object used to talk to other tasks.
dpython.AId	Provides access to DRAMA’s SDS objects (A for argument).
dpython.pmonitor	Provides the ability to monitor parameters in other tasks.
dpython.SdsToDict	Convert an SDS structure to a dictionary.
dpython.MonitorToDict	A subclass of pmonitor which puts parameter values in a dictionary.

The source code library includes the file pythontest.py, which is a general test program. The examples below are based on that file.

Here is how you might define a class to run the DRAMA message loop in a thread:

class dramaTask (threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
    def run(self):
        self.task = dpython.Task("AAA")

        self.task.AddExitAction()
        self.task.RunDrama()
    def GetTask(self):
        return dpython.TaskWeakPtr(self.task)
    def SignalExit(self):
        self.task.SignalDramaToExit(0)

And then to actually start the task, use:

dramaThread = dramaTask()
dramaThread.start()
time.sleep(1)

Note that after executing the start() method, you typically wait a second for the DRAMA task to actually start (really, to force a thread context switch). Otherwise the DRAMA task may not be running when you try to do something.

The following show how you might load and run the task TICKER, from the file ticker in the directory $DITS_DEV (presuming the DRAMA networking is running):

myPath = dpython.Path(dramaThread.GetTask(), "TICKER", "", "DITS_DEV:ticker")
myPath.SetBuffers(1600,1600);
myPath.GetPath()

To send an obey message to TICKER using this path, you would do:

myPath.Obey("TICK")

And this shows how to create a command structure (might use for an Obey as well) and then to use it to set a parameter value:

a = dpython.AId.CreateArgStruct()
a.Puti("Argument1", 10)
myPath.SetParam("PARAM1", a)

Here a get followed by various AId (SDS) calls to examine the result:

b = myPath.GetParam("PARAM1")
b.List()

name = b.GetName()
code = b.Code()
if code == "SDS_STRUCT":
    item = b.Find("PARAM1")
    name = item.GetName()
    code = item.Code()
    dims = item.GetNumArrayDims()
    value = item.toString()

Here we create an SDS structure, list it and then convert it to a dictionary to demostrate the required calls:

a = dpython.AId.CreateArgStruct()
a.Putc("CharItem", 3);
a.Puts("ShortItem", 4);
a.Putus("UShortItem", 5);
a.Puti("INT32Item", 6);
a.Putui("UINT32Item", 7);
a.Puti64("INT64Item", 8);
a.Putui64("UINT64Item", 9);
a.Putf("FloatItem", 10.1);
a.Putd("DoubleItem", 11.2);
a.PutStr("StringItem1", "12");

# Have a look at what we have created (to stdout).
a.List()

#
# We can convert things like this to python dictionaries.
#
print (dpython.SdsToDict(a))

Or reading an SDS structure from a file and converting it to a dictionary:

b = dpython.AId.Read("tict_pars.sds")
print (dpython.SdsToDict(b))

Note that to cause the DRAMA task thread to exit, you need to execute:

dramaThread.SignalExit()

DRAMA Parameter monitoring allows your task to be notified automatically if the value of a parameter in another task changes. It requires that, having created your DRAMA message reading thread, that you create and run another thread to do the monitoring. In the example below, I first create a class which will be used to implement thread. This class uses an object of type dpython.MonitorToDict, itself a sub-class of dpython.pmonitor, to do the monitoring. The MonitorToDict class is used to cause the parameter values to appear in a dictionary. You could create your own sub-class of pmonitor to have other things happen on the changes.:

class dramaMonitor (threading.Thread):
    # Constructor.  Save details to be used when thread starts.
    def __init__(self, task, path, parameters):
        threading.Thread.__init__(self)
        self.task = task.GetTask()
        self.param = parameters
        self.path  = path
        self.monitor = None

    # After the thread is started, this is run within the thread
    #  to do the real work of thread.  We set up and run our monitor,
    #  using a variable of type myMonitor (a sub-class of
    #   dpython.pmonitor) to do the monitoring.
    def run(self):
        self.monitor = dpython.MonitorToDict(self.task, self.param)
        self.monitor.RunMonitor(self.path)
        print ("Dictionary contains:")
        print (self.monitor.dict)
    # Method used to cancel the monitor.
    def cancel(self):
        if self.monitor:
            self.monitor.Cancel()

Once I have this class definitoin, I can ran the monitor like so:

# Create the monitor thread object.  Specifying parameter to monitor
#   Presumes the dramaThread and myPath objects from previous examples.
monitorThread = dramaMonitor(dramaThread, myPath,
                             dpython.StringVector(["PARAM1",
                                                   "PARAM2",
                                                   "PARAM3",
                                                   "PARAM4",
                                                   "STRUCT_PARAM"]))
# And now start the monitor thread.
monitorThread.start()

Parameter monitoring is particuarly usefull to allow a GUI to automatically update displayed items.

Contents:

DRAMA 2 Python Code Documentation

Indices and tables¶

Example code¶

The full dpython test program is below. This can be found in the dpython ACMM sub-system.

# "@(#) $Id: ACMM:dpython/dpythontest.py,v 1.13 12-Jun-2020 14:26:46+10 aidanfarrell $"
#
#  Basic dpython test script.
#
# This script is meant to exercise most of the features of dpython.
#
# It creates a DRAMA task named AAA and runs it in a thread. It creates
#  two other threads which work whilst the DRAMA task is running (but
#  don't do anything real - they just demostrate when you can do)
#
# It presumes the TICKER program (DRAMA Task) is running (at the
# moment, a future version may load it)
#
# Various messages are sent to the TICKER task before it is told to exit, and
# then the DRAMA thread is exited. We then wait for the other threads to exit.
#


# Script to test dpython

# Load the DRAMA code.
import dpython
# And the threading interfaces
import threading

# We need to access time of day.
import time

 
exitFlag = 0

#
# This class is used to implement the two worker threads.
#
#   These are just used to demostrate that you can have
#   other things runnings whilst the DRAMA task is working.
#
class myThread (threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):
        print ("Starting " + self.name)
        print_time(self.name, 1, self.counter)
        print ("Exiting " + self.name)

#
# This class is used to implement the DRAMA message processing
# thread.
#
class dramaTask (threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
    def run(self):
        """ Make ourselves a DRAMA task named AAA and add an EXIT action.
        Run the DRAMA message processing loop
        """
        self.task = dpython.Task("AAA")
        self.task.AddExitAction()
        print ("Starting DRAMA task")
        self.task.RunDrama()
        print ("DRAMA task shutdown")
    def GetTask(self):
        """ References to the underlying DRAMA task are needed at times, return it"""
        return dpython.TaskWeakPtr(self.task)
    def SignalExit(self):
        """ Tell the DRAMA messaging thread to exit """
        self.task.SignalDramaToExit(0)


#
#
# This class is used to implement a thread which
#  sends a DRAMA Parameter monitor message and
#  processes the replies.
#
# The results of the monitor are put in a dictionary,
#  which can be accessed from the monitor.dict item.
#
class dramaMonitor (threading.Thread):
    def __init__(self, task, path, parameters):
        threading.Thread.__init__(self)
        self.task = task.GetTask()
        self.param = parameters
        self.path  = path
        self.monitor = None
        
    # After the thread is started, this is run within the thread
    #  to do the real work of thread.  We set up and run our monitor,
    #  using a variable of type myMonitor (a sub-class of
    #   dpython.pmonitor) to do the monitoring.
    def run(self):
        print ("Starting dramaMonitor thread")
        #self.monitor = dpython.pmonitor(self.task, self.param)
        self.monitor = dpython.MonitorToDict(self.task, self.param)
        self.monitor.RunMonitor(self.path)
        print ("Ended dramaMonitor thread, dictionary contains:")
        print (self.monitor.dict)
    # Method used to cancel the monitor.
    def cancel(self):
        if self.monitor:
            print("### Will cancel monitor")
            self.monitor.Cancel()
        else:
            print("Monitor not running to can't cancel it")
        
        

def print_time(threadName, delay, counter):
    print ("Running thread:" + threadName + ", delay =" + str(delay) + ", count=" + str(counter))
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print ("" +threadName + " " + str(counter) + ", " + time.ctime(time.time()))
        counter -= 1

# Create new thread objects.
thread1 = myThread(1, "Thread-1", 5)
thread2 = myThread(2, "Thread-2", 7)
dramaThread = dramaTask()

# Start new Threads
thread1.start()
thread2.start()
dramaThread.start()

# So at this point, 4 threads are running - the thread running this
# code, thread1 and thread2, which are worker threads, and the dramaThread
# which is running the drama message processing loop.


# This wait is needed to allow the DRAMA thread to start, otherwise
#   things won't work, in particular, the DRAMA thread won't be properly
#   started by the time we start using DRAMA below.
time.sleep(1)




# Create a path object to task to the TICKER task, and get the path.
print("Creating DRAMA path object")
myPath = dpython.Path(dramaThread.GetTask(), "TICKER", "", "DITS_DEV:ticker")
print("Have new DRAMA path object");
# DRAMA requires you to set buffer sizes before you set up the path.
#   See the DRAMA documentation for details.
myPath.SetBuffers(1600,1600);
#
# Now actually get the path - this is where we confirm the task TICKER
#  is running and set up for commuicating with it.
myPath.GetPath()


#
# Set up a parameter monitor.
#
monitorThread = dramaMonitor(dramaThread, myPath,
                             dpython.StringVector(["PARAM1",
                                                   "PARAM2",
                                                   "PARAM3",
                                                   "PARAM4",
                                                   "STRUCT_PARAM"]))
monitorThread.start()



#
#  The TICKER task has an action (command) named "TICK".  The basic
#   DRAMA test is to send a set of these.  So loop whilst sending
#   the commands.
#
#   Note - myPath.Obey() will block this thread until the TICK action
#    is complete.  If you want this thread to do other things, you need
#     to offload the "Obey" to another thread.
#
print("Sending 100 tick messages to the TICKER task")
ticks = 100
while (ticks):
    myPath.Obey("TICK")
    ticks -= 1

#
# Show how we can set a task parameter.  TICKER has a parameter named
# PARAM1 which we set the value of.
#
# Also demostrates how to create an argument for a message (Via the dpython.Aid
#  class).  All DRAMA command arguments (and reply arguments) are DRAMA
#  SDS structures.  So here we are creating an SDS structure, a very simple
#  one.
#
print("Sending SET message for PARAM1 to TICKER")
a = dpython.AId.CreateArgStruct()
a.Puti("Argument1", 10)
myPath.SetParam("PARAM1", a)

# Allow monitor update to occur and output - so that output
#  occurs in a well defined order.
time.sleep(1)
#
# Similarly for getting a parameter value.  We get the value back and
#  in this case, just List the value to stdout.  You can of course get
#  the value into your code.  The value you have gotten back is again
#  an SDS structure, so you could get something of considerable complexity
#  and size back (including large Image structures).
#
print("Sending GET message for PARAM1 to TICKER")
b = myPath.GetParam("PARAM1")
name = b.GetName()
code = b.Code()
print("Get Parameter result TopLevel:Name = ", name, ", Code = ", code)
if code == "SDS_STRUCT":
    item = b.Find("PARAM1")
    name = item.GetName()
    code = item.Code()
    dims = item.GetNumArrayDims()
    print("Get Parameter result Item    :Name = ", name, ", Code = ", code, ", NumArrayDims=", dims)
    value = item.toString()
    print("Get Parameter result Value:", value)


# Get with exception handling to see how this work.
try:
    b = myPath.GetParam("PARAM1Y")
except RuntimeError as e:
    print("Runtime Error received getting PARAM1Y - expected, exception handling test")
    print(e.args)



# Cancel the monitor.
monitorThread.cancel()
# temp wait
time.sleep(2)

#
# Shutdown the TICKER task by sending it an EXIT command.
#
print("Will shutdown TICKER tick")
myPath.Obey("EXIT")

#
# Wait for it to go.  (It is not really clear we need to do this)
#
time.sleep(1)


#
# SignalExit() causes the task.RunDrama() method to return, and as
#  a result this task stops processing DRAMA messages and the drama
#  thread will exit.
#
dramaThread.SignalExit()

#
# Extra examples,
#
# Here, creating a more complex SDS structure.
#
print("Creating and listing an argument structure")
a = dpython.AId.CreateArgStruct()
a.Putc("CharItem", 3);
a.Puts("ShortItem", 4);
a.Putus("UShortItem", 5);
a.Puti("INT32Item", 6);
a.Putui("UINT32Item", 7);
a.Puti64("INT64Item", 8);
a.Putui64("UINT64Item", 9);
a.Putf("FloatItem", 10.1);
a.Putd("DoubleItem", 11.2);
a.PutStr("StringItem1", "12");

# Have a look at what we have created (to stdout).
a.List()

#
# We can convert things like this to python dictionaries.
#
print (dpython.SdsToDict(a))

#
# We can also read SDS structures from files.
#
b = dpython.AId.Read("tict_pars.sds")

print (dpython.SdsToDict(b))


print ("Exiting Main Thread")

Welcome to DRAMA Python Interface’s documentation!¶

Indices and tables¶

Example code¶

DRAMA Python Interface

Navigation

Related Topics