我正在尝试用python编写DDE服务器,该服务器需要向作为DDE客户端连接的程序发送连续变化的字符串。

连接到DDE服务器的程序使用以下DDE设置来连接[服务:Orbitron,主题:跟踪,项目:跟踪]。
该程序必须接收DDE服务器发送的以下字符串格式的信息:
“ UP0 DN145000001 UMusb DMfm AZ040 EL005 SNNO SATELLITE”。

此字符串的内容大约每秒钟更改一次,我希望DDE服务器将新字符串发送到所连接的DDE客户端,例如每秒一次。

我当前正在使用以下代码,该代码是原始ddeserver.py文件see here的略微修改版本。



import win32ui
from pywin.mfc import object
import dde


class MySystemTopic(object.Object):
        def __init__(self):
                object.Object.__init__(self, dde.CreateServerSystemTopic())

        def Exec(self, cmd):
                print "System Topic asked to exec", cmd

class MyOtherTopic(object.Object):
        def __init__(self, topicName):
                object.Object.__init__(self, dde.CreateTopic(topicName))

        def Exec(self, cmd):
                print "Other Topic asked to exec", cmd

class MyRequestTopic(object.Object):
        def __init__(self, topicName):
                topic = dde.CreateTopic(topicName)
                topic.AddItem(dde.CreateStringItem(""))
                object.Object.__init__(self, topic)

        def Request(self, aString):
                print "Request Topic sent: ", aString
                a="UP0 DN145800001 UMusb DMfm AZ040 EL005 SNNO SATELLITE"
                print a
                return(a)


server = dde.CreateServer()
server.AddTopic(MyRequestTopic("Tracking"))
server.Create('Orbitron')

while 1:
        win32ui.PumpWaitingMessages(0, -1)
        





运行代码时,我可以成功连接程序,并且一次收到字符串(如代码中所提供)。我尝试了几种不同的方法,但是我还想不出一种方法来更改为python代码,以使DDE服务器以循环或类似的方式连续重新发送字符串。

任何建议将非常欢迎。

附言我对python,DDE和这个论坛比较陌生,如果不清楚,我深表歉意。请让我知道。

最佳答案

# coded by JayleoPlayGround
# use Portable Python 2.7.5.1 + pywin32-214


import time
import win32ui, dde
from pywin.mfc import object


class DDETopic(object.Object):
    def __init__(self, topicName):
        self.topic = dde.CreateTopic(topicName)
        object.Object.__init__(self, self.topic)
        self.items = {}

    def setData(self, itemName, value):
        try:
            self.items[itemName].SetData( str(value) )
        except KeyError:
            if itemName not in self.items:
                self.items[itemName] = dde.CreateStringItem(itemName)
                self.topic.AddItem( self.items[itemName] )
                self.items[itemName].SetData( str(value) )


ddeServer = dde.CreateServer()
ddeServer.Create('Orbitron')
ddeTopic = DDETopic('Tracking')
ddeServer.AddTopic(ddeTopic)

while True:
    yourData = time.ctime() + ' UP0 DN145000001 UMusb DMfm AZ040 EL005 SNNO SATELLITE'
    ddeTopic.setData('Tracking', yourData)
    win32ui.PumpWaitingMessages(0, -1)
    time.sleep(0.1)

关于python - 用python创建DDE服务器并连续发送数据,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/30152460/

10-12 00:07