仪器模拟器

实时硬件的模拟器。本项目提供了一台服务器，能够同时生成多个模拟设备并处理请求。

本项目仅提供从配置文件（YAML、TOML或json）启动服务器的所需基础设施，以及通过python入口机制注册第三方设备插件的方法。

迄今为止，本项目提供了TCP、UDP和串行线路的传输。根据需要正在实施对新的传输协议的支持（例如：USB、GPIB或SPI）。

欢迎PR！

安装

(TL;DR: pip install sinstruments[all])

在您喜欢的Python环境中

$ pip install sinstruments

此外，如果您想使用YAML编写YAML配置文件

$ pip install sinstruments[yaml]

...或者，对于基于TOML的配置

$ pip install sinstruments[toml]

执行

安装后，可以使用以下命令运行服务器

$ sinstruments-server -c <config file name>

配置文件描述了服务器应实例化哪些设备，以及一系列选项，如每个设备监听请求的传输类型。

示例

假设您需要模拟2个可通过TCP端口访问的GE Pace 5000和可通过串行线路访问的CryoCon 24C。

首先，确保使用以下命令安装了依赖项

$ pip install gepace[simulator] cryoncon[simulator]

现在我们可以准备一个名为 simulator.yml 的 YAML 配置文件。

devices:
- class: Pace
  name: pace-1
  transports:
  - type: tcp
    url: :5000
- class: Pace
  name: pace-2
  transports:
  - type: tcp
    url: :5001
- class: CryoCon
  name: cryocon-1
  transports:
  - type: serial
    url: /tmp/cryocon-1

我们现在可以启动服务器了。

$ sinstruments-server -c simulator.yml

这就完成了！您现在应该能够通过 TCP 或 CryoCon 使用本地模拟串行线连接到 Pace 的任何设备。

让我们使用 nc（即 netcat）Linux 命令行工具连接到第一个 Pace，并请求众所周知的 *IDN? SCPI 命令。

$ nc localhost 5000
*IDN?
GE,Pace5000,204683,1.01A

设备目录

这是已知第三方仪器库的总结，它们提供了自己的模拟器。

• cryocon
• fast-spinner
• gepace
• icepap
• julabo
• vacuubrand
• xia-pfcu
• Mythen 检测器（来自 Dectris）- 目前尚未公开

如果您编写了公开的设备，请自由地通过创建 PR 来完善上述列表。

提示： sinstruments-server ls 显示可用的插件列表。

配置

配置文件可以是 YAML、TOML 或 JSON 文件，只要它转换成以下描述的字典。

在本章中，我们将使用 YAML 作为参考示例。

文件应至少包含一个顶层键 devices。值需要是设备描述的列表。

devices:
  - class: Pace
    name: pace-1
    transports:
    - type: tcp
      url: :5000

每个设备描述必须包含

• class：每个第三方插件应描述用于标识自己的文本
• name：一个唯一的名称。每个设备都必须给出一个唯一的名称，由您选择
• transports：设备可访问的传输列表。大多数设备仅提供一种传输。
  • type：每个传输必须定义其类型（支持的类型有 tcp、udp、serial）
  • url：设备监听的 URL

提供给每个设备的任何其他选项都将直接在运行时传递给特定的插件对象。每个插件应描述它支持哪些附加选项以及如何使用它们。

TCP 和 UDP

对于 TCP 和 UDP 传输，url 采用 <host>:<port> 格式。

空主机（如上例所示）解释为 0.0.0.0（表示监听所有网络接口）。如果主机为 127.0.0.1 或 localhost，则设备只能从运行模拟器的机器访问。

端口号为 0 表示请求操作系统分配一个空闲端口号（用于运行测试套件）。否则必须是一个有效的 TCP 或 UDP 端口号。

串行线

url 代表由模拟器创建的特殊文件，用于模拟类似于 /dev/ttyS0 的 Linux 串行线文件可访问的串行线。

此功能仅适用于 Linux 和在 Python 中实现了伪终端 pty 的系统。

url 是可选的。模拟器将始终创建一个非确定性的名称，例如 /dev/pts/4，并在需要访问时记录此信息。此功能在运行测试套件时最有用。

您可以选择任何喜欢的 url 路径，只要您确信模拟器有权限创建符号文件。

模拟通信延迟

对于任何传输（TCP、UDP 和串行线），通过向配置提供额外的 baudrate 参数，可以模拟通信通道的速度。示例

- class: CryoCon
  name: cryocon-1
  transports:
  - type: serial
    url: /tmp/cryocon-1
    baudrate: 9600

后门

模拟器提供了一个 gevent 后门 Python 控制台，如果您想远程访问正在运行的模拟器进程，可以激活此功能。要激活此功能，只需将以下内容添加到配置文件的顶层

backdoor: ["localhost": 10001]
devices:
  - ...

您可以选择任何其他TCP端口和绑定地址。请注意，此后门不提供任何身份验证，也不会尝试限制远程用户可以做什么。任何可以访问服务器的人都可以执行正在运行的Python进程可以执行的所有操作。因此，尽管您可以绑定到任何接口，出于安全考虑，建议您仅绑定到本地机器可访问的一个接口，例如127.0.0.1/localhost。

用法

一旦配置好后门并启动服务器，在另一个终端中，通过以下方式连接

$ nc 127.0.0.1 10001
Welcome to Simulator server console.
You can access me through the 'server()' function. Have fun!
>>> print(server())
...

开发新的模拟器

编写新设备很简单。让我们假设您想模拟一个SCPI示波器。您需要做的唯一一件事就是编写一个继承自BaseDevice的类，并实现handle_message(self, message)方法，其中您应该处理您的设备支持的不同命令

# myproject/simulator.py

from sinstruments.simulator import BaseDevice

class Oscilloscope(BaseDevice):

    def handle_message(self, message):
        self._log.info("received request %r", message)
        message = message.strip().decode()
        if message == "*IDN?":
            return b"ACME Inc,O-3000,23l032,3.5A"
        elif message == "*RST":
            self._log.info("Resetting myself!")
        ...

别忘了始终返回bytes！模拟器不会对如何编码str做出任何假设

假设此文件simulator.py是名为myproject的Python包的一部分，接下来要做的是在setup.py中注册您的模拟器插件

setup(
    ...
    entry_points={
        "sinstruments.device": [
            "Oscilloscope=myproject.simulator:Oscilloscope"
        ]
    }
)

现在您可以通过编写配置文件来启动您的模拟器

# oscilo.yml

devices:
- class: Oscilloscope
  name: oscilo-1
  transports:
  - type: tcp
    url: :5000

现在通过以下方式启动服务器

$ sinstruments-server -c oscillo.yml

然后您应该能够连接

$ nc localhost 5000
*IDN?
ACME Inc,O-3000,23l032,3.5A

配置消息终止符

默认情况下，eol设置为\n。您可以使用以下方式将其更改为任何字符

class Oscilloscope(BaseDevice):

    newline = b"\r"

请求多个答案

如果您的设备实现了一种协议，该协议针对单个请求提供多个（可能延迟的）答案，您可以通过将handle_message()转换为生成器来支持此功能

class Oscilloscope(BaseDevice):

    def handle_message(self, message):
        self._log.info("received request %r", message)
        message = message.strip().decode()
        if message == "*IDN?":
            yield b"ACME Inc,O-3000,23l032,3.5A"
        elif message == "*RST":
            self._log.info("Resetting myself!")
        elif message == "GIVE:ME 10":
            for i in range(1, 11):
                yield f"Here's {i}\n".encode()
        ...

别忘了始终yieldbytes！模拟器不会对如何编码str做出任何假设

支持特定配置选项

如果您的模拟设备需要额外的配置，可以通过相同的YAML文件提供。

假设您想在启动时能够配置设备是否处于CONTROL模式。另外，如果没有配置初始值，则默认为'OFF'。

首先，让我们将其添加到我们的配置示例中

# oscilo.yml

devices:
- class: Oscilloscope
  name: oscilo-1
  control: ON
  transports:
  - type: tcp
    url: :5000

然后，我们重新实现我们的示波器__init__()以拦截此新参数，并在handle_message()中处理它

class Oscilloscope(BaseDevice):

    def __init__(self, name, **opts):
        self._control = opts.pop("control", "OFF").upper()
        super().__init__(name, **opts)

    def handle_message(self, message):
        ...
        elif message == "CONTROL":
            return f"CONTROL {self._control}\n".encode()
        ...

只要它们不与保留键name、class和transports冲突，您可以添加尽可能多的选项。

编写特定的消息协议

一些仪器实现的协议不适合由基于EOL的消息协议管理。

模拟器允许您编写自己的消息协议。以下是一个示例

from sinstruments.simulator import MessageProtocol


class FixSizeProtocol(MessageProtocol):

    Size = 32

    def read_messages(self):
        transport = self.transport
        buff = b''
        while True:
            buff += transport.read(self.channel, size=4096)
            if not buff:
                return
            for i in range(0, len(buff), self.Size):
                message = buff[i:i+self.Size]
                if len(message) < self.Size:
                    buff = message
                    break
                yield message


class Oscilloscope(BaseDevice):

    protocol = FixSizeProtocol

    ...

Pytest fixture

如果您正在开发一个提供通过套接字或串行线访问的仪器访问的Python库，并且为它编写了一个模拟器，您可能对测试您的库与模拟器感兴趣。

sinstruments提供了一对pytest辅助工具，在单独的线程中启动模拟器。

server_context

第一种用法是简单地使用server_context辅助工具。实际上，此辅助工具没有pytest特定的功能，所以您可以考虑在其他场景中使用它。

以下是一个示例

import pytest

from sinstruments.pytest import server_context

cfg = {
    "devices": [{
        "name": "oscillo-1",
        "class": "Oscilloscope",
        "transports": [
            {"type": "tcp", "url": "localhost:0"}
        ]
    }]
}

def test_oscilloscope_id():
    with server_context(cfg) as server:
        # put here code to perform your tests that need to communicate with
        # the simulator. In this example an oscilloscope client
        addr = server.devices["oscillo-1"].transports[0].address
        oscillo = Oscilloscope(addr)
        assert oscillo.idn().startswith("ACME Inc,O-3000")

您可能注意到在配置中我们使用端口0。这告诉模拟器监听操作系统提供的任何可用端口。

实际的测试检索操作系统分配的当前地址，并在测试中使用它。

如您所见，测试不依赖于特定端口的可用性，这使得它们具有可移植性。

以下是如何使用 server_context 辅助工具编写自己的 fixture 的建议。目标是减少您编写测试时所需的样板代码量。

@pytest.fixture
def oscillo_server():
    with server_context(config) as server:
        server.oscillo1 = server.devices["oscillo-1"]
        server.oscillo1.addr = server.oscillo1.transports[0].address
        yield server


def test_oscilloscope_current(oscillo_server):
    oscillo = Oscilloscope(oscillo_server.oscillo1.addr)
    assert .05 < oscillo.current() < 0.01

服务器

第二个辅助工具是 server fixture。这个 fixture 依赖于您模块中必须存在的现有 config 功能。以下是一个遵循前面代码的示例

from sinstruments.pytest import server

@pytest.fixture
def config()
    yield cfg

def test_oscilloscope_voltage(server):
    addr = server.devices["oscillo-1"].transports[0].address
    oscillo = Oscilloscope(addr)
    assert 5 < oscillo.voltage() < 10