python - 在 Linux 上将终端程序从 Bash 迁移到 Python

Question

我正在处理一个工业环境，我必须在现场记录串行设备的读数。我当前的设置涉及连接到 Moxa 5150A 或 5130 串行设备服务器的设备，该服务器根据各个设备的位置通过有线或无线方式连接到我的网络。目前，我依靠硬件供应商提供的软件来管理数据收集。当前的软件基于Windows X86，由于以前的硬件出现故障，因此在我们购买的新硬件上造成了无穷无尽的麻烦。我专门因为 Linux 驱动程序和兼容性而选择了 Moxa 串行服务器。该设备通过 RS-422(4 线半双工)7 位 1 停止奇奇偶校验和 NO 流量控制进行通信，除非有要求，否则不会发送任何信息。

我正在开发自己的解决方案，将信息存储在数据库(可能是 MySQL)中。就终端通信而言，这是两部分。我必须发送请求并接收响应。

到目前为止，我已将串行服务器的 Moxa 驱动程序安装到我的 Linux 服务器上。我编写了一个 bash 脚本，该脚本从/dev/ttyrXX 设备读取、解释输入并将信息加载到 mysql 数据库中。

      while read line
      do
      something something....
      done < /dev/ttyrXX

我仍在解决一些小的脚本错误，例如当设备离线然后重新上线时正确重新启动以及其他此类问题，但我的脚本正在读取和记录我的数据，完全符合我的需求。

该解决方案依赖于两个因素。

1) Moxa 设备支持 FIFO 并允许同时进行多个连接

2) 供应商提供的在单独的 Windows 计算机上运行的软件当前正在提交信息请求。

我需要将基于 Windows 的软件全部删除。我已经编写并测试了一个脚本，该脚本打印到/dev/ttyrXX 发出我想要的请求，并收到适当的响应，并根据我的第一个脚本的需要记录它们。

     some list of things
     for x in list
     do         
     printf "request" > /dev/ttyrXX
     done

我在使用供应商提供的软件时遇到的主要问题之一是其硬编码依赖于接收对每个请求的及时响应。这会产生许多问题，包括完全卡住窗口以及设备在不适当的时间停止服务时崩溃。

我的两个脚本协同工作，以明显更好的时间间隔收集数据，并且人工干预更少。

这是我的问题。我可以在 Python 中使用什么方法来获得与 Bash 脚本相当的结果？

我一直无法找到相当于 printf 或 cat 的 Python，它不需要我打开/关闭文件。此外，我看过的每个终端界面都有内置的或必需的等待响应或读取很长时间才放弃。我确信那里存在着某种东西，但我只是没有运气找到它。我承认我对 Python 还很陌生，只写了几行代码来探索它的能力。

让我说得非常清楚。我想在编写时写入我的终端而不用担心其他任何事情(使用我不关心的线程或单独的脚本)。我想从我的终端上阅读，而不关心下一段文本何时到来。 (我并不是建议一个永无止境的循环，但我确实需要它来简单地等待输入)。我希望/需要我的写入和读取完全相互独立，而不关心对方昨天、现在或下周在做什么。

提前谢谢您。

如果有必要，我将能够提供更详细的代码，但是我觉得我在搜索中更需要指导而不是错误修复。尽管我仍然完全愿意接受建议，并输入“永远不要这样做......因为这样......”输入内容

Answer 1

扩展伊格纳西奥所说的内容。听起来最简单的解决方案是有两个在启动时启动的脚本。

其中一个打开打开tty设备并永远监听它。这是阻塞读。它继续寻找新的输入行，直到收到(或到达)EOF。想必在这种情况下这永远不会发生。它看起来像这样:

# tty reader

def listen_to_device():
    with open('/dev/ttyXX', 'r') as tty:
        for line in tty:
            create_db_record(line)

if __name__ == '__main__':
    listen_to_device()

要使代码成为完整的工作位，您唯一需要实现的是“create_db_record”函数。

第二个脚本可能非常相似。但也许它会继续在目录中查找包含命令的特定文件。当它找到任何一个时，它会将它们一个接一个地发送到 tty 设备。

# tty controller

import os
import shutil
import time

def wait_for_commands():
    while True:
        if os.path.exists('/tmp/commands.txt'):
            with open('/tmp/commands.txt', 'r') as c:
                for command in c:
                    send_command(command)
            shutil.move('/tmp/commands.txt', 'tmp/proccesed-' str(datetime.utcnow()))
        time.sleep(1)


def send_command(command):
    with open('/dev/ttyXX', 'w') as tty:
        # maybe do some additional changes to command here?
        tty.write(command)


if __name__ == '__main__':
    wait_for_command()

您可以在 send_command 函数中添加一些额外的处理代码，以便commands.txt 文件中的指令变得非常简单。