新手区 利用 monkey pacth 加快 poco 节点定位

奥利嗷嗷嗷叫 · April 16, 2022 · Last by 奥利嗷嗷嗷叫 replied at August 04, 2022 · 3387 hits

​ 最近在为公司的 unity 项目接入 poco 进行自动化测试,在进行某一类特定节点的坐标获取时,发现不是一般的慢,代码如下:

last_time = time.time()
nodes = poco(textMatches="Project.*")

print("poco do query time:" + str(time.time() - last_time))
last_time = time.time()
for node in nodes:
    node.get_position()
print("poco get all node position time:" + str(time.time() - last_time))

​ 执行结果如下:

​ 300 多秒。。。够去买瓶酱油了,本人对这个速度根本无法接受。记得官方的加速定位教程中有使用 freeze 函数,让我们试试用 freeze 试试:

last_time = time.time()

nodes = freeze(textMatches="Project.*")

print("poco do query time:" + str(time.time() - last_time))
last_time = time.time()
for node in nodes:
    node.get_position()
print("poco get all node position time:" + str(time.time() - last_time))

​ 结果:

​ 0.006s!?别高兴太早,这块有坑:

nodes = freeze(textMatches="Project.*")

print(nodes[0].get_position())
poco.swipe([0.5,0.5],direction=[0.5,0.7])
#官方文档中该函数用来声明重新获取信息,实测发现只对poco生成的节点有效,freeze生成的节点无效
nodes[0].invalidate()
print(nodes[0].get_position())

​ 滑动后坐标点却不变:

​ 看来 invalidate() 并未生效,先看看 freeze 干了什么:

def freeze(this):
    """
    Snapshot current **hierarchy** and cache it into a new poco instance. This new poco instance is a copy from
    current poco instance (``self``). The hierarchy of the new poco instance is fixed and immutable. It will be
    super fast when calling ``dump`` function from frozen poco. See the example below.

    Examples:
        ::

            poco = Poco(...)
            frozen_poco = poco.freeze()
            hierarchy_dict = frozen_poco.agent.hierarchy.dump()  # will return the already cached hierarchy data


    Returns:
        :py:class:`Poco <poco.pocofw.Poco>`: new poco instance copy from current poco instance (``self``)
    """

    class FrozenPoco(Poco):
        def __init__(self, **kwargs):
            hierarchy_dict = this.agent.hierarchy.dump()
            hierarchy = create_immutable_hierarchy(hierarchy_dict)
            agent_ = PocoAgent(hierarchy, this.agent.input, this.agent.screen)
            kwargs['action_interval'] = 0.01
            kwargs['pre_action_wait_for_appearance'] = 0
            super(FrozenPoco, self).__init__(agent_, **kwargs)
            self.this = this

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            pass

        def __getattr__(self, item):
            return getattr(self.this, item)

    return FrozenPoco()

​ 按官方文档,结合代码实际上是由于生成了一个静态的 hierarchy 并生产了 PocoAgent 对象,所以后续生成的 poco 对象是结构固定且不可变的,ok,找到对应问题原因了,让我们改进下:

def monkey_freeze(this):
    """
    重写freeze方法
    """

    class FrozenPoco(Poco):
        def __init__(self, **kwargs):
            hierarchy_dict = this.agent.hierarchy.dump()
            hierarchy = create_immutable_hierarchy(hierarchy_dict)
            agent_ = PocoAgent(hierarchy, this.agent.input, this.agent.screen)
            kwargs['action_interval'] = 0.01
            kwargs['pre_action_wait_for_appearance'] = 0
            super(FrozenPoco, self).__init__(agent_, **kwargs)
            self.this = this
            # 判断当前是否是冻结对象
            self.is_agent_freeze = True
            self.old_agent = this.agent

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            pass

        def __getattr__(self, item):
            return getattr(self.this, item)

        def __call__(self, name=None, **kw):
            """
            重写poco类的call方法,用于freeze()后返回结果的回调
            """

            if not name and len(kw) == 0:
                warnings.warn("Wildcard selector may cause performance trouble. Please give at least one condition to "
                              "shrink range of results")
            # 替换invalidate方法
            UIObjectProxy.invalidate = monkey_invalidate
            return UIObjectProxy(self, name, **kw)

    return FrozenPoco()


def monkey_invalidate(self):
    """
    Clear the flag to indicate to re-query or re-select the UI element(s) from hierarchy.
    """
    self._evaluated = False
    self._nodes = None
    # 判断是否是freeze的agent
    if self.poco.is_agent_freeze:
        # 重新dump
        hierarchy_dict = self.poco.old_agent.hierarchy.dump()
        hierarchy = create_immutable_hierarchy(hierarchy_dict)
        self.poco._agent = PocoAgent(hierarchy, self.poco.agent.input, self.poco.agent.screen)

​ 然后使用猴子补丁:

Poco.freeze = monkey_freeze
poco = UnityPoco(addr, device=dev)
# 或者
# poco.freeze = types.MethodType(poco,monkey_freeze())
freeze = poco.freeze()

​ 再运行刚刚的代码:

​ 完美解决。

​ 完整代码如下:

def monkey_freeze(this):
    """
    Snapshot current **hierarchy** and cache it into a new poco instance. This new poco instance is a copy from
    current poco instance (``self``). The hierarchy of the new poco instance is fixed and immutable. It will be
    super fast when calling ``dump`` function from frozen poco. See the example below.

    Examples:
        ::

            poco = Poco(...)
            frozen_poco = poco.freeze()
            hierarchy_dict = frozen_poco.agent.hierarchy.dump()  # will return the already cached hierarchy data


    Returns:
        :py:class:`Poco <poco.pocofw.Poco>`: new poco instance copy from current poco instance (``self``)
    """

    class FrozenPoco(Poco):
        def __init__(self, **kwargs):
            hierarchy_dict = this.agent.hierarchy.dump()
            hierarchy = create_immutable_hierarchy(hierarchy_dict)
            agent_ = PocoAgent(hierarchy, this.agent.input, this.agent.screen)
            kwargs['action_interval'] = 0.01
            kwargs['pre_action_wait_for_appearance'] = 0
            super(FrozenPoco, self).__init__(agent_, **kwargs)
            self.this = this
            # 判断当前是否是冻结对象
            self.is_agent_freeze = True
            self.old_agent = this.agent

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            pass

        def __getattr__(self, item):
            return getattr(self.this, item)

        def __call__(self, name=None, **kw):
            """
            重写poco类的call方法,用于freeze()后返回结果的回调
            """

            if not name and len(kw) == 0:
                warnings.warn(
                    "Wildcard selector may cause performance trouble. Please give at least one condition to "
                    "shrink range of results")
            # 替换invalidate方法
            UIObjectProxy.invalidate = monkey_invalidate
            return UIObjectProxy(self, name, **kw)

    return FrozenPoco()


def monkey_invalidate(self):
    """
    Clear the flag to indicate to re-query or re-select the UI element(s) from hierarchy.
    """
    self._evaluated = False
    self._nodes = None
    # 判断是否是freeze的agent
    if self.poco.is_agent_freeze:
        # 重新dump
        hierarchy_dict = self.poco.old_agent.hierarchy.dump()
        hierarchy = create_immutable_hierarchy(hierarchy_dict)
        self.poco._agent = PocoAgent(hierarchy, self.poco.agent.input, self.poco.agent.screen)

Poco.freeze = monkey_freeze
poco = UnityPoco(addr, device=dev)
# 或者
# poco.freeze = types.MethodType(poco,monkey_freeze())
freeze = poco.freeze()

last_time = time.time()
nodes = poco(textMatches="Project.*")
print("poco do query time:" + str(time.time() - last_time))
last_time = time.time()
for node in nodes:
    node.get_position()
print("poco get all node position time:" + str(time.time() - last_time))

last_time = time.time()
freeze = poco.freeze()
nodes = freeze(textMatches="^Project.*")
print("freeze do query time:" + str(time.time() - last_time))
last_time = time.time()
for node in nodes:
    node.get_position()
print("freeze get all node position time:" + str(time.time() - last_time))

​ PS:由于从 freeze 下通过其 child、offspring 等这类操作获取的节点都是共享的同一个 agent,所以从 freeze 产生的任意一个 poco 对象调用 invaliate 后,其他的 poco 对象都会同步 dump,刷新节点信息时不需要全部都进行 invaliate,例如:

a = freeze("xx")
b = freeze('cc').child('ee')

b.invalidate()

print(a.poco.agent.hierarchy.dump()==b.poco.agent.hierarchy.dump()) #始终为 True

共收到 3 条回复 时间 点赞

谢谢分享,学习了。

楼主出现过 BrokenPipeError: [Errno 32] Broken pipe 这个报错吗?使用 freeze 方法时候

大雄 回复

这个错误我没遇到过,不过如果你也是用 poco 的话,按官方说法应该是端口被占用了,unity 的话根据源码你得从 5001-5006 这个范围里尝试连接 poco 的 tcp:https://github.com/AirtestProject/Poco/issues/211

需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up