多头畸形

Polycephaly 是一个 Python 模块，可让您轻松创建能够在 I/O 和 CPU 级别进行并行操作的程序：

• 用于进程间通信的类似电子邮件的语法，没有 SQLite 或 Redis 等外部依赖项。
• 消息路由（例如在 Polycephaly 下运行的进程与其他平台（如 Unix Domain Socketfile、D-Bus、MQTT、USB 等）之间）
• 用于分配回调以接收不同路由上的消息的消息过滤器。
• 用于外部消息传递的 JSON 编码和解析（例如 Unix Domain Socketfile 或 MQTT）。
• 先进先出或优先消息排队。
  • 优先消息队列仍在开发中，因为优先队列仅在线程模式下工作。

概述

最初，Polycephaly 是作为一个共享框架创建的，用于为基于机器人操作系统的机器人构建基于 Linux 的嵌入式系统，并带有一个带有 Python 的大型触摸屏显示器。该框架在作为服务器和客户端串联运行的 2 个独立进程之间共享：

1. 系统服务
   • 共享消息库（例如 Unix Domain Socketfile Server）。
   • 系统和网络管理：
     • 用于远程监控和管理的持久 Internet 连接。
     • 使用机会性停机时间应用系统更新（例如，在高峰使用和工作时间之外重新启动 UEFI、Grub 和/或内核更新）。
   • 软件和硬件通信，包括但不限于：
     • MQTT
     • 工业PC
       • Unix 域套接字
       • 消息队列
       • 管道
     • USB
     • TTL（通过 USB）
2. 触摸屏应用
   • 共享消息库（例如 Unix 域 Socketfile 客户端）。
   • 管理功能和硬件诊断。
   • 通过二维码与移动设备交互。

请注意：此模块不应被视为生产就绪，许多改进仍在进行中。

由于朋友们表示他们可以在他们的一些项目中使用它，所以我发布的时间比我应该的要早得多。如果您发现错误或有建议，请考虑打开错误报告，或者更好的是，提交拉取请求。

用法

得到

下载多头症：

通过以下方式从PyPIpip安装：

$ pip install polycephaly

或者

通过以下方式从该存储库pip安装：

$ pip install git+ssh://git@gitlab.com/ltgiv/polycephaly.git

或者

克隆此存储库：

$ git clone https://gitlab.com/ltgiv/polycephaly.git

格罗克

多头症是 2 个组件，它们是相同的派生类以保持一致性。因此，每个过程都是可以互换的。

成分

最起码有主流程，再有子流程。

1. 主进程 由于许多 TUI/GUI 框架（例如Urwid、curses、Kivy、pyglet、PyQt、Tkinter等）作为（或必须是）主进程工作得更好，这很容易实现。如果此过程结束，则所有子过程都将关闭。

2. 子流程 这些回答主要流程。设置应用程序时，每个子进程都有一个模式，可以在线程或分叉之间切换，前者是默认模式，后者更适合用于 CPU 密集型任务（例如训练机器学习模型）或需要时（例如rospy需要这个。）

对于操作模式设置为分叉的子进程，它们将在单独的 Python 进程下独立运行，并且不能像线程化时那样共享现有对象（例如字典）。这就是 Polycephaly 类似电子邮件的通信方式大放异彩的地方，因为它允许您传递可序列化的对象。

如果您想共享一个对象，例如网络或数据库连接，您只需将一个子进程用于此任务，然后使用 Polycephaly 来自其他进程的通信来执行它们的请求，包括等待的能力回复。

方法

进程的生命周期分为 3 个阶段：

1. birth( self )- 您可以将其视为构造函数。这在循环方法之前执行初步操作（例如初始化硬件或连接到数据库）life()。
   • args- 从应用程序的构建端传递到进程的参数元组。
   • kwargs- 从应用程序的构建端传递到进程的关键字参数字典。
2. life( self )- 这是整个过程。在构建基于多头症的应用程序时，您可以调整全局和/或局部频率值，这决定了该方法应该运行的频率。例如，如果您将此值设置为 30，Polycephaly 将尝试每秒运行此方法 30 次，每次运行life()定义为time.sleep( 1 / 30 )
3. death( self )- 你可以把它想象成一个析构函数。这会在通知主进程生命周期结束并完全关闭之前执行进程的清理部分（例如，完成数据库事务，然后关闭数据库连接）。

许多辅助方法是从父类继承的，用于每个流程定义。API 文档提供了更多信息，但需要注意的一些常见信息是：

• frequency( i=None )- 如果没有提供参数，这将返回本地进程频率的当前整数。使用参数，这将写入一个用于频率的整数。如果实例布尔值被切换，则增加和减少频率的示例：

  if self.stayAlert and self.frequency() < 60:
      self.frequency( 60 )
  
  elif self.frequency() != 30:
      self.frequency( 30 )
  

• mailman()- 在不提供参数的情况下，这将检查来自主进程或子进程的内部消息总线上的进程的新消息。

  • 如果过滤器匹配消息，则执行回调。这通常是一个阻塞事件，但如果需要，可以将回调分离到线程或分叉进程中。
  • 每次运行只能从进程的队列中读取一条消息。
  • 每个循环只运行一次life()。
  • 这可以轻松扩展以涵盖其他消息队列和路由，例如检查从 MQTT、XMPP 或 USB 接收的（基于 JSON 的）消息。

• die()- 被进程用来关闭自己，并death()作为最后一步。

  在应用程序的构建阶段，会生成一个毒丸，它只是一个保存为字符串的 UUID。然后为每个进程自动添加一个消息过滤器，在内部路由的主题行中查找该值。当一个进程从主进程或子进程接收到这个值时，接收进程将运行它的die()方法。

• ebrake( reason=None )- 顾名思义，这种紧急制动方法允许应用程序中的任何进程向主进程发送请求，请求立即关闭应用程序，并允许将显示为关闭消息的一部分的可选原因.

• send()- 从一个进程向另一个进程发送消息。一个例子：

  self.send(
      recipient   =   "main",
      subject     =   "salutation",
      body        =   "Hello, World!"
  )
  

• waitForReply()- 使用超时选项阻止事件，在继续之前等待收件人的响应。一个例子：

  message =   self.send(
                  recipient   =   "main",
                  subject     =   "salutation",
                  body        =   "Hello, World!"
              )
  
  reply   =   self.waitForReply( message, timeout=10 )
  

去！

这是“Hello, World!”的高级概述。例子。示例目录通常是完整功能代码的最佳起点：

建造

launch.py：

#!/usr/bin/env python -u
# -*- coding: utf-8 -*-

class Application( polycephaly.Application ):

    def build( self ):

        # Update global frequency
        self.globalFrequency( 15 )              # Run fifteen times per second.

        # Add process : Hello, World!
        self.addProcess(

            processes.helloWorld,               # If the default class name of `Process` is used, it doesn't need to be specified here.

            # Arguments to pass through to the process.
            'Arg1',
            'Arg2',

            # Keyword arguments to pass through to the process.
            abc             =   123,
            xyz             =   789,

            # Process Parameters
            name            =   'Hello',        # Override the default name of `helloWorld` with a shorter name of `hello`.
            mode            =   'Thread',       # Run the process as a thread.
            frequency       =   1 / 5,          # Update the local frequency to run once every 5 seconds.
            autostart       =   True,           # This is default behavior, with the alternative being to setup a process, and then start it at a later time.
            boundShutdown   =   False,          # Run independently without binding to main process.

        )

        pass # END METHOD : Build

    pass # END CLASS : Application

if __name__ == '__main__':

    logger.notice( "Start : 'Hello, World!'." )

    Application(

        # Add process : Main
        processes.main,
        name            =   'Main',         # Set the name of the process that we refer to, or specify an added process as main.
        ppill           =   'STOP!',        # Case-sensitive poison pill.
        queueSize       =   25,             # Maximum number of messages to keep in each queue.
        queueType       =   'FIFO',         # FIFO or Priority message queue.
        frequency       =   5,              # Update the local frequency to run five times per second.
        forceStop       =   False,          # Allow the process to ignore repeated shutdown requests.
        threadsTimeout  =   30,             # Application will wait on threads for this long.

    ).run()

    logger.notice( "Stop : 'Hello, World!'." )

    pass # END MAIN

helloWorld.py

#!/usr/bin/env python -u
# -*- coding: utf-8 -*-

class Process( polycephaly.Process ):

    def life( self ):

        # Send a message to the Main process.
        message     =   self.send(

                            # Message parameters
                            recipient   =   'main',
                            subject     =   'salutation',
                            body        =   'Hello, World!',

                            # Extra message headers
                            args        =   self.args,
                            kwargs      =   self.kwargs,

                        )
        logger.debug( f"'{ self.name }' sent a message to '{ message[ 'recipient' ] }':\n{ pf( message ) }" )

        # Wait for reply from the Main process.
        reply       =   self.waitForReply( message, timeout=10 )
        logger.debug( f"'{ self.name }' received a reply from '{ reply.get( 'sender' ) }':\n{ pf( reply ) }" )

        # Check for new messages, and run appropriate callbacks.
        self.mailman()

        pass # END METHOD : Life

    pass # END CLASS : PROCESS : Hello, World!

main.py

#!/usr/bin/env python -u
# -*- coding: utf-8 -*-

class Process( polycephaly.Process ):

    # This is a callback method that replies back to a message.
    def salutation( self, message ):

        logger.debug( f"'{ self.name }' received a salutation message from '{ message[ 'sender' ] }':\n{ pf( message ) }" )

        # Respond to a received message.
        reply   =   self.reply(

                        # Message parameters
                        message,
                        body    =   "Hi, thanks for writing.",

                        # Extra message headers
                        acme    =   123,

                    )
        logger.debug( f"'{ self.name }' sent a reply to '{ reply[ 'recipient' ] }':\n{ pf( reply ) }" )

        pass # END CALLBACK : Salutation

    def birth( self ):

        # Add a message filter for case-insensitive matching of "salutation" in the subject line, with self.salutation() set as the callback method.
        self.addFilter(
            subject     =   r'(?i)^SALUTATION$',
            callback    =   self.salutation,
        )

        pass # END METHOD : Birth

    pass # END CLASS : PROCESS : Main

跑

运行这个应用程序很简单~/helloWorld/launch.py

现在，每 5-6 秒，您将在控制台上看到输出：

[1970-01-01 00:00:00.00] DEBUG: processes.helloWorld: 'hello' sent a message to 'main':
{
    'args': ('Arg1', 'Arg2'),
    'body': 'Hello, World!',
    'kwargs': {'abc': 123, 'xyz': 789},
    'messageid': '9da03bcf-af3d-49ad-a9d4-17617c974de6',
    'recipient': 'main',
    'sender': 'hello',
    'subject': 'salutation',
    'threadid': '05cc7e85-f291-4c63-af34-6d47b8fc2594',
    'threadindex': 1,
    'time': 123456789.0000
}

[1970-01-01 00:00:00.05] DEBUG: processes.main: 'main' received a salutation message from 'hello':
{
    'args': ('Arg1', 'Arg2'),
    'body': 'Hello, World!',
    'kwargs': {'abc': 123, 'xyz': 789},
    'messageid': '52c7cca0-0f0a-4791-a1d8-7de96d5b2c68',
    'recipient': 'main',
    'sender': 'hello',
    'subject': 'salutation',
    'threadid': '05cc7e85-f291-4c63-af34-6d47b8fc2594',
    'threadindex': 1,
    'time': 123456789.0005
}

[1970-01-01 00:00:00.10] DEBUG: processes.main: 'main' sent a reply to 'hello':
{
    'acme': 123,
    'args': ('Arg1', 'Arg2'),
    'body': 'Hi, thanks for writing.',
    'kwargs': {'abc': 123, 'xyz': 789},
    'messageid': '270c8371-0467-4648-9630-5376e33ababa',
    'recipient': 'hello',
    'sender': 'main',
    'subject': 'reply',
    'threadid': '05cc7e85-f291-4c63-af34-6d47b8fc2594',
    'threadindex': 2,
    'time': 123456789.0010
}

[1970-01-01 00:00:00.15] DEBUG: processes.helloWorld: 'hello' received a reply from 'main':
{
    'acme': 123,
    'args': ('Arg1', 'Arg2'),
    'body': 'Hi, thanks for writing.',
    'kwargs': {'abc': 123, 'xyz': 789},
    'messageid': '270c8371-0467-4648-9630-5376e33ababa',
    'recipient': 'hello',
    'sender': 'main',
    'subject': 'reply',
    'threadid': '05cc7e85-f291-4c63-af34-6d47b8fc2594',
    'threadindex': 2,
    'time': 123456789.0015
}

解释

由于我们设置上述频率的方式，主要过程是每秒检查五次消息，而作为“你好，世界！” 子进程每 5 秒发送一条消息。

除了您可以添加的任意标头（例如args，kwargs在send()上面的调用中）之外，还会向此消息添加其他标头（例如，发件人、UTC时间、消息 ID 和线程 ID），这些标头可用于其他目的，例如等待回复。就像电子邮件一样，Message ID 不同，但 Thread ID 相同，并且 Thread Index 增加了。

贡献者

• 路易斯·盖特曼四世

对这个项目有什么贡献吗？请参阅文档，如何贡献。

统计数据

链接

• 包裹
• 文档
• 资源
• 追踪器

执照

MIT License

Copyright (c) 2019 Louis T. Getterman IV

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

用StackEdit 编写。