小编jan*_*jan的帖子

Python项目使用协议缓冲区,部署问题

我有一个使用setuptools进行部署的Python项目,我大多数都遵循本指南的项目结构.该项目使用Google Protocol Buffers定义网络消息格式.我的主要问题是如何在安装过程中使setup.py调用protoc-compiler以将定义构建到_pb2.py文件中.

这个问题中,建议只是将生成的_pb2.py文件与项目一起分发.虽然这可能适用于非常类似的平台,但我发现有几种情况不适用.例如,当我在使用Anaconda Python的Mac上进行开发并将生成的_pb2.py与项目的其余部分一起复制到运行Raspbian的Raspberry Pi时,总会有来自_pb2.py模块的导入错误.但是,如果我在Pi上新编译.proto文件,项目将按预期工作.因此,分发编译的文件似乎不是一个选项.

在这里寻找工作和最佳实践解决方案.可以假设protoc-compiler安装在目标平台上.

编辑:

由于人们询问失败的原因.在Mac上,protobuf版本是2.6.1.在Pi上它是2.4.1.显然,生成的protoc编译器输出使用的内部API已更改.输出基本上是:

  File "[...]network_manager.py", line 8, in <module>
      import InstrumentControl.transports.serial_bridge_protocol_pb2 as protocol
  File "[...]serial_bridge_protocol_pb2.py", line 9, in <module>
      from google.protobuf import symbol_database as _symbol_database
  ImportError: cannot import name symbol_database
Run Code Online (Sandbox Code Playgroud)

python installation software-distribution setuptools protocol-buffers

11
推荐指数
1
解决办法
7904
查看次数

Python单元测试有两个模拟对象,如何验证调用顺序?

我正在编写一个编排两个仪器(一个可远程控制的电源单元和一个用于控制被测器件的总线控制器)的课程,以便在被测器件(DUT)上执行各种测量.

对两个工具的访问都是作为Python类实现的,并且对每个工具的引用都可用于新类.DUT有些微妙,并且具有非常特定的上电序列,涉及对电源和总线控制器的调用,并且必须以特定顺序发生以避免损坏DUT.

现在我想为这个类写一个单元测试.我目前正在使用nosetests和mock-package.所以我的想法是模拟两个乐器类并验证那些正确的调用顺序.

验证每个模拟类本身的调用顺序似乎很容易.因此,我能够找出电源是否被命令首先正确地施加电池电压,然后是数字域电压然后是模拟域电压.我还可以发现数字寄存器已被编程为正确的值.但是,我很难确定在应用数字域电压和模拟域电压之间是否发生了写入数字寄存器的调用.

所以我的问题是:如果我有两个模拟对象,我如何验证这些对象之间的特定调用顺序?我的第一个问题是检查呼叫的时间戳,但那些似乎不存在.

python unit-testing mocking

10
推荐指数
1
解决办法
1362
查看次数

在 Python 中正确丢弃指向 mmap 内存的 ctypes 指针

我遇到的问题是,在创建指向 mmap-s 的指针后,我无法在 Python 中正确关闭 mmap-s。我的用例是打开文件(通常是与硬件一起工作的 UIO 设备,但普通文件也会出现此问题),对它们进行内存映射,然后将它们用作 ctypes 结构的缓冲区。通常是数据结构或数组。一个最小的例子如下所示:

import ctypes as ct
import mmap
import os

fileno = os.open('/tmp/testfile', os.O_RDWR | os.O_SYNC)
map = mmap.mmap(fileno, 32768, flags=mmap.MAP_SHARED)
memory = (ct.c_uint32 * 8192).from_buffer(map)

# Use the memory object to do things here

del memory
map.close()
os.close(fileno)
Run Code Online (Sandbox Code Playgroud)

那时一切都很好。

但是,有时我需要调用一些也需要访问该内存的 C 库函数,因此我必须传入指向它们的指针。我使用以下方法创建该指针:

ptr = ct.cast(memory, ct.c_void_p)
Run Code Online (Sandbox Code Playgroud)

除了一件事之外,所有这些都运作良好。一旦我创建了这样的指针,我就无法再关闭内存映射。举个稍微扩展一下的例子:

import ctypes as ct
import mmap
import os

fileno = os.open('/tmp/testfile', os.O_RDWR | os.O_SYNC)
map = mmap.mmap(fileno, 32768, flags=mmap.MAP_SHARED)
memory = (ct.c_uint32 * 8192).from_buffer(map) …
Run Code Online (Sandbox Code Playgroud)

python ctypes pointers mmap memoryview

7
推荐指数
1
解决办法
3271
查看次数

谷歌protobuf和大型二进制blob

我正在构建一个软件来远程控制连接到另一台PC的无线电硬件.

我计划使用ZeroMQ进行传输,并使用类似RPC的请求 - 回复,并在其上面显示代表操作的不同消息.

虽然我的大多数消息只是一些控制和状态信息,但应该有一个选项来设置要传输的数据blob或请求接收的blob数据.这些数据blob通常在5-10MB范围内,但也应该可以使用大到100MB的更大的blob.

对于消息格式,我发现谷歌协议缓冲区非常吸引人,因为我可以在传输链接上定义一个消息类型,它具有所有命令和响应的可选元素.但是,protobuf常见问题解答表明,如此大的消息会对性能产生负面影响.

所以问题是,它实际上有多糟糕?会有什么负面影响?我真的不想将整个通信基于protobuf,只是为了发现它不起作用.

rpc protocol-buffers

5
推荐指数
2
解决办法
1万
查看次数

Sympy复指数的绝对值

当处理极地形式的复数时,我经历了一种奇怪的行为.例如,做

from sympy import *
simplify(Abs(exp(I)))
Run Code Online (Sandbox Code Playgroud)

我期望结果1,因为如果指数只是虚数,复指数的绝对值应该总是1.然而,sympy给出了答案

Abs(exp(I))
Run Code Online (Sandbox Code Playgroud)

做另的选择

phi=symbols('phi', real=True)
y=exp(I*phi)
sqrt(y*conj(y))
Run Code Online (Sandbox Code Playgroud)

给出了预期的结果,但在我看来不如绝对.我是否错过了一些限制,以防止只使用abs时症状执行此简化?

python sympy simplification

5
推荐指数
1
解决办法
1752
查看次数

使用Python Asyncio等待GPIO中断

我正在尝试将程序从 Tornado 转换为 Asyncio,第一步是使用实际的 asyncio eventloop,如此处所述

该应用程序在嵌入式 Linux 机器上运行,我通过sysfs/gpio 子系统使用 GPIO ,并且在其中一些 GPIO 上我正在等待中断。我可以通过执行以下操作将其直接集成到 Tornado IOLoop 中:

# Register with the queue
self.io_loop.add_handler(
    self.gpio._value_file, self._handle_interrupt, self.io_loop._EPOLLPRI | self.io_loop.ERROR
)
Run Code Online (Sandbox Code Playgroud)

在代码段中,_value_file是可以读取 GPIO 的文件的文件句柄。只要 GPIO 上有可用中断,就会触发 EPOLLPRI 事件。在龙卷风上,这非常有效。_handle_interrupt它会在中断到来后不久调用该函数。

我的问题是,我无法将其转换为本机异步事件循环。在监视文件描述符的文档中,我只找到了添加读取器和写入器的函数,但没有发现文件描述符上的通用事件掩码。我无法深入研究代码,因为它进入了 C。但是,查看 Tornado 层以将 Tornado IOLoop 的调用转换为 asyncio IOLoop ,情况似乎是这样:

def add_handler(self, fd, handler, events):
    fd, fileobj = self.split_fd(fd)
    if fd in self.handlers:
        raise ValueError("fd %s added twice" % fd)
    self.handlers[fd] = (fileobj, stack_context.wrap(handler)) …
Run Code Online (Sandbox Code Playgroud)

python linux interrupt python-3.x python-asyncio

4
推荐指数
1
解决办法
1710
查看次数