Twico Control Engine

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: Ensure all required virtual actors are registered and connected. 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 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: 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 Create a new folder under actors/ (e.g. actors/MyRobotActor/) 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. Ensure the actor is importable (package with __init__.py) 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)