Windows 10 + python 3.6.3 64 位(也尝试过 32 位)。我是一名 Python 开发人员,(几乎)第一次尝试使用 COM,并遇到了这个巨大的障碍。
当我尝试通过
win32com
或 comtypes
使用在 dll 中实现的 IRTDServer(不是我编写的)时,遇到了各种错误。使用 win32com
变得更加困难。我为下面的两个库提供了一个示例单元测试。
从 Excel 2016 访问服务器按预期工作;这将返回预期值:
=RTD("foo.bar", , "STAT1", "METRIC1")
这是一个简单的测试用例,应该连接到服务器,但没有。 (这只是一个版本,因为我已经多次更改它以尝试调试问题。)
from unittest import TestCase
class COMtest(TestCase):
def test_win32com(self):
import win32com.client
from win32com.server.util import wrap
class RTDclient:
# are these only required when implementing the server?
_com_interfaces_ = ["IRTDUpdateEvent"]
_public_methods_ = ["Disconnect", "UpdateNotify"]
_public_attrs_ = ["HeartbeatInterval"]
def __init__(self, *args, **kwargs):
self._comObj = win32com.client.Dispatch(*args, **kwargs)
def connect(self):
self._rtd = win32com.client.CastTo(self._comObj, 'IRtdServer')
result = self._rtd.ServerStart(wrap(self))
assert result > 0
def UpdateNotify(self):
print("UpdateNotify() callback")
def Disconnect(self):
print("Disconnect() called")
HeartbeatInterval = -1
_rtd = RTDclient("foo.bar")
_rtd.connect()
结果:
Traceback (most recent call last):
File "env\lib\site-packages\win32com\client\gencache.py", line 532, in EnsureDispatch
ti = disp._oleobj_.GetTypeInfo()
pywintypes.com_error: (-2147467263, 'Not implemented', None, None)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "test\test.py", line 23, in test_win32com
_rtd.connect()
File "test\test.py", line 16, in connect
self._rtd = win32com.client.CastTo(dispatch, 'IRtdServer')
File "env\lib\site-packages\win32com\client\__init__.py", line 134, in CastTo
ob = gencache.EnsureDispatch(ob)
File "env\lib\site-packages\win32com\client\gencache.py", line 543, in EnsureDispatch
raise TypeError("This COM object can not automate the makepy process - please run makepy manually for this object")
TypeError: This COM object can not automate the makepy process - please run makepy manually for this object
按照这些指示,我成功运行了
makepy
脚本:
> env\Scripts\python.exe env\lib\site-packages\win32com\client\makepy.py "foo.bar"
Generating to C:\Users\user1\AppData\Local\Temp\gen_py\3.5\longuuid1x0x1x0.py
Building definitions from type library...
Generating...
Importing module
(出于隐私考虑,我替换了 stackoverflow 上的 UUID。此 UUID 与“foo.bar”的 typelib UUID 相同。)
生成的文件包含
IRtdServer
和IRTDUpdateEvent
的各种函数和类型定义。但在这个文件中,两个接口都是win32com.client.DispatchBaseClass
的子类,而根据OleViewDotNet,它们应该是IUnknown
的子类?
但是,当我尝试再次运行单元测试时,我收到了与之前完全相同的错误。好像查找机制没有找到生成的模块?
此外,
GetTypeInfo
返回Not implemented
让我感到震惊。根据我的理解,win32com 使用该方法(IDispatch
COM 接口的一部分)来确定其他接口中所有其他函数的参数和返回类型,包括 IRtdServer
。如果不实现,将无法正确确定类型。然而,生成的文件似乎包含这些信息,这也令人困惑。
from unittest import TestCase
class COMtest(TestCase):
def test_comtypes(self):
import comtypes.client
class RTDclient:
# are these for win32com only?
_com_interfaces_ = ["IRTDUpdateEvent"]
_public_methods_ = ["Disconnect", "UpdateNotify"]
_public_attrs_ = ["HeartbeatInterval"]
def __init__(self, clsid):
self._comObj = comtypes.client.CreateObject(clsid)
def connect(self):
self._rtd = self._comObj.IRtdServer()
result = self._rtd.ServerStart(self)
assert result > 0
def UpdateNotify(self):
print("UpdateNotify() callback")
def Disconnect(self):
print("Disconnect() called")
HeartbeatInterval = -1
_rtd = RTDclient("foo.bar")
_rtd.connect()
结果:
File "test\test.py", line 27, in test_comtypes
_rtd.connect()
File "test\test.py", line 16, in connect
self._rtd = self._comObj.IRTDServer()
File "env\lib\site-packages\comtypes\client\dynamic.py", line 110, in __getattr__
dispid = self._comobj.GetIDsOfNames(name)[0]
File "env\lib\site-packages\comtypes\automation.py", line 708, in GetIDsOfNames
self.__com_GetIDsOfNames(riid_null, arr, len(names), lcid, ids)
_ctypes.COMError: (-2147352570, 'Unknown name.', (None, None, None, 0, None))
(基于谷歌搜索和下面评论中的答案)
python.exe
的兼容模式设置为Windows XP SP3尝试不实例化IRtdServer,即替换这两行:
self._rtd = self._comObj.IRtdServer()
result = self._rtd.ServerStart(self)
与:
result = self._comObj.ServerStart(self)
这次的错误是:
TypeError: 'NoneType' object is not callable
这似乎表明
ServerStart
函数存在,但未定义? (看起来真的很奇怪。这个谜一定还有更多。)尝试将
interface="IRtdServer"
参数传递给 CreateObject
:
def __init__(self, clsid):
self._comObj = comtypes.client.CreateObject(clsid, interface="IRtdServer")
def connect(self):
result = self._comObj.ServerStart(self)
...
收到的错误是:
File "test\test.py", line 13, in __init__
self._comObj = comtypes.client.CreateObject(clsid, interface="IRtdServer")
File "env\lib\site-packages\comtypes\client\__init__.py", line 238, in CreateObject
obj = comtypes.CoCreateInstance(clsid, clsctx=clsctx, interface=interface)
File "env\lib\site-packages\comtypes\__init__.py", line 1223, in CoCreateInstance
p = POINTER(interface)()
TypeError: Cannot create instance: has no _type_
跟踪
comtypes
库中的代码,这似乎表明接口参数需要一个接口类,而不是一个字符串。我发现comtypes
库中定义了各种接口:IDispatch
、IPersist
、IServiceProvider
。都是IUnknown
的子类。根据 OleViewDotNet,IRtdServer
也是 IUnknown
的子类。这让我相信我需要在 python 中类似地编写一个 IRtdServer 类才能使用该接口,但我不知道该怎么做。我注意到了
dynamic
的CreateObject
参数。代码表明这与 interface
参数是互斥的,所以我尝试了:
def __init__(self, clsid):
self._comObj = comtypes.client.CreateObject(clsid, dynamic=True)
def connect(self):
self._rtd = self._comObj.IRtdServer()
result = self._rtd.ServerStart(self)
但是错误与我原来的错误相同:
IRtdServer
有_ctypes.COMError: (-2147352570, 'Unknown name.', (None, None, None, 0, None))
任何帮助或线索将不胜感激。预先感谢您。
(不太知道我在做什么,)我尝试使用 OleViewDotNet 查看 DLL:
我遇到了同样的问题。
我还尝试使用 win32com 让 excel 运行,这对我来说,说实话有点不稳定......我什至不能碰我的 Excel。
因此我花了一些时间研究这个。问题出在 CastTo 上。认为您(和我)加载的 COM 对象不包含足够的信息来进行转换(某些方法如 GetTypeInfo 未实现等...)
因此,我创建了一个包装器,使这些 COM 对象的方法可调用……这并不明显。这似乎对我有用。
客户端代码是从一个名为pyrtd的项目修改而来的,该项目由于各种原因而不起作用(认为是由于RTD模型的更改......RefreshData的返回现在完全不同了)。
import functools
import pythoncom
import win32com.client
from win32com import universal
from win32com.client import gencache
from win32com.server.util import wrap
EXCEL_TLB_GUID = '{00020813-0000-0000-C000-000000000046}'
EXCEL_TLB_LCID = 0
EXCEL_TLB_MAJOR = 1
EXCEL_TLB_MINOR = 4
gencache.EnsureModule(EXCEL_TLB_GUID, EXCEL_TLB_LCID, EXCEL_TLB_MAJOR, EXCEL_TLB_MINOR)
universal.RegisterInterfaces(EXCEL_TLB_GUID,
EXCEL_TLB_LCID, EXCEL_TLB_MAJOR, EXCEL_TLB_MINOR,
['IRtdServer', 'IRTDUpdateEvent'])
# noinspection PyProtectedMember
class ObjectWrapperCOM:
"""
This object can act as a wrapper for an object dispatched using win32com.client.Dispatch
Sometimes the object written by 3rd party is not well constructed that win32com will not be able to obtain
type information etc in order to cast the object to a certain interface. win32com.client.CastTo will fail.
This wrapper class will enable the object to call its methods in this case, even if we do not know what exactly
the wrapped object is.
"""
LCID = 0x0
def __init__(self, obj):
self._impl = obj # type: win32com.client.CDispatch
def __getattr__(self, item):
flags, dispid = self._impl._find_dispatch_type_(item)
if dispid is None:
raise AttributeError("{} is not a valid property or method for this object.".format(item))
return functools.partial(self._impl._oleobj_.Invoke, dispid, self.LCID, flags, True)
# noinspection PyPep8Naming
class RTDUpdateEvent:
"""
Implements interface IRTDUpdateEvent from COM imports
"""
_com_interfaces_ = ['IRTDUpdateEvent']
_public_methods_ = ['Disconnect', 'UpdateNotify']
_public_attrs_ = ['HeartbeatInterval']
# Implementation of IRTDUpdateEvent.
HeartbeatInterval = -1
def __init__(self, event_driven=True):
self.ready = False
self._event_driven = event_driven
def UpdateNotify(self):
if self._event_driven:
self.ready = True
def Disconnect(self):
pass
class RTDClient:
"""
Implements a Real-Time-Data (RTD) client for accessing COM data sources that provide an IRtdServer interface.
"""
MAX_REGISTERED_TOPICS = 1024
def __init__(self, class_id):
"""
:param classid: can either be class ID or program ID
"""
self._class_id = class_id
self._rtd = None
self._update_event = None
self._topic_to_id = {}
self._id_to_topic = {}
self._topic_values = {}
self._last_topic_id = 0
def connect(self, event_driven=True):
"""
Connects to the RTD server.
Set event_driven to false if you to disable update notifications.
In this case you'll need to call refresh_data manually.
"""
dispatch = win32com.client.Dispatch(self._class_id)
self._update_event = RTDUpdateEvent(event_driven)
try:
self._rtd = win32com.client.CastTo(dispatch, 'IRtdServer')
except TypeError:
# Automated makepy failed...no detailed construction available for the class
self._rtd = ObjectWrapperCOM(dispatch)
self._rtd.ServerStart(wrap(self._update_event))
def update(self):
"""
Check if there is data waiting and call RefreshData if necessary. Returns True if new data has been received.
Note that you should call this following a call to pythoncom.PumpWaitingMessages(). If you neglect to
pump the message loop you'll never receive UpdateNotify callbacks.
"""
# noinspection PyUnresolvedReferences
pythoncom.PumpWaitingMessages()
if self._update_event.ready:
self._update_event.ready = False
self.refresh_data()
return True
else:
return False
def refresh_data(self):
"""
Grabs new data from the RTD server.
"""
(ids, values) = self._rtd.RefreshData(self.MAX_REGISTERED_TOPICS)
for id_, value in zip(ids, values):
if id_ is None and value is None:
# This is probably the end of message
continue
assert id_ in self._id_to_topic, "Topic ID {} is not registered.".format(id_)
topic = self._id_to_topic[id_]
self._topic_values[topic] = value
def get(self, topic: tuple):
"""
Gets the value of a registered topic. Returns None if no value is available. Throws an exception if
the topic isn't registered.
"""
assert topic in self._topic_to_id, 'Topic %s not registered.' % (topic,)
return self._topic_values.get(topic)
def register_topic(self, topic: tuple):
"""
Registers a topic with the RTD server. The topic's value will be updated in subsequent data refreshes.
"""
if topic not in self._topic_to_id:
id_ = self._last_topic_id
self._last_topic_id += 1
self._topic_to_id[topic] = id_
self._id_to_topic[id_] = topic
self._rtd.ConnectData(id_, topic, True)
def unregister_topic(self, topic: tuple):
"""
Un-register topic so that it will not get updated.
:param topic:
:return:
"""
assert topic in self._topic_to_id, 'Topic %s not registered.' % (topic,)
self._rtd.DisconnectData(self._topic_to_id[topic])
def disconnect(self):
"""
Closes RTD server connection.
:return:
"""
self._rtd.ServerTerminate()
Excel 2002 似乎已经有了服务器/客户端。
pyrtd
查看该源代码,一旦创建了调度对象,它似乎就被转换为 IRtdServer。
提取相关部分,就变成下图了。
from win32com import client, universal
from win32com.server.util import wrap
def __init__(self, classid):
self._classid = classid
self._rtd = None
def connect(self, event_driven=True):
dispatch = client.Dispatch(self._classid)
self._rtd = client.CastTo(dispatch, 'IRtdServer')
if event_driven:
self._rtd.ServerStart(wrap(self))
else:
self._rtd.ServerStart(None)
请参考以下来源的client.py和examples/rtdtime.py。
pyrtd - 默认
Pyrtd/rtd/client.py
Pyrtd/示例/rtdtime.py
我已经发布了一个使用 comtypes 的版本。 @马赫
https://github.com/tifoji/pyrtdc/
win32com 也可以使用与 @Aincrad 建议的包装器类似的方法来工作