python - 在树莓派上模拟 Gpio 输入进行测试

Question

我有一个在我的 RPi 上运行的 python 脚本。它使用 Gpiozero 库(顺便说一句，这真的很棒)。

出于测试目的，我想知道是否有可能以某种方式模拟 GPIO 状态(比如模拟按下按钮)并让 gpiozero 库获取它。

谢谢!

Answer 1

TLDNR:是的，这是可能的。

我不知道有任何已经准备好的解决方案可以帮助您实现您想要做的事情。因此，我发现它是否可行非常有趣。

我一直在寻找可用于 stub GPIO 功能的接缝，我发现 gpiozero uses GPIOZERO_PIN_FACTORY 用于选择后端的环境变量。计划是写自己的pin factory，这将提供测试其他脚本的可能性。

注意:请将我的解决方案视为概念证明。它远未做好生产准备。

想法是在测试范围内从脚本中获取 GPIO 状态。我的解决方案使用环境变量 RPI_STUB_URL 获取 unix 套接字的路径，该套接字将用于与 stub Controller 通信。

我已经为每个连接协议(protocol)引入了一个非常简单的请求/响应:

• "GF {pin}\n"- 询问 pin 的当前功能是什么。 Stub 不验证响应，但我希望使用“输入”、“输出”。
• "SF {pin} {function}\n"- 请求更改引脚的当前功能。 Stub 不验证函数，但我希望使用“输入”、“输出”。 Stub 期望“OK”作为响应。
• "GS {pin}\n"- 询问 pin 的当前状态。 Stub 需要值“0”或“1”作为响应。
• "SS {pin} {value|]n"- 请求更改引脚的当前状态。 Stub 期望“OK”作为响应。

我的“ stub 包”包含以下文件:

- setup.py # This file is needed in every package, isn't it?
- rpi_stub/ 
   - __init__.py # This file collects entry points
   - stubPin.py # This file implements stub backend for gpiozero
   - controller.py # This file implements server for my stub
   - trigger.py # This file implements client side feature of my stub

让我们从setup.py内容开始:

from setuptools import setup, find_packages

setup(
    name="Raspberry PI GPIO stub",
    version="0.1",
    description="Package with stub plugin for gpiozero library",
    packages=find_packages(),
        install_requires = ["gpiozero"],
    include_package_data=True,
    entry_points="""
[console_scripts]
stub_rpi_controller=rpi_stub:controller_main
stub_rpi_trigger=rpi_stub:trigger_main
[gpiozero_pin_factories]
stub_rpi=rpi_stub:make_stub_pin
"""
)

它定义了两个 console_scripts 入口点，一个用于 Controller ，一个用于触发器。以及一个用于 gpiozero 的引脚工厂。

现在rpi_stub/__init__.py:

import rpi_stub.stubPin 
from rpi_stub.controller import controller_main
from rpi_stub.trigger import trigger_main

def make_stub_pin(number):
    return stubPin.StubPin(number)

这是一个相当简单的文件。

文件rpi_stub/trigger.py:

import socket
import sys

def trigger_main():
   socket_addr = sys.argv[1]
   sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
   sock.connect(socket_addr)
   request = "{0}\n".format(" ".join(sys.argv[2:]))
   sock.sendall(request.encode())
   data = sock.recv(1024)
   sock.close()
   print(data.decode("utf-8"))

trigger 允许您发出自己的请求。您可以使用它来检查 GPIO 引脚的状态或更改它。

文件rpi_stub/controller.py:

import socketserver
import sys

functions = {}
states = {}

class MyHandler(socketserver.StreamRequestHandler):

    def _respond(self, response):
        print("Sending response: {0}".format(response))
        self.wfile.write(response.encode())

    def _handle_get_function(self, data):
        print("Handling get_function: {0}".format(data))
        try:
          self._respond("{0}".format(functions[data[0]]))
        except KeyError:
          self._respond("input")

    def _handle_set_function(self, data):
        print("Handling set_function: {0}".format(data))
        functions[data[0]] = data[1]
        self._respond("OK")

    def _handle_get_state(self, data):
        print("Handling get_state: {0}".format(data))
        try:
          self._respond("{0}".format(states[data[0]]))
        except KeyError:
          self._respond("0")

    def _handle_set_state(self, data):
        print("Handling set_state: {0}".format(data))
        states[data[0]] = data[1]
        self._respond("OK")

    def handle(self):
        data = self.rfile.readline()
        print("Handle: {0}".format(data))
        data = data.decode("utf-8").strip().split(" ")

        if data[0] == "GF":
            self._handle_get_function(data[1:])
        elif data[0] == "SF":
            self._handle_set_function(data[1:])
        elif data[0] == "GS":
            self._handle_get_state(data[1:])
        elif data[0] == "SS":
            self._handle_set_state(data[1:])
        else:
            self._respond("Not understood")

def controller_main():
    socket_addr = sys.argv[1]
    server = socketserver.UnixStreamServer(socket_addr, MyHandler)
    server.serve_forever()

此文件包含我能够编写的最简单的服务器。

还有最复杂的文件rpi_stub/stubPin.py:

from gpiozero.pins import Pin
import os
import socket
from threading import Thread
from time import sleep

def dummy_func():
   pass

def edge_detector(pin):
   print("STUB: Edge detector for pin: {0} spawned".format(pin.number))
   while pin._edges != "none":
      new_state = pin._get_state()
      print("STUB: Edge detector for pin {0}: value {1} received".format(pin.number, new_state))
      if new_state != pin._last_known:
          print("STUB: Edge detector for pin {0}: calling callback".format(pin.number))
          pin._when_changed()
      pin._last_known = new_state 
      sleep(1)
   print("STUB: Edge detector for pin: {0} ends".format(pin.number))


class StubPin(Pin):

    def __init__(self, number):
        super(StubPin, self).__init__()
        self.number = number
        self._when_changed = dummy_func
        self._edges = "none"
        self._last_known = 0

    def _make_request(self, request):
        server_address = os.getenv("RPI_STUB_URL", None)
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(server_address)
        sock.sendall(request.encode())
        data = sock.recv(1024)
        sock.close()
        return data.decode("utf-8")

    def _get_function(self):
        response = self._make_request("GF {pin}\n".format(pin=self.number))
        return response;

    def _set_function(self, function):
        response = self._make_request("SF {pin} {function}\n".format(pin=self.number, function=function))
        if response != "OK":
          raise Exception("STUB Not understood", response)

    def _get_state(self):
        response = self._make_request("GS {pin}\n".format(pin=self.number))
        if response == "1":
           return 1
        else:
           return 0

    def _set_pull(self, value):
         pass

    def _set_edges(self, value):
         print("STUB: set edges called: {0}".format(value))
         if self._edges == "none" and value != "none":
             self._thread = Thread(target=edge_detector,args=(self,))
             self._thread.start()
         if self._edges != "none" and value == "none":
             self._edges = value;
             self._thread.join()
         self._edges = value
         pass

    def _get_when_changed(self, value):
         return self._when_changed

    def _set_when_changed(self, value):
         print("STUB: set when changed: {0}".format(value))
         self._when_changed = value

    def _set_state(self, value):
        response = self._make_request("SS {pin} {value}\n".format(pin=self.number, value=value))
        if response != "OK":
          raise Exception("Not understood", response)

该文件定义了 StubPin，它从 gpiozero 扩展了 Pin .它定义了所有必须被覆盖的功能。它还包含非常简单的边缘检测，因为它是 gpio.Button 工作所必需的。

让我们做一个演示 :)。让我们创建安装了 gpiozero 和我的包的 virtualenv:

$ virtualenv -p python3 rpi_stub_env
[...] // virtualenv successfully created
$ source ./rpi_stub_env/bin/activate
(rpi_stub_env)$ pip install gpiozero
[...] // gpiozero installed
(rpi_stub_env)$ python3 setup.py install
[...] // my package installed

现在让我们创建 stub Controller (在其他终端等中打开):

(rpi_stub_env)$ stub_rpi_controller /tmp/socket.sock

我将使用以下脚本 example.py:

from gpiozero import Button                                                                                         
from time import sleep                                                                                              

button = Button(2)

while True:
    if button.is_pressed:
       print("Button is pressed")
    else:
       print("Button is not pressed")
    sleep(1)

让我们执行它: (rpi_stub_env)$ RPI_STUB_URL=/tmp/socket.sock GPIOZERO_PIN_FACTORY=stub_rpi python example.py

默认情况下，脚本打印按钮被按下。现在让我们按下按钮:

(rpi_stub_env)$ stub_rpi_trigger /tmp/socket.sock SS 2 1

现在脚本应该打印出按钮没有被按下。如果您执行以下命令，它将再次被按下:

(rpi_stub_env)$ stub_rpi_trigger /tmp/socket.sock SS 2 0

希望对您有所帮助。