DOI: 10.18419/darus-5649
title: Twico Control Engine

description: &lt;!doctype html&gt;
&lt;html lang="en"&gt;
&lt;head&gt;
 &lt;meta charset="utf-8" /&gt;
 &lt;meta name="viewport" content="width=device-width, initial-scale=1" /&gt;
 &lt;title&gt;Twico Control Engine&lt;/title&gt;
&lt;/head&gt;
&lt;body&gt;
 &lt;h1&gt;Twico Control Engine&lt;/h1&gt;

 &lt;p&gt;
 Twico Control is a &lt;strong&gt;Flask-based orchestration engine&lt;/strong&gt; for &lt;em&gt;digital materialisation&lt;/em&gt; /
 &lt;em&gt;digital twin&lt;/em&gt; workflows.
 It coordinates &lt;strong&gt;virtual actors&lt;/strong&gt; (software representations) and their paired
 &lt;strong&gt;physical actors&lt;/strong&gt; (robots, tools, sensors), executes &lt;strong&gt;tasks&lt;/strong&gt; in a controlled order,
 forward information to external services, and optionally synchronises state of the local DB with a
 &lt;strong&gt;cloud database via REST API&lt;/strong&gt;.
 A web UI is included for managing actors, tasks, jobs, and monitoring execution.
 &lt;/p&gt;

 &lt;blockquote&gt;
 &lt;p&gt;&lt;strong&gt;Status&lt;/strong&gt;: research / prototyping-friendly codebase. Contributions and issues are welcome.&lt;/p&gt;
 &lt;/blockquote&gt;

 &lt;hr /&gt;

 &lt;h2&gt;What problem it solves&lt;/h2&gt;

 &lt;p&gt;In a typical digital materialisation process you have:&lt;/p&gt;
 &lt;ul&gt;
 &lt;li&gt;&lt;strong&gt;Tasks&lt;/strong&gt; to execute (move, read data, fabricate, etc.)&lt;/li&gt;
 &lt;li&gt;&lt;strong&gt;Physical actors&lt;/strong&gt; that do the work (robots, PLCs, devices)&lt;/li&gt;
 &lt;li&gt;&lt;strong&gt;Virtual actors&lt;/strong&gt; that encapsulate communication + pre/post-processing&lt;/li&gt;
 &lt;li&gt;&lt;strong&gt;Services&lt;/strong&gt; that generate tasks or react to task completion&lt;/li&gt;
 &lt;/ul&gt;

 &lt;p&gt;
 Twico Control acts as the &lt;strong&gt;execution core&lt;/strong&gt;: it keeps a local “authoritative” state,
 dispatches tasks to actors, listens for events, and updates external systems.
 &lt;/p&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Key concepts&lt;/h2&gt;

 &lt;ul&gt;
 &lt;li&gt;
 &lt;strong&gt;Actor (Virtual Actor)&lt;/strong&gt;: a Python class that implements a standard interface
 (&lt;code&gt;actors/ActorBase.py&lt;/code&gt;) and handles: connection, task preparation (&lt;code&gt;preSend&lt;/code&gt;), task dispatch,
 completion monitoring, and response handling (&lt;code&gt;postSend&lt;/code&gt;).
 &lt;/li&gt;
 &lt;li&gt;
 &lt;strong&gt;Task&lt;/strong&gt;: an executable unit that is assigned to a &lt;code&gt;main_actor&lt;/code&gt; with extra per-actor
 parameters under &lt;code&gt;actors_data&lt;/code&gt;.
 &lt;/li&gt;
 &lt;li&gt;
 &lt;strong&gt;Engine&lt;/strong&gt;: a Flask server that runs the UI + API endpoints and coordinates task execution.
 &lt;/li&gt;
 &lt;li&gt;
 &lt;strong&gt;RabbitMQ&lt;/strong&gt;: optional event bus for “task received” / “task complete” style workflows.
 &lt;/li&gt;
 &lt;/ul&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Execution model (high level)&lt;/h2&gt;

 &lt;p&gt;The engine runs continuously and coordinates execution across actors:&lt;/p&gt;

 &lt;ol&gt;
 &lt;li&gt;Ensure all required virtual actors are registered and connected.&lt;/li&gt;
 &lt;li&gt;
 For each available actor:
 &lt;ul&gt;
 &lt;li&gt;fetch the next task from the local DB (or tasks queue)&lt;/li&gt;
 &lt;li&gt;call &lt;code&gt;preSend(task)&lt;/code&gt; then send the task to the paired physical actor&lt;/li&gt;
 &lt;li&gt;wait for completion / acknowledgements (depending on your communication pattern)&lt;/li&gt;
 &lt;li&gt;call &lt;code&gt;postSend(response)&lt;/code&gt; and log results&lt;/li&gt;
 &lt;li&gt;trigger any linked services&lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;li&gt;
 In parallel, background listeners can:
 &lt;ul&gt;
 &lt;li&gt;ingest new tasks (e.g. via RabbitMQ)&lt;/li&gt;
 &lt;li&gt;propagate task completion updates to a cloud DB (if REST API is configurated)&lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;/ol&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Project structure&lt;/h2&gt;

 &lt;pre&gt;&lt;code&gt;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&lt;/code&gt;&lt;/pre&gt;

 &lt;h3&gt;Notable folders&lt;/h3&gt;

 &lt;ul&gt;
 &lt;li&gt;
 &lt;code&gt;actors/&lt;/code&gt; — &lt;strong&gt;Virtual actor implementations&lt;/strong&gt; (your integration layer)
 &lt;ul&gt;
 &lt;li&gt;&lt;code&gt;ActorBase.py&lt;/code&gt; defines the interface for all actors&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;DemoActor/&lt;/code&gt; provides an example actor&lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;li&gt;
 &lt;code&gt;apps/&lt;/code&gt; — Flask application code (blueprints, models, services)
 &lt;ul&gt;
 &lt;li&gt;&lt;code&gt;apps/authentication/&lt;/code&gt; login + license gating&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;apps/tasks/&lt;/code&gt; task models, services, and UI routes&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;apps/actors/&lt;/code&gt; actor registration/management in the UI&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;apps/RabbitMQ/&lt;/code&gt; RabbitMQ listener&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;apps/Logger/&lt;/code&gt; logging utilities&lt;/li&gt;
 &lt;li&gt;&lt;code&gt;apps/templates/&lt;/code&gt; HTML templates for the UI&lt;/li&gt;
 &lt;li&gt;
 &lt;code&gt;apps/static/backups&lt;/code&gt; folder that save current state of tasks in the DB after execution or when
 clearing tasks from DB.
 &lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;li&gt;&lt;code&gt;Examples/&lt;/code&gt; — example scripts and task JSON&lt;/li&gt;
 &lt;/ul&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Requirements&lt;/h2&gt;

 &lt;ul&gt;
 &lt;li&gt;Python &lt;strong&gt;3.10+&lt;/strong&gt;&lt;/li&gt;
 &lt;li&gt;A virtualenv or conda environment (recommended)&lt;/li&gt;
 &lt;li&gt;
 Optional but recommended:
 &lt;ul&gt;
 &lt;li&gt;&lt;strong&gt;RabbitMQ&lt;/strong&gt; server (local, Docker, or remote)&lt;/li&gt;
 &lt;li&gt;A REST API backend for cloud synchronisation (if you use that workflow)&lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;/ul&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Quick start (local)&lt;/h2&gt;

 &lt;h3&gt;1) Create and activate an environment&lt;/h3&gt;

 &lt;p&gt;Using conda:&lt;/p&gt;
 &lt;pre&gt;&lt;code class="language-bash"&gt;conda create -n Twico python=3.10
conda activate Twico&lt;/code&gt;&lt;/pre&gt;

 &lt;p&gt;Or with venv:&lt;/p&gt;
 &lt;pre&gt;&lt;code class="language-bash"&gt;python -m venv .venv
# Windows
.venv\Scripts\activate&lt;/code&gt;&lt;/pre&gt;

 &lt;h3&gt;2) Install dependencies&lt;/h3&gt;
 &lt;pre&gt;&lt;code class="language-bash"&gt;pip install -r requirements.txt&lt;/code&gt;&lt;/pre&gt;

 &lt;h3&gt;3) Configure environment variables&lt;/h3&gt;

 &lt;p&gt;
 Create a &lt;strong&gt;&lt;code&gt;.env&lt;/code&gt; in the project root&lt;/strong&gt; (do not commit it) based on
 &lt;code&gt;.env.example&lt;/code&gt;.
 &lt;/p&gt;

 &lt;pre&gt;&lt;code class="language-ini"&gt;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=/&lt;/code&gt;&lt;/pre&gt;

 &lt;h3&gt;4) Run&lt;/h3&gt;
 &lt;pre&gt;&lt;code class="language-bash"&gt;python run.py&lt;/code&gt;&lt;/pre&gt;

 &lt;p&gt;Open the UI at:&lt;/p&gt;
 &lt;ul&gt;
 &lt;li&gt;&lt;code&gt;http://127.0.0.1:5000&lt;/code&gt;&lt;/li&gt;
 &lt;/ul&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Working with tasks&lt;/h2&gt;

 &lt;p&gt;Tasks can be loaded from:&lt;/p&gt;
 &lt;ul&gt;
 &lt;li&gt;the &lt;strong&gt;cloud REST API&lt;/strong&gt; (if configured)&lt;/li&gt;
 &lt;li&gt;a &lt;strong&gt;local JSON file&lt;/strong&gt; (see &lt;code&gt;Examples&lt;/code&gt; folder)&lt;/li&gt;
 &lt;/ul&gt;

 &lt;p&gt;A task have the following structure:&lt;/p&gt;

 &lt;pre&gt;&lt;code class="language-json"&gt;{
 "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
 }
 }
}&lt;/code&gt;&lt;/pre&gt;

 &lt;p&gt;where the actors_data feild can contain any key-value pair.&lt;/p&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Adding a new actor&lt;/h2&gt;

 &lt;ol&gt;
 &lt;li&gt;Create a new folder under &lt;code&gt;actors/&lt;/code&gt; (e.g. &lt;code&gt;actors/MyRobotActor/&lt;/code&gt;)&lt;/li&gt;
 &lt;li&gt;
 Implement the interface in &lt;code&gt;ActorBase&lt;/code&gt; (connect, send, monitor, etc.)
 &lt;ul&gt;
 &lt;li&gt;Make sure the file name and the class name are the same as the folder name.&lt;/li&gt;
 &lt;/ul&gt;
 &lt;/li&gt;
 &lt;li&gt;Ensure the actor is importable (package with &lt;code&gt;__init__.py&lt;/code&gt;)&lt;/li&gt;
 &lt;li&gt;
 Register/configure it via the UI (&lt;code&gt;apps/actors/&lt;/code&gt;) or your project-specific bootstrap code
 &lt;/li&gt;
 &lt;/ol&gt;

 &lt;hr /&gt;

 &lt;h2&gt;RabbitMQ (optional)&lt;/h2&gt;

 &lt;p&gt;If you use RabbitMQ, you can run a local instance with Docker:&lt;/p&gt;
 &lt;pre&gt;&lt;code class="language-bash"&gt;docker compose up -d rabbitmq&lt;/code&gt;&lt;/pre&gt;

 &lt;p&gt;See &lt;code&gt;docker-compose.yml&lt;/code&gt; in this repo for a ready-to-use configuration.&lt;/p&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Documentation&lt;/h2&gt;

 &lt;p&gt;
 This repository is compatible with Sphinx autodoc (docstrings → HTML docs).
 If you add a &lt;code&gt;docs/&lt;/code&gt; folder, you can build API docs automatically.
 Docs for the folder can be found in &lt;code&gt;apps/static/docs&lt;/code&gt; or under the &lt;code&gt;docs&lt;/code&gt; in the UI navigation sidebar.
 &lt;/p&gt;

 &lt;hr /&gt;

 &lt;h2&gt;Author / Maintainers&lt;/h2&gt;

 &lt;ul&gt;
 &lt;li&gt;Lior Skoury (original author)&lt;/li&gt;
 &lt;/ul&gt;

 &lt;hr /&gt;

&lt;/body&gt;
&lt;/html&gt;