Name: Twico Control Engine
Creator: DaRUS
Published: 2026-01-22 13:36:41
License: 暂无描述

下载链接：

https://darus.uni-stuttgart.de/citation?persistentId=doi:10.18419/DARUS-5649

下载链接

链接失效反馈

官方服务：

资源简介：

<!doctype html> <html lang="en"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1" /> <title>Twico Control Engine</title> </head> <body> <h1>Twico Control Engine</h1> <p> Twico Control is a <strong>Flask-based orchestration engine</strong> for <em>digital materialisation</em> / <em>digital twin</em> workflows. It coordinates <strong>virtual actors</strong> (software representations) and their paired <strong>physical actors</strong> (robots, tools, sensors), executes <strong>tasks</strong> in a controlled order, forward information to external services, and optionally synchronises state of the local DB with a <strong>cloud database via REST API</strong>. A web UI is included for managing actors, tasks, jobs, and monitoring execution. </p> <blockquote> <p><strong>Status</strong>: research / prototyping-friendly codebase. Contributions and issues are welcome.</p> </blockquote> <hr /> <h2>What problem it solves</h2> <p>In a typical digital materialisation process you have:</p> <ul> <li><strong>Tasks</strong> to execute (move, read data, fabricate, etc.)</li> <li><strong>Physical actors</strong> that do the work (robots, PLCs, devices)</li> <li><strong>Virtual actors</strong> that encapsulate communication + pre/post-processing</li> <li><strong>Services</strong> that generate tasks or react to task completion</li> </ul> <p> Twico Control acts as the <strong>execution core</strong>: it keeps a local “authoritative” state, dispatches tasks to actors, listens for events, and updates external systems. </p> <hr /> <h2>Key concepts</h2> <ul> <li> <strong>Actor (Virtual Actor)</strong>: a Python class that implements a standard interface (<code>actors/ActorBase.py</code>) and handles: connection, task preparation (<code>preSend</code>), task dispatch, completion monitoring, and response handling (<code>postSend</code>). </li> <li> <strong>Task</strong>: an executable unit that is assigned to a <code>main_actor</code> with extra per-actor parameters under <code>actors_data</code>. </li> <li> <strong>Engine</strong>: a Flask server that runs the UI + API endpoints and coordinates task execution. </li> <li> <strong>RabbitMQ</strong>: optional event bus for “task received” / “task complete” style workflows. </li> </ul> <hr /> <h2>Execution model (high level)</h2> <p>The engine runs continuously and coordinates execution across actors:</p> <ol> <li>Ensure all required virtual actors are registered and connected.</li> <li> For each available actor: <ul> <li>fetch the next task from the local DB (or tasks queue)</li> <li>call <code>preSend(task)</code> then send the task to the paired physical actor</li> <li>wait for completion / acknowledgements (depending on your communication pattern)</li> <li>call <code>postSend(response)</code> and log results</li> <li>trigger any linked services</li> </ul> </li> <li> In parallel, background listeners can: <ul> <li>ingest new tasks (e.g. via RabbitMQ)</li> <li>propagate task completion updates to a cloud DB (if REST API is configurated)</li> </ul> </li> </ol> <hr /> <h2>Project structure</h2> <pre><code>TWICO_Release/ ├── actors/ │ ├── __init__.py │ ├── ActorBase.py │ ├── RMQ_setup.py │ ├── DemoActor/ │ │ ├──__init__.py │ │ └── DemoActor.py │ ├── ServiceConnectedActor/ │ │ ├──__init__.py │ │ └── ServiceConnectedActor.py │ └── TaskInjectingActor/ │ │ ├──__init__.py │ └── TaskInjectingActor.py ├── apps/ │ ├── actors/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── authentication/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── home/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── routes.py │ │ └── services.py │ ├── Logger/ │ │ ├──__init__.py │ │ ├── logs/ │ │ ├── Logger.py │ ├── RabbitMQ/ │ │ ├──__init__.py │ │ └── listener.py │ ├── run/ │ │ ├──__init__.py │ │ ├── routes.py │ │ └── services.py │ ├── static/ │ │ └── assets/ │ ├── tasks/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── templates/ │ │ ├── actors/ │ │ ├── auth/ │ │ ├── dashboard/ │ │ ├── includes/ │ │ ├── layouts/ │ │ ├── run/ │ │ └── tasks/ │ ├── .env.example │ ├── __init__.py │ └── config.py ├── Examples/ │ ├── Basic/ │ │ ├── basic_tasks.py │ ├── Parallel_actors/ │ │ ├── parallel_actors_tasks.py │ ├── Self_injecting_actor/ │ │ ├── self_injecting_actor_tasks.py │ ├── Task_injecting_service/ │ │ ├── task_injecting_service.py │ │ ├── task_injecting_service_tasks.py │ └── tasks_builder.py ├── .gitignore ├── CHANGELOG.md ├── CONTRIBUTING.md ├── docker-compose.yml ├── LICENSE ├── README.md ├── requirements.txt └── run.py</code></pre> <h3>Notable folders</h3> <ul> <li> <code>actors/</code> — <strong>Virtual actor implementations</strong> (your integration layer) <ul> <li><code>ActorBase.py</code> defines the interface for all actors</li> <li><code>DemoActor/</code> provides an example actor</li> </ul> </li> <li> <code>apps/</code> — Flask application code (blueprints, models, services) <ul> <li><code>apps/authentication/</code> login + license gating</li> <li><code>apps/tasks/</code> task models, services, and UI routes</li> <li><code>apps/actors/</code> actor registration/management in the UI</li> <li><code>apps/RabbitMQ/</code> RabbitMQ listener</li> <li><code>apps/Logger/</code> logging utilities</li> <li><code>apps/templates/</code> HTML templates for the UI</li> <li> <code>apps/static/backups</code> folder that save current state of tasks in the DB after execution or when clearing tasks from DB. </li> </ul> </li> <li><code>Examples/</code> — example scripts and task JSON</li> </ul> <hr /> <h2>Requirements</h2> <ul> <li>Python <strong>3.10+</strong></li> <li>A virtualenv or conda environment (recommended)</li> <li> Optional but recommended: <ul> <li><strong>RabbitMQ</strong> server (local, Docker, or remote)</li> <li>A REST API backend for cloud synchronisation (if you use that workflow)</li> </ul> </li> </ul> <hr /> <h2>Quick start (local)</h2> <h3>1) Create and activate an environment</h3> <p>Using conda:</p> <pre><code class="language-bash">conda create -n Twico python=3.10 conda activate Twico</code></pre> <p>Or with venv:</p> <pre><code class="language-bash">python -m venv .venv # Windows .venv\Scripts\activate</code></pre> <h3>2) Install dependencies</h3> <pre><code class="language-bash">pip install -r requirements.txt</code></pre> <h3>3) Configure environment variables</h3> <p> Create a <strong><code>.env</code> in the project root</strong> (do not commit it) based on <code>.env.example</code>. </p> <pre><code class="language-ini">AUTHOR_NAME=Your Name APP_VERSION=0.1.0 SERVER_NAME=127.0.0.1:5000 SECRET_KEY=change-me API_URL= ASSETS_ROOT=/static/assets DB_ENGINE=sqlite DB_NAME=db.sqlite3 RABBIT_MQ_HOST=localhost RABBIT_MQ_PORT=5672 RABBIT_MQ_USER=guest RABBIT_MQ_PASS=guest RABBIT_MQ_VH=/</code></pre> <h3>4) Run</h3> <pre><code class="language-bash">python run.py</code></pre> <p>Open the UI at:</p> <ul> <li><code>http://127.0.0.1:5000</code></li> </ul> <hr /> <h2>Working with tasks</h2> <p>Tasks can be loaded from:</p> <ul> <li>the <strong>cloud REST API</strong> (if configured)</li> <li>a <strong>local JSON file</strong> (see <code>Examples</code> folder)</li> </ul> <p>A task have the following structure:</p> <pre><code class="language-json">{ "name": "A_temp_0", "type": "Move", "main_actor": "DemoActor_A", "description": "", "message": "Moving DemoActor to position", "job": 1, "level": 1, "index": 0, "element_id": [], "actors_data": { "DemoActor_A": { "X": 1, "Y": 2, "Z": 3, "Speed": 0.2 } } }</code></pre> <p>where the actors_data feild can contain any key-value pair.</p> <hr /> <h2>Adding a new actor</h2> <ol> <li>Create a new folder under <code>actors/</code> (e.g. <code>actors/MyRobotActor/</code>)</li> <li> Implement the interface in <code>ActorBase</code> (connect, send, monitor, etc.) <ul> <li>Make sure the file name and the class name are the same as the folder name.</li> </ul> </li> <li>Ensure the actor is importable (package with <code>__init__.py</code>)</li> <li> Register/configure it via the UI (<code>apps/actors/</code>) or your project-specific bootstrap code </li> </ol> <hr /> <h2>RabbitMQ (optional)</h2> <p>If you use RabbitMQ, you can run a local instance with Docker:</p> <pre><code class="language-bash">docker compose up -d rabbitmq</code></pre> <p>See <code>docker-compose.yml</code> in this repo for a ready-to-use configuration.</p> <hr /> <h2>Documentation</h2> <p> This repository is compatible with Sphinx autodoc (docstrings → HTML docs). If you add a <code>docs/</code> folder, you can build API docs automatically. Docs for the folder can be found in <code>apps/static/docs</code> or under the <code>docs</code> in the UI navigation sidebar. </p> <hr /> <h2>Author / Maintainers</h2> <ul> <li>Lior Skoury (original author)</li> </ul> <hr /> </body> </html>

<!doctype html> <html lang="zh-CN"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1" /> <title>Twico控制引擎（Twico Control Engine）</title> </head> <body> <h1>Twico控制引擎（Twico Control Engine）</h1> <p> Twico控制引擎是一款基于Flask的编排引擎（Flask-based orchestration engine），用于<em>数字化实体化（digital materialisation）</em> / <em>数字孪生（digital twin）</em>工作流。该引擎可协调<em>虚拟智能体（virtual actors）</em>（软件化表征实体）与其配对的<em>物理智能体（physical actors）</em>（机器人、工具、传感器等实体），按受控时序执行<em>任务（tasks）</em>，将数据转发至外部服务，并可通过REST API选择性地将本地数据库状态与<em>云数据库（cloud database）</em>进行同步。内置Web用户界面（Web UI），用于管理智能体、任务、作业并监控执行流程。 </p> <blockquote> <p><strong>项目状态</strong>：适用于研究与原型开发的代码库，欢迎贡献代码与反馈问题。</p> </blockquote> <hr /> <h2>解决的问题</h2> <p>在典型的数字化实体化工作流程中，通常包含以下要素：</p> <ul> <li><strong>任务（tasks）</strong>：待执行的作业，例如移动、读取数据、制造等</li> <li><strong>物理智能体（physical actors）</strong>：执行实际作业的实体，涵盖机器人、可编程逻辑控制器（PLC）、各类设备</li> <li><strong>虚拟智能体（virtual actors）</strong>：封装了通信逻辑与前后处理流程的软件实体</li> <li><strong>服务</strong>：用于生成任务或响应任务完成事件的外部系统</li> </ul> <p> Twico控制引擎作为<em>执行核心（execution core）</em>，维护本地的“权威”状态，将任务分发至各智能体，监听事件并更新外部系统。 </p> <hr /> <h2>核心概念</h2> <ul> <li> <strong>智能体（虚拟智能体，Virtual Actor）</strong>：实现了标准接口（位于<code>actors/ActorBase.py</code>）的Python类，用于处理连接建立、任务准备（<code>preSend</code>）、任务分发、完成监控以及响应处理（<code>postSend</code>）等逻辑。 </li> <li> <strong>任务（Task）</strong>：可执行的最小单元，被分配至主智能体（<code>main_actor</code>），并在<code>actors_data</code>字段中附带各智能体的专属参数。 </li> <li> <strong>引擎（Engine）</strong>：承载Web UI与API端点的Flask服务器，用于协调任务执行流程。 </li> <li> <strong>RabbitMQ</strong>：可选的事件总线，用于支持“任务已接收”/“任务已完成”类型的工作流。 </li> </ul> <hr /> <h2>执行模型（高层级）</h2> <p>该引擎持续运行并协调各智能体的执行流程：</p> <ol> <li>确保所有所需的虚拟智能体均已注册并建立连接。</li> <li> 针对每个可用智能体： <ul> <li>从本地数据库（或任务队列）获取下一个待执行任务</li> <li>调用<code>preSend(task)</code>，随后将任务发送至配对的物理智能体</li> <li>等待任务完成或确认信号（取决于通信模式）</li> <li>调用<code>postSend(response)</code>并记录执行结果</li> <li>触发所有关联的外部服务</li> </ul> </li> <li> 并行运行的后台监听器可执行以下操作： <ul> <li>接收新任务（例如通过RabbitMQ）</li> <li>若已配置REST API，则将任务完成更新同步至云数据库</li> </ul> </li> </ol> <hr /> <h2>项目结构</h2> <pre><code>TWICO_Release/ ├── actors/ │ ├── __init__.py │ ├── ActorBase.py │ ├── RMQ_setup.py │ ├── DemoActor/ │ │ ├──__init__.py │ │ └── DemoActor.py │ ├── ServiceConnectedActor/ │ │ ├──__init__.py │ │ └── ServiceConnectedActor.py │ └── TaskInjectingActor/ │ ├──__init__.py │ └── TaskInjectingActor.py ├── apps/ │ ├── actors/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── authentication/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── home/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── routes.py │ │ └── services.py │ ├── Logger/ │ │ ├──__init__.py │ │ ├── logs/ │ │ └── Logger.py │ ├── RabbitMQ/ │ │ ├──__init__.py │ │ └── listener.py │ ├── run/ │ │ ├──__init__.py │ │ ├── routes.py │ │ └── services.py │ ├── static/ │ │ └── assets/ │ ├── tasks/ │ │ ├──__init__.py │ │ ├── forms.py │ │ ├── model.py │ │ ├── routes.py │ │ └── services.py │ ├── templates/ │ │ ├── actors/ │ │ ├── auth/ │ │ ├── dashboard/ │ │ ├── includes/ │ │ ├── layouts/ │ │ ├── run/ │ │ └── tasks/ │ ├── .env.example │ ├── __init__.py │ └── config.py ├── Examples/ │ ├── Basic/ │ │ ├── basic_tasks.py │ ├── Parallel_actors/ │ │ ├── parallel_actors_tasks.py │ ├── Self_injecting_actor/ │ │ ├── self_injecting_actor_tasks.py │ ├── Task_injecting_service/ │ │ ├── task_injecting_service.py │ │ ├── task_injecting_service_tasks.py │ └── tasks_builder.py ├── .gitignore ├── CHANGELOG.md ├── CONTRIBUTING.md ├── docker-compose.yml ├── LICENSE ├── README.md ├── requirements.txt └── run.py</code></pre> <h3>主要目录说明</h3> <ul> <li> <code>actors/</code>：虚拟智能体实现目录（即集成层） <ul> <li><code>ActorBase.py</code>：定义所有智能体的标准接口</li> <li><code>DemoActor/</code>：提供示例智能体实现</li> </ul> </li> <li> <code>apps/</code>：Flask应用代码目录，包含蓝图、数据模型与业务服务 <ul> <li><code>apps/authentication/</code>：登录与权限控制模块</li> <li><code>apps/tasks/</code>：任务模型、业务服务与UI路由模块</li> <li><code>apps/actors/</code>：智能体注册与UI管理模块</li> <li><code>apps/RabbitMQ/</code>：RabbitMQ事件监听器模块</li> <li><code>apps/Logger/</code>：日志工具模块</li> <li><code>apps/templates/</code>：Web UI的HTML模板目录</li> <li> <code>apps/static/backups</code>：用于在任务执行完成或从数据库清除任务时，保存数据库中任务的当前状态的备份目录。 </li> </ul> </li> <li><code>Examples/</code>：示例脚本与任务JSON文件目录</li> </ul> <hr /> <h2>依赖要求</h2> <ul> <li>Python <strong>3.10+</strong></li> <li>推荐使用virtualenv或conda创建并管理虚拟环境</li> <li> 可选但推荐配置： <ul> <li><strong>RabbitMQ</strong>服务器（本地、Docker容器或远程服务器均可）</li> <li>若需使用云同步工作流，则需配套REST API后端</li> </ul> </li> </ul> <hr /> <h2>本地快速启动指南</h2> <h3>1. 创建并激活虚拟环境</h3> <p>使用conda创建：</p> <pre><code class="language-bash">conda create -n Twico python=3.10 conda activate Twico</code></pre> <p>或使用venv创建：</p> <pre><code class="language-bash">python -m venv .venv # Windows .venvScriptsactivate</code></pre> <h3>2. 安装依赖项</h3> <pre><code class="language-bash">pip install -r requirements.txt</code></pre> <h3>3. 配置环境变量</h3> <p> 在项目根目录下创建<code>.env</code>文件（请勿提交至版本控制系统），可参考<code>.env.example</code>模板进行配置。 </p> <pre><code class="language-ini">AUTHOR_NAME=Your Name APP_VERSION=0.1.0 SERVER_NAME=127.0.0.1:5000 SECRET_KEY=change-me API_URL= ASSETS_ROOT=/static/assets DB_ENGINE=sqlite DB_NAME=db.sqlite3 RABBIT_MQ_HOST=localhost RABBIT_MQ_PORT=5672 RABBIT_MQ_USER=guest RABBIT_MQ_PASS=guest RABBIT_MQ_VH=/</code></pre> <h3>4. 启动项目</h3> <pre><code class="language-bash">python run.py</code></pre> <p>访问Web UI的地址为：</p> <ul> <li><code>http://127.0.0.1:5000</code></li> </ul> <hr /> <h2>任务使用说明</h2> <p>任务可通过以下方式加载：</p> <ul> <li>已配置的云REST API</li> <li>本地JSON文件（可参考<code>Examples</code>目录中的示例）</li> </ul> <p>任务的标准结构示例如下：</p> <pre><code class="language-json">{ "name": "A_temp_0", "type": "Move", "main_actor": "DemoActor_A", "description": "", "message": "Moving DemoActor to position", "job": 1, "level": 1, "index": 0, "element_id": [], "actors_data": { "DemoActor_A": { "X": 1, "Y": 2, "Z": 3, "Speed": 0.2 } } }</code></pre> <p>其中<code>actors_data</code>字段可包含任意键值对参数。</p> <hr /> <h2>添加新智能体</h2> <ol> <li>在<code>actors/</code>目录下创建新的智能体目录（例如<code>actors/MyRobotActor/</code>）</li> <li> 实现<code>ActorBase</code>中定义的标准接口（包括连接、任务发送、监控等逻辑） <ul> <li>确保智能体的文件名与类名与目录名保持一致。</li> </ul> </li> <li>确保智能体可被正常导入（需包含<code>__init__.py</code>以作为Python包）</li> <li> 通过Web UI（<code>apps/actors/</code>模块）或项目专属的启动代码注册并配置该智能体 </li> </ol> <hr /> <h2>RabbitMQ（可选）</h2> <p>若需使用RabbitMQ，可通过Docker启动本地实例：</p> <pre><code class="language-bash">docker compose up -d rabbitmq</code></pre> <p>可参考本仓库中的<code>docker-compose.yml</code>文件获取预配置的启动参数。</p> <hr /> <h2>文档说明</h2> <p> 本仓库兼容Sphinx自动文档生成工具（可将代码注释转换为HTML文档）。若创建<code>docs/</code>目录，可自动构建API文档。项目的文档可在<code>apps/static/docs</code>目录中查看，或通过Web UI导航栏的<code>docs</code>菜单访问。 </p> <hr /> <h2>作者与维护者</h2> <ul> <li>利奥尔·斯库里（Lior Skoury，原始作者）</li> </ul> <hr /> </body> </html>

应用场景：