brukeropus.control.dde

  1# Python DDE interface by David Naylor, http://code.activestate.com/recipes/577654-dde-client/
  2# Slightly modified to work with python 3 and streamline the code a bit
  3# I do not understand how this works, just use it for sending/recieving DDE communication through OPUS
  4
  5from ctypes import POINTER, WINFUNCTYPE, c_char_p, c_void_p, c_int, c_ulong, c_char_p, byref, windll, create_string_buffer
  6from ctypes.wintypes import BOOL, DWORD, BYTE, INT, LPCWSTR, UINT, ULONG, HWND, MSG
  7
  8# DECLARE_HANDLE(name) typedef void *name;
  9HCONV     = c_void_p  # = DECLARE_HANDLE(HCONV)
 10HDDEDATA  = c_void_p  # = DECLARE_HANDLE(HDDEDATA)
 11HSZ       = c_void_p  # = DECLARE_HANDLE(HSZ)
 12LPBYTE    = c_char_p  # POINTER(BYTE)
 13LPDWORD   = POINTER(DWORD)
 14LPSTR    = c_char_p
 15ULONG_PTR = c_ulong
 16
 17# See windows/ddeml.h for declaration of struct CONVCONTEXT
 18PCONVCONTEXT = c_void_p
 19
 20DMLERR_NO_ERROR = 0
 21
 22# Predefined Clipboard Formats
 23CF_TEXT         =  1
 24CF_BITMAP       =  2
 25CF_METAFILEPICT =  3
 26CF_SYLK         =  4
 27CF_DIF          =  5
 28CF_TIFF         =  6
 29CF_OEMTEXT      =  7
 30CF_DIB          =  8
 31CF_PALETTE      =  9
 32CF_PENDATA      = 10
 33CF_RIFF         = 11
 34CF_WAVE         = 12
 35CF_UNICODETEXT  = 13
 36CF_ENHMETAFILE  = 14
 37CF_HDROP        = 15
 38CF_LOCALE       = 16
 39CF_DIBV5        = 17
 40CF_MAX          = 18
 41
 42DDE_FACK          = 0x8000
 43DDE_FBUSY         = 0x4000
 44DDE_FDEFERUPD     = 0x4000
 45DDE_FACKREQ       = 0x8000
 46DDE_FRELEASE      = 0x2000
 47DDE_FREQUESTED    = 0x1000
 48DDE_FAPPSTATUS    = 0x00FF
 49DDE_FNOTPROCESSED = 0x0000
 50
 51DDE_FACKRESERVED  = (~(DDE_FACK | DDE_FBUSY | DDE_FAPPSTATUS))
 52DDE_FADVRESERVED  = (~(DDE_FACKREQ | DDE_FDEFERUPD))
 53DDE_FDATRESERVED  = (~(DDE_FACKREQ | DDE_FRELEASE | DDE_FREQUESTED))
 54DDE_FPOKRESERVED  = (~(DDE_FRELEASE))
 55
 56XTYPF_NOBLOCK        = 0x0002
 57XTYPF_NODATA         = 0x0004
 58XTYPF_ACKREQ         = 0x0008
 59
 60XCLASS_MASK          = 0xFC00
 61XCLASS_BOOL          = 0x1000
 62XCLASS_DATA          = 0x2000
 63XCLASS_FLAGS         = 0x4000
 64XCLASS_NOTIFICATION  = 0x8000
 65
 66XTYP_ERROR           = (0x0000 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK)
 67XTYP_ADVDATA         = (0x0010 | XCLASS_FLAGS)
 68XTYP_ADVREQ          = (0x0020 | XCLASS_DATA | XTYPF_NOBLOCK)
 69XTYP_ADVSTART        = (0x0030 | XCLASS_BOOL)
 70XTYP_ADVSTOP         = (0x0040 | XCLASS_NOTIFICATION)
 71XTYP_EXECUTE         = (0x0050 | XCLASS_FLAGS)
 72XTYP_CONNECT         = (0x0060 | XCLASS_BOOL | XTYPF_NOBLOCK)
 73XTYP_CONNECT_CONFIRM = (0x0070 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK)
 74XTYP_XACT_COMPLETE   = (0x0080 | XCLASS_NOTIFICATION )
 75XTYP_POKE            = (0x0090 | XCLASS_FLAGS)
 76XTYP_REGISTER        = (0x00A0 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK )
 77XTYP_REQUEST         = (0x00B0 | XCLASS_DATA )
 78XTYP_DISCONNECT      = (0x00C0 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK )
 79XTYP_UNREGISTER      = (0x00D0 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK )
 80XTYP_WILDCONNECT     = (0x00E0 | XCLASS_DATA | XTYPF_NOBLOCK)
 81XTYP_MONITOR         = (0x00F0 | XCLASS_NOTIFICATION | XTYPF_NOBLOCK)
 82
 83XTYP_MASK            = 0x00F0
 84XTYP_SHIFT           = 4
 85
 86TIMEOUT_ASYNC        = 0xFFFFFFFF
 87
 88def get_winfunc(libname, funcname, restype=None, argtypes=(), _libcache={}):
 89    """Retrieve a function from a library, and set the data types."""
 90    if libname not in _libcache:
 91        _libcache[libname] = windll.LoadLibrary(libname)
 92    func = getattr(_libcache[libname], funcname)
 93    func.argtypes = argtypes
 94    func.restype = restype
 95
 96    return func
 97
 98
 99DDECALLBACK = WINFUNCTYPE(HDDEDATA, UINT, UINT, HCONV, HSZ, HSZ, HDDEDATA, 
100                          ULONG_PTR, ULONG_PTR)
101
102class DDE(object):
103    """Object containing all the DDE functions"""
104    AccessData         = get_winfunc("user32", "DdeAccessData",          LPBYTE,   (HDDEDATA, LPDWORD))
105    ClientTransaction  = get_winfunc("user32", "DdeClientTransaction",   HDDEDATA, (LPBYTE, DWORD, HCONV, HSZ, UINT, UINT, DWORD, LPDWORD))
106    Connect            = get_winfunc("user32", "DdeConnect",             HCONV,    (DWORD, HSZ, HSZ, PCONVCONTEXT))
107    CreateStringHandle = get_winfunc("user32", "DdeCreateStringHandleW", HSZ,      (DWORD, LPCWSTR, UINT))
108    Disconnect         = get_winfunc("user32", "DdeDisconnect",          BOOL,     (HCONV,))
109    GetLastError       = get_winfunc("user32", "DdeGetLastError",        UINT,     (DWORD,))
110    Initialize         = get_winfunc("user32", "DdeInitializeW",         UINT,     (LPDWORD, DDECALLBACK, DWORD, DWORD))
111    FreeDataHandle     = get_winfunc("user32", "DdeFreeDataHandle",      BOOL,     (HDDEDATA,))
112    FreeStringHandle   = get_winfunc("user32", "DdeFreeStringHandle",    BOOL,     (DWORD, HSZ))
113    QueryString        = get_winfunc("user32", "DdeQueryStringA",        DWORD,    (DWORD, HSZ, LPSTR, DWORD, c_int))
114    UnaccessData       = get_winfunc("user32", "DdeUnaccessData",        BOOL,     (HDDEDATA,))
115    Uninitialize       = get_winfunc("user32", "DdeUninitialize",        BOOL,     (DWORD,))
116
117class DDEError(RuntimeError):
118    """Exception raise when a DDE errpr occures."""
119    def __init__(self, msg, idInst=None):
120        if idInst is None:
121            RuntimeError.__init__(self, msg)
122        else:
123            RuntimeError.__init__(self, "%s (err=%s)" % (msg, hex(DDE.GetLastError(idInst))))
124
125class DDEClient(object):
126    """The DDEClient class.
127
128    Use this class to create and manage a connection to a service/topic.  To get
129    classbacks subclass DDEClient and overwrite callback."""
130
131    def __init__(self, service, topic):
132        """Create a connection to a service/topic."""
133        self._idInst = DWORD(0)
134        self._hConv = HCONV()
135
136        self._callback = DDECALLBACK(self._callback)
137        res = DDE.Initialize(byref(self._idInst), self._callback, 0x00000010, 0)
138        if res != DMLERR_NO_ERROR:
139            raise DDEError("Unable to register with DDEML (err=%s)" % hex(res))
140
141        hszService = DDE.CreateStringHandle(self._idInst, service, 1200)
142        hszTopic = DDE.CreateStringHandle(self._idInst, topic, 1200)
143        self._hConv = DDE.Connect(self._idInst, hszService, hszTopic, PCONVCONTEXT())
144        DDE.FreeStringHandle(self._idInst, hszTopic)
145        DDE.FreeStringHandle(self._idInst, hszService)
146        if not self._hConv:
147            raise DDEError("Unable to establish a conversation with server", self._idInst)
148
149    def __del__(self):
150        """Cleanup any active connections."""
151        if self._hConv:
152            DDE.Disconnect(self._hConv)
153        if self._idInst:
154            DDE.Uninitialize(self._idInst)
155
156    def advise(self, item, stop=False):
157        """Request updates when DDE data changes."""
158        hszItem = DDE.CreateStringHandle(self._idInst, item, 1200)
159        hDdeData = DDE.ClientTransaction(LPBYTE(), 0, self._hConv, hszItem, CF_TEXT, XTYP_ADVSTOP if stop else XTYP_ADVSTART, TIMEOUT_ASYNC, LPDWORD())
160        DDE.FreeStringHandle(self._idInst, hszItem)
161        if not hDdeData:
162            raise DDEError("Unable to %s advise" % ("stop" if stop else "start"), self._idInst)
163        DDE.FreeDataHandle(hDdeData)
164
165    def execute(self, command, timeout=5000):
166        """Execute a DDE command."""
167        pData = c_char_p(command)
168        cbData = DWORD(len(command) + 1)
169        hDdeData = DDE.ClientTransaction(pData, cbData, self._hConv, HSZ(), CF_TEXT, XTYP_EXECUTE, timeout, LPDWORD())
170        if not hDdeData:
171            raise DDEError("Unable to send command", self._idInst)
172        DDE.FreeDataHandle(hDdeData)
173
174    def request(self, item, timeout=5000):
175        """Request data from DDE service."""
176        hszItem = DDE.CreateStringHandle(self._idInst, item, 1200)
177        hDdeData = DDE.ClientTransaction(LPBYTE(), 0, self._hConv, hszItem, CF_TEXT, XTYP_REQUEST, timeout, LPDWORD())
178        DDE.FreeStringHandle(self._idInst, hszItem)
179        if not hDdeData:
180            raise DDEError("Unable to request item", self._idInst)
181
182        if timeout != TIMEOUT_ASYNC:
183            pdwSize = DWORD(0)
184        pData = DDE.AccessData(hDdeData, byref(pdwSize))
185        if not pData:
186            DDE.FreeDataHandle(hDdeData)
187            raise DDEError("Unable to access data", self._idInst)
188            # TODO: use pdwSize
189            DDE.UnaccessData(hDdeData)
190        DDE.FreeDataHandle(hDdeData)
191        return pData
192
193    def callback(self, value, item=None):
194        """Calback function for advice."""
195        print("%s: %s" % (item, value))
196
197    def _callback(self, wType, uFmt, hConv, hsz1, hsz2, hDdeData, dwData1, dwData2):
198        dwSize = DWORD(0)
199        pData = DDE.AccessData(hDdeData, byref(dwSize))
200        if pData:
201            item = create_string_buffer('\000' * 128)
202            DDE.QueryString(self._idInst, hsz2, item, 128, 1004)
203            self.callback(pData, item.value)
204            DDE.UnaccessData(hDdeData)
205        return DDE_FACK
206        return 0
207
208def WinMSGLoop():
209    """Run the main windows message loop."""
210    LPMSG = POINTER(MSG)
211    LRESULT = c_ulong
212    GetMessage = get_winfunc("user32", "GetMessageW", BOOL, (LPMSG, HWND, UINT, UINT))
213    TranslateMessage = get_winfunc("user32", "TranslateMessage", BOOL, (LPMSG,))
214    # restype = LRESULT
215    DispatchMessage = get_winfunc("user32", "DispatchMessageW", LRESULT, (LPMSG,))
216
217    msg = MSG()
218    lpmsg = byref(msg)
219    while GetMessage(lpmsg, HWND(), 0, 0) > 0:
220        TranslateMessage(lpmsg)
221        DispatchMessage(lpmsg)
222
223if __name__ == "__main__":
224    # ------------------------------------------
225    # Test with Excel
226    # open workbook and enter something in Sheet 1, R1C1
227    dde = DDEClient("Excel", "Sheet1")
228    print(dde.request("R1C1").decode('ASCII'))
229    
230    # ------------------------------------------
231    # Test with OPUS
232    # make sure OPUS software is open
233    dde = DDEClient("Opus", "System")
234    print(dde.request("GET_VERSION_EXTENDED"))