Use email button above to contact.

Twico Control Engine

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

Status: research / prototyping-friendly codebase. Contributions and issues are welcome.

What problem it solves

In a typical digital materialisation process you have:

• Tasks to execute (move, read data, fabricate, etc.)
• Physical actors that do the work (robots, PLCs, devices)
• Virtual actors that encapsulate communication + pre/post-processing
• Services that generate tasks or react to task completion

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

Key concepts

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

Execution model (high level)

The engine runs continuously and coordinates execution across actors:

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

Project structure

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

Notable folders

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

Requirements

• Python 3.10+
• A virtualenv or conda environment (recommended)
• Optional but recommended:
  • RabbitMQ server (local, Docker, or remote)
  • A REST API backend for cloud synchronisation (if you use that workflow)

Quick start (local)

1) Create and activate an environment

Using conda:

conda create -n Twico python=3.10
conda activate Twico

Or with venv:

python -m venv .venv
# Windows
.venv\Scripts\activate

2) Install dependencies

pip install -r requirements.txt

3) Configure environment variables

Create a .env in the project root (do not commit it) based on .env.example.

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=/

4) Run

python run.py

Open the UI at:

• http://127.0.0.1:5000

Working with tasks

Tasks can be loaded from:

• the cloud REST API (if configured)
• a local JSON file (see Examples folder)

A task have the following structure:

{
  "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
    }
  }
}

where the actors_data feild can contain any key-value pair.

Adding a new actor

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

RabbitMQ (optional)

If you use RabbitMQ, you can run a local instance with Docker:

docker compose up -d rabbitmq

See docker-compose.yml in this repo for a ready-to-use configuration.

Documentation

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

Author / Maintainers

• Lior Skoury (original author)