JSON

.devcontainer/Dockerfile

@@ -1,14 +1,5 @@
 FROM debian:stable-slim
 
-ARG USERNAME=vscode
-ARG USER_UID=1000
-ARG USER_GID=1000
-
-RUN groupadd --gid $USER_GID $USERNAME \
-    && useradd --uid $USER_UID --gid $USER_GID -m $USERNAME \
-    && apt-get update && apt-get install -y sudo \
-    && echo "$USERNAME ALL=(ALL) NOPASSWD:ALL" >> /etc/sudoers
-
 RUN RUN set -eux; \
   apt-get update; \
   apt-get install -y \
@@ -27,6 +18,7 @@ RUN set -eux; \
     prometheus-cpp-dev \
     nlohmann-json3-dev
 
+
 RUN set -eux; \
   apt-get update; \
   apt-get install -y \
@@ -42,14 +34,13 @@ RUN set -eux; \
   cmake --build . --target install; \
   ldconfig;
 
-RUN set -eux; \
+  RUN set -eux; \
   apt-get update; \
   apt-get install -y \
     librabbitmq4 \
     librabbitmq-dev;\
   apt-get clean
 
-USER $USERNAME
-WORKDIR /workspace
+WORKDIR /root
 
 CMD ["sleep infinity"]

.devcontainer/compose.yml

@@ -1,54 +0,0 @@
-services:
-  dev:
-    build:
-      context: .
-      dockerfile: Dockerfile
-    volumes:
-      - ..:/workspace:cached
-    command: sleep infinity
-    networks:
-      - dev_net
-
-  nodered:
-    image: nodered/node-red:4.1
-    container_name: nodered
-    ports:
-      - "1880:1880"
-    networks:
-      - dev_net
-    environment:
-      TZ: Europe/Paris
-    volumes:
-    - ./nodered-data:/data
-    - ./nodered-flows/flows.json:/data/flows.json
-
-  rabbitmq:
-    image: rabbitmq:4.1.4-management
-    container_name: rabbitmq
-    environment:
-      RABBITMQ_DEFAULT_USER: "admin"
-      RABBITMQ_DEFAULT_PASS: "geii2025"
-
-      # Activation MQTT sur le port 1883
-      RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: >
-        -rabbitmq_mqtt tcp_listeners [1883]
-
-    ports:
-      - "5672:5672"  # AMQP
-      - "1883:1883"  # MQTT
-      - "15672:15672" # RabbitMQ Manager
-    networks:
-      - dev_net
-    volumes:
-      - rabbitmq:/var/lib/rabbitmq
-
-    # Activation des plugins + démarrage serveur
-    command: >
-      sh -c "rabbitmq-plugins enable --offline rabbitmq_mqtt rabbitmq_management &&
-             rabbitmq-server"
-
-networks:
-  dev_net:
-
-volumes:
-  rabbitmq:

.devcontainer/devcontainer.json

@@ -1,10 +1,13 @@
 {
   "name": "Developpement C",
-   "dockerComposeFile": [
-    "compose.yml"
+  "build": {
+    "dockerfile": "Dockerfile"
+  },
+  "runArgs": [
+    "--label", "prometheus=true",
+    "--network=tp_net",
+    "--name=pompes"
   ],
-  "service": "dev",
-  "workspaceFolder": "/workspace",
   "customizations": {
     "vscode": {
       "settings": {

AutomForArduino.cpp

@@ -52,9 +52,9 @@ typedef struct PinIO
   unsigned long time;
   double duration;
   unsigned int nb; // compteur d'activation
-  int memory;
-  unsigned char raising;
-  unsigned char falling;
+  int memory;             // valeur précédente
+  unsigned char raising;  // front montant
+  unsigned char falling;  // front descendant
 } PinIO;
 
 PinIO _digital[256];
@@ -75,8 +75,6 @@ void pinMode(unsigned char p, unsigned char mode)
 }
 
 
-
-
 /* KEYBOARD */
 
 typedef struct
@@ -351,6 +349,8 @@ void digitalWrite(unsigned int p, int value)
   // En panne !
   if (!(_digital[p].mode & 0x01))
   {
+	_digital[p].ivalue = 0;
+    _digital[p].dvalue = 0.0;
     return;
   }
 
@@ -409,3 +409,19 @@ void analogWrite(unsigned int p, double value)
 
   _digital[p].dvalue = value;
 }
+
+/* ********************************************************
+ *  Console                                               *
+ *                                                        *
+ ******************************************************** */
+void ConsoleInit()
+{
+  setlocale(LC_ALL, "");  // Activer le support des caractères Unicode
+  setlocale(LC_NUMERIC, "C");
+  initscr();              // Initialise ncurses
+  raw();                  // Mode brut, sans besoin de validation par Entrée
+  keypad(stdscr, TRUE);   // Active les touches spéciales comme ESC
+  nodelay(stdscr, TRUE);  // Mode non-bloquant pour getch()
+  noecho();               // Ne pas afficher les touches appuyées
+  curs_set(0);            // Masquer le curseur
+}

main-mqtt.cpp

@@ -8,7 +8,7 @@
 #define TOPIC       "test/topic"
 #define PAYLOAD     "Hello MQTT"
 #define QOS         1
-#define TIMEOUT     10000L
+#define TIMEOUT_MQTT     10000L
 
 int main() {
     MQTTClient client;
@@ -33,7 +33,7 @@ int main() {
     MQTTClient_deliveryToken token;
     MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
 
-    MQTTClient_waitForCompletion(client, token, TIMEOUT);
+    MQTTClient_waitForCompletion(client, token, TIMEOUT_MQTT);
     printf("Message publié !\n");
 
     MQTTClient_disconnect(client, 10000);

main-mqtt2.cpp

@@ -0,0 +1,163 @@
+#include <iostream>
+#include <string>
+#include <thread>
+#include <atomic>
+#include <queue>
+#include <mutex>
+#include <condition_variable>
+#include <csignal>
+#include <chrono>
+#include <cstring>
+#include "mqtt/async_client.h"
+
+using namespace std::chrono_literals;
+
+/* Configuration MQTT */
+const std::string ADDRESS = "tcp://rabbitmq:1883";
+const std::string CLIENTID = "CppClientTP";
+const std::string TOPIC = "geii/ordre/#";
+const int QOS = 1;
+const int CYCLE_MS = 100;
+
+/* Queue thread-safe */
+std::queue<std::string> orders_queue;
+std::mutex queue_mtx;
+
+std::string pop_all_and_get_last() {
+    std::lock_guard<std::mutex> lock(queue_mtx);
+    if (orders_queue.empty()) return "";
+    std::string last;
+    while (!orders_queue.empty()) {
+        last = orders_queue.front();
+        orders_queue.pop();
+    }
+    return last;
+}
+
+void push_order(const std::string &msg) {
+    std::lock_guard<std::mutex> lock(queue_mtx);
+    orders_queue.push(msg);
+}
+
+/* Etats machine */
+enum class MachineState { STOPPED, RUNNING, ESTOP };
+std::atomic<MachineState> machine_state(MachineState::STOPPED);
+std::atomic<bool> estop_flag(false);
+std::atomic<bool> running(true);
+
+/* Fonctions d'application */
+void apply_start() {
+    if(machine_state!=MachineState::RUNNING){
+        std::cout << "[MACHINE] -> START\n";
+        machine_state = MachineState::RUNNING;
+    }
+}
+
+void apply_stop() {
+    if(machine_state!=MachineState::STOPPED){
+        std::cout << "[MACHINE] -> STOP\n";
+        machine_state = MachineState::STOPPED;
+    }
+}
+
+void apply_estop() {
+    if(machine_state!=MachineState::ESTOP){
+        std::cout << "[MACHINE] -> E-STOP\n";
+        machine_state = MachineState::ESTOP;
+    }
+}
+
+/* Thread machine */
+void machine_thread_fn() {
+    while(running) {
+        if(estop_flag) {
+            apply_estop();
+            std::this_thread::sleep_for(std::chrono::milliseconds(CYCLE_MS));
+            continue;
+        }
+        std::string last = pop_all_and_get_last();
+        if(!last.empty()) {
+            if(last=="START") apply_start();
+            else if(last=="STOP") apply_stop();
+            else std::cout << "[MACHINE] Commande inconnue: '" << last << "'\n";
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(CYCLE_MS));
+    }
+    apply_stop();
+}
+
+/* Callback MQTT */
+class callback : public virtual mqtt::callback {
+public:
+    void message_arrived(mqtt::const_message_ptr msg) override {
+        std::string payload = msg->to_string();
+        if(payload == "E_STOP") {
+            estop_flag = true;
+            std::cout << "[MQTT] E-STOP reçu\n";
+        } else {
+            push_order(payload);
+            std::cout << "[MQTT] Reçu: '" << payload << "'\n";
+        }
+    }
+};
+
+/* SIGINT handler */
+void sigint_handler(int) {
+    std::cout << "[MAIN] SIGINT reçu\n";
+    running = false;
+}
+
+int main() {
+    std::signal(SIGINT, sigint_handler);
+
+    /* MQTT async client */
+    mqtt::async_client client(ADDRESS, CLIENTID);
+    callback cb;
+    client.set_callback(cb);
+
+    mqtt::connect_options connOpts;
+    connOpts.set_clean_session(true);
+    connOpts.set_user_name("admin");
+    connOpts.set_password("ChangeMe");
+    try {
+        client.connect(connOpts)->wait();
+        client.start_consuming();
+        client.subscribe(TOPIC, QOS)->wait();
+    } catch (const mqtt::exception &exc) {
+        std::cerr << "Erreur MQTT: " << exc.what() << "\n";
+        return 1;
+    }
+
+    /* Threads */
+    std::thread th_machine(machine_thread_fn);
+
+    /* Boucle principale pour la réception */
+    while(running) {
+
+        std::string payload = R"({
+                "order": "STATUS",
+                "speed": 120,
+                "temperature": 36.1
+            })";
+
+        auto msg = mqtt::make_message("geii/telemetry", payload);
+        msg->set_qos(1);
+        client.publish(msg);
+
+        std::this_thread::sleep_for(100ms);
+    }
+
+    /* Arrêt */
+    try {
+        client.unsubscribe(TOPIC)->wait();
+        client.stop_consuming();
+        client.disconnect()->wait();
+    } catch(const mqtt::exception &exc){
+        std::cerr << "Erreur déconnexion MQTT: " << exc.what() << "\n";
+    }
+
+    th_machine.join();
+
+    std::cout << "[MAIN] Terminé\n";
+    return 0;
+}

main.cpp

@@ -1,7 +1,5 @@
-#include <iostream>
-#include <iomanip>
-
 #include <unistd.h>
+#include <ncurses.h>
 #include <math.h>
 #include <locale.h>
 #include <array>
@@ -16,6 +14,7 @@
 
 #include <curl/curl.h>
 #include <string>
+#include <iostream>
 
 #include <thread>
 #include <atomic>
@@ -29,6 +28,8 @@
 #include "mqtt/async_client.h"
 
 #include <nlohmann/json.hpp>
+
+using namespace std::chrono_literals;
 using json = nlohmann::json;
 
 // Constantes de fonctionnement
@@ -42,6 +43,8 @@ const std::string TOPIC = "geii/ordre/#";
 const int QOS = 1;
 const int CYCLE_MS = 100;
 
+WINDOW *window;
+
 int etape = 0; // Étape du grafcet : début Automatique
 int bp_mode, bp_mode_fm;
 unsigned short pompe1, pompe2, pompe3, pompe4; // bouton des pompes 0 (arrêt) / 1 (marche)
@@ -50,34 +53,175 @@ unsigned short sensor_max, sensor_high, sensor_low, sensor_min;
 
 float TankInitalValue = 7;
 
-// Réception des messages MQTT
+TemporisationRetardMontee tempo1(1500);
+TemporisationRetardMontee tempo2(3000);
+TemporisationRetardMontee tempo3(4000);
+TemporisationRetardMontee tempo4(6000);
+
+// Prometheus
 // ************************************************************
+using namespace prometheus;
+
+std::shared_ptr<Registry> registry;
+Gauge *debit_entree = nullptr;
+Gauge *debit_sortie = nullptr;
+
+Gauge *debit_p1 = nullptr;
+Gauge *debit_p2 = nullptr;
+Gauge *debit_p3 = nullptr;
+Gauge *debit_p4 = nullptr;
+Gauge *tank_gauge = nullptr;
+
+Counter *volume_p1 = nullptr;
+Counter *volume_p2 = nullptr;
+Counter *volume_p3 = nullptr;
+Counter *volume_p4 = nullptr;
+
+Histogram::BucketBoundaries buckets = {
+  2, 5, 6, 7, 8, 9, 9.5
+};
+
+Histogram *tank_histogram = nullptr;
+// ************************************************************
+
+
+/* Queue thread-safe */
+std::queue<std::string> orders_queue;
+std::mutex queue_mtx;
+
+std::string pop_all_and_get_last() {
+    std::lock_guard<std::mutex> lock(queue_mtx);
+    if (orders_queue.empty()) return "";
+    std::string last;
+    while (!orders_queue.empty()) {
+      last = orders_queue.front();
+      orders_queue.pop();
+    }
+    return last;
+}
+
+void push_order(const std::string &msg) {
+    std::lock_guard<std::mutex> lock(queue_mtx);
+    orders_queue.push(msg);
+}
+
+/* Etats machine */
+enum class MachineState { STOPPED, RUNNING, ESTOP };
+std::atomic<MachineState> machine_state(MachineState::STOPPED);
+std::atomic<bool> estop_flag(false);
+std::atomic<bool> running(true);
+
+/* Fonctions d'application */
+void apply_start() {
+    if(machine_state!=MachineState::RUNNING){
+        std::cout << "[MACHINE] -> START\n";
+        machine_state = MachineState::RUNNING;
+    }
+}
+
+void apply_stop() {
+    if(machine_state!=MachineState::STOPPED){
+        std::cout << "[MACHINE] -> STOP\n";
+        machine_state = MachineState::STOPPED;
+    }
+}
+
+void apply_estop() {
+    if(machine_state!=MachineState::ESTOP){
+        std::cout << "[MACHINE] -> E-STOP\n";
+        machine_state = MachineState::ESTOP;
+    }
+}
+
+/* Thread machine */
+void machine_thread_fn() {
+  while(running) {
+      if(estop_flag) {
+        apply_estop();
+        std::this_thread::sleep_for(std::chrono::milliseconds(CYCLE_MS));
+        continue;
+      }
+      pompe1 = 1;
+      std::string last = pop_all_and_get_last();
+      if(!last.empty()) {
+        if(last=="START") apply_start();
+        if(last=="P1") {
+           pompe1 = 1;
+        } else if(last=="P2") {
+           pompe2 = 1;
+        } else if(last=="P3") {
+           pompe3 = 1;
+        } else if(last=="P4") {
+           pompe4 = 1;
+        }
+        else if(last=="STOP") apply_stop();
+        else std::cout << "[MACHINE] Commande inconnue: '" << last << "'\n";
+      }
+      std::this_thread::sleep_for(std::chrono::milliseconds(CYCLE_MS));
+  }
+  apply_stop();
+}
+
+/* Callback MQTT */
 class callback : public virtual mqtt::callback {
 public:
-  void message_arrived(mqtt::const_message_ptr msg) override {
-    std::string payload = msg->to_string();
-
-    if (payload == "p1") {
-      pompe1 = !pompe1;
-    } else if (payload == "p2") {
-      pompe2 = !pompe2;
-    } else if (payload == "p3") {
-      pompe3 = !pompe3;
-    } else if (payload == "p4") {
-      pompe4 = !pompe4;
+    void message_arrived(mqtt::const_message_ptr msg) override {
+        std::string payload = msg->to_string();
+        if(payload == "E_STOP") {
+            estop_flag = true;
+            std::cout << "[MQTT] E-STOP reçu\n";
+        } else if(payload == "P1")  {
+          pompe2 = 1;
+        } else  {
+            push_order(payload);
+            std::cout << "[MQTT] Reçu: '" << payload << "'\n";
+        }
     }
-  }
 };
-// ************************************************************
+
+/* SIGINT handler */
+void sigint_handler(int) {
+    std::cout << "[MAIN] SIGINT reçu\n";
+    running = false;
+}
+
+void send_to_influx(double value) {
+    CURL* curl = curl_easy_init();
+    if (!curl) {
+        std::cerr << "Erreur CURL\n";
+        return;
+    }
+
+    // Line protocol
+    std::string line = "machine_cycle,machine=convoyeur1 value=" + std::to_string(value);
+
+    // Endpoint InfluxDB 2.x
+    std::string url =
+        "http://influxdb:8086/api/v2/write?org=geii&bucket=mesures&precision=s";
+
+    struct curl_slist* headers = nullptr;
+    headers = curl_slist_append(headers, "Authorization: Token MON_TOKEN");
+    headers = curl_slist_append(headers, "Content-Type: text/plain");
+
+    curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+    curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
+    curl_easy_setopt(curl, CURLOPT_POSTFIELDS, line.c_str());
+
+    CURLcode res = curl_easy_perform(curl);
+    if (res != CURLE_OK) {
+        std::cerr << "Erreur CURL: " << curl_easy_strerror(res) << "\n";
+    }
+
+    curl_slist_free_all(headers);
+    curl_easy_cleanup(curl);
+}
+
 
 int main()
 {
-  /* Initialisation */
-  InitPrometheus();
-
-  ProcessInitIO();
-  ProcessInitValues();
+  std::signal(SIGINT, sigint_handler);
 
+  /* MQTT async client */
   mqtt::async_client client(ADDRESS, CLIENTID);
   callback cb;
   client.set_callback(cb);
@@ -85,7 +229,7 @@ int main()
   mqtt::connect_options connOpts;
   connOpts.set_clean_session(true);
   connOpts.set_user_name("admin");
-  connOpts.set_password("geii2025");
+  connOpts.set_password("ChangeMe");
   try {
     client.connect(connOpts)->wait();
     client.start_consuming();
@@ -95,19 +239,37 @@ int main()
     return 1;
   }
 
+  /* Initialisation */
+  ConsoleInit();
+  AffichageWindow();
+
+  InitPrometheus();
+  ProcessInitKeyboard();
+  ProcessInitIO();
+  ProcessInitValues();
+
   while (1)
   {
+    int ch = getch(); // Lit l'entrée du clavier sans bloquer
+
+    // **** Break loop if escape key (27) is pressed
+    if (ch == 27 || _digital[OUT_END].ivalue) {
+      break;
+    }
+
+    // **** Beep
+    if (_digital[OUT_BEEP].ivalue)
+    {
+      beep();
+      _digital[OUT_BEEP].ivalue = false;
+    }
+
     Process();
 
-    sensor_min = digitalRead(IN_SENSOR_MIN);
-    sensor_low = digitalRead(IN_SENSOR_LOW);
-    sensor_high = digitalRead(IN_SENSOR_HIGH);
-    sensor_max = digitalRead(IN_SENSOR_MAX);
-
-    digitalWrite(OUT_PUMP_1, (pompe1 == 1));
-    digitalWrite(OUT_PUMP_2, (pompe2 == 1));
-    digitalWrite(OUT_PUMP_3, (pompe3 == 1));
-    digitalWrite(OUT_PUMP_4, (pompe4 == 1));
+    LireClavier(ch);
+    LireEntree();
+    EvolutionGrafcet();
+    Actions();
 
     ProcessPrometheus();
     ProcessMQTT(&client);
@@ -116,23 +278,269 @@ int main()
     usleep(100000);
   }
 
+  endwin();             // Termine ncurses et rétablit le terminal
+
+  /* Arrêt */
   try {
     client.unsubscribe(TOPIC)->wait();
     client.stop_consuming();
     client.disconnect()->wait();
   } catch(const mqtt::exception &exc){
-    std::cerr << "Erreur déconnexion MQTT: " << exc.what() << std::endl;
+    std::cerr << "Erreur déconnexion MQTT: " << exc.what() << "\n";
   }
 
-  std::cout << "Fin du programme" << std::endl;
+  //th_machine.join();
+
+  std::cout << "[MAIN] Terminé\n";
+
   return 0;
 }
 
+/**
+ * Programme
+ */
+void LireEntree()
+{
+  int input;
+  input = digitalRead(IN_KEYBOARD_A);
+  bp_mode_fm = (input > bp_mode);
+  bp_mode = input;
+
+  sensor_min = digitalRead(IN_SENSOR_MIN);
+  sensor_low = digitalRead(IN_SENSOR_LOW);
+  sensor_high = digitalRead(IN_SENSOR_HIGH);
+  sensor_max = digitalRead(IN_SENSOR_MAX);
+}
+
+void EvolutionGrafcet()
+{
+  int etape_futur = etape;
+
+  if (etape < 10 && bp_mode_fm)
+  {
+    etape_futur = 10;
+    pompe1 = pompe2 = pompe3 = pompe4 = 0;
+  }
+
+  if (etape <= 2 && _digital[IN_KEYBOARD_1].raising)
+  {
+    pompe1 = !pompe1;
+  }
+
+  if (etape <= 2 && _digital[IN_KEYBOARD_2].raising)
+  {
+    pompe2 = !pompe2;
+  }
+
+  if (etape <= 2 && _digital[IN_KEYBOARD_3].raising)
+  {
+    pompe3 = !pompe3;
+  }
+
+  if (etape <= 2 && _digital[IN_KEYBOARD_4].raising)
+  {
+    pompe4 = !pompe4;
+  }
+
+  if (etape == 0 && !sensor_min)
+  {
+    etape_futur = 1;
+  }
+
+  if (etape == 1)
+  {
+    etape_futur = 2;
+  }
+
+  if (etape == 2 && sensor_min)
+  {
+    etape_futur = 0;
+  }
+
+  if (etape >= 10 && bp_mode_fm)
+  {
+    etape_futur = 0;
+    pompe1 = pompe2 = pompe3 = pompe4 = 0;
+  }
+
+  if (sensor_max)
+  {
+    pompe1 = pompe2 = pompe3 = pompe4 = 0;
+  }
+
+  /* Automatique */
+  if (etape == 10 && !sensor_low && !sensor_high)
+  {
+    etape_futur = 11;
+  }
+
+  if (etape == 11 && sensor_high)
+  {
+    etape_futur = 10;
+  }
+
+  if (etape == 11 && tempo1.getSortie())
+  {
+    etape_futur = 12;
+  }
+
+  if (etape == 12 && sensor_high)
+  {
+    etape_futur = 13;
+  }
+
+  if (etape == 12 && tempo2.getSortie())
+  {
+    etape_futur = 14; // Allumer le moteur 2
+  }
+
+  if (etape == 13 && tempo1.getSortie())
+  {
+    etape_futur = 10;
+  }
+
+  if (etape == 13 && !sensor_low && !sensor_high)
+  {
+    etape_futur = 12;
+  }
+
+  if (etape == 14 && sensor_high)
+  {
+    etape_futur = 15;
+  }
+
+  if (etape == 14 && tempo3.getSortie())
+  {
+    etape_futur = 16;
+  }
+
+  if (etape == 15 && tempo1.getSortie())
+  {
+    etape_futur = 13;
+  }
+
+  if (etape == 15 && !sensor_low && !sensor_high)
+  {
+    etape_futur = 14;
+  }
+
+  if (etape == 16 && sensor_high)
+  {
+    etape_futur = 17;
+  }
+
+  if (etape == 16 && tempo4.getSortie())
+  {
+    etape_futur = 18;
+  }
+
+  if (etape == 17 && tempo1.getSortie())
+  {
+    etape_futur = 15;
+  }
+
+  if (etape == 17 && !sensor_low && !sensor_high)
+  {
+    etape_futur = 16;
+  }
+
+  if (etape == 18 && sensor_high)
+  {
+    etape_futur = 19;
+  }
+
+  if (etape == 19 && tempo1.getSortie())
+  {
+    etape_futur = 17;
+  }
+
+  if (etape == 19 && !sensor_low && !sensor_high)
+  {
+    etape_futur = 18;
+  }
+
+  /* Fin de mode automatique */
+
+  if (etape != etape_futur)
+  {
+    etape = etape_futur;
+  }
+}
+
+void Actions()
+{
+  digitalWrite(OUT_DISPLAY_GRAFCET, etape);
+  digitalWrite(OUT_DISPLAY_MODE, etape >= 10);
+  digitalWrite(OUT_PUMP_1, !sensor_max && (pompe1 == 1 || etape >= 12));
+  digitalWrite(OUT_PUMP_2, !sensor_max && (pompe2 == 1 || etape >= 14));
+  digitalWrite(OUT_PUMP_3, !sensor_max && (pompe3 == 1 || etape >= 16));
+  digitalWrite(OUT_PUMP_4, !sensor_max && (pompe4 == 1 || etape >= 18));
+  // digitalWrite(OUT_BEEP, etape == 1);
+
+  if (etape >= 11)
+  {
+    tempo1.activation();
+    tempo2.activation();
+    tempo3.activation();
+    tempo4.activation();
+  }
+}
+
 /**
  * Process
  */
+void ProcessInitKeyboard()
+{
+  _keyboard[0].vKey = '1';
+  _keyboard[0].input = IN_KEYBOARD_1;
+
+  _keyboard[1].vKey = '2';
+  _keyboard[1].input = IN_KEYBOARD_2;
+
+  _keyboard[2].vKey = '3';
+  _keyboard[2].input = IN_KEYBOARD_3;
+
+  _keyboard[3].vKey = '4';
+  _keyboard[3].input = IN_KEYBOARD_4;
+
+  _keyboard[4].vKey = 'a';
+  _keyboard[4].input = IN_KEYBOARD_A;
+
+  _keyboard[5].vKey = 'x';
+  _keyboard[5].input = IN_KEYBOARD_X;
+
+  _keyboard[6].vKey = '6';
+  _keyboard[6].input = IN_KEYBOARD_7;
+
+  _keyboard[7].vKey = '7';
+  _keyboard[7].input = IN_KEYBOARD_8;
+
+  _keyboard[8].vKey = '8';
+  _keyboard[8].input = IN_KEYBOARD_9;
+
+  _keyboard[9].vKey = '9';
+  _keyboard[9].input = IN_KEYBOARD_0;
+
+  for (int i = 0; i < NB_KEYBOARD; i++)
+  {
+    _digital[_keyboard[i].input].mode = OP_DIGITAL_READ;
+  }
+}
+
 void ProcessInitIO()
 {
+  pinMode(IN_KEYBOARD_1, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_2, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_3, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_4, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_A, IO_INPUT | DIGITAL);
+
+  pinMode(IN_KEYBOARD_7, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_8, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_9, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_0, IO_INPUT | DIGITAL);
+  pinMode(IN_KEYBOARD_X, IO_INPUT | DIGITAL);
+
   pinMode(IN_SENSOR_MIN, IO_INPUT | DIGITAL);
   pinMode(IN_SENSOR_LOW, IO_INPUT | DIGITAL);
   pinMode(IN_SENSOR_HIGH, IO_INPUT | DIGITAL);
@@ -248,10 +656,11 @@ double ProcessMoteur(int i)
 
 void ProcessException()
 {
-  if (t_elapsed > 30) {
+  if (t_elapsed > 60) {
     _digital[OUT_PUMP_1].mode = 0;
+    digitalWrite(OUT_PUMP_1, 0);
   } else if (t_elapsed > 15) {
-    _digital[IN_SENSOR_LOW].mode = 0;
+    //_digital[IN_SENSOR_LOW].mode = 0;
   }
 }
 
@@ -265,8 +674,8 @@ void Process()
   // ***** FLOW OUT
   if (_digital[IN_TANK_LEVEL].dvalue > 1.0)
   {
-    //_digital[IN_FLOW_OUT].dvalue = SimulConsoSinusoidale(t);
-    _digital[IN_FLOW_OUT].dvalue = SimulConsoBrown(_digital[IN_FLOW_OUT].dvalue);
+    _digital[IN_FLOW_OUT].dvalue = SimulConsoSinusoidale(t);
+    //_digital[IN_FLOW_OUT].dvalue = SimulConsoBrown(_digital[IN_FLOW_OUT].dvalue);
   }
   else
   {
@@ -300,6 +709,24 @@ void Process()
     _digital[IN_TANK_MIN].dvalue = _digital[IN_TANK_LEVEL].dvalue;
   }
 
+  // **** KEYBOARD
+  if (_digital[IN_KEYBOARD_X].raising)
+  {
+    _digital[IN_SENSOR_LOW].mode = _digital[IN_SENSOR_LOW].mode ^ 0x01;
+  }
+
+  for (int i = IN_KEYBOARD_7; i <= IN_KEYBOARD_0; i++)
+  {
+    if (_digital[i].raising)
+    {
+      unsigned char p = i + (OUT_PUMP_1 - IN_KEYBOARD_7);
+      _digital[p].mode ^= 0x01;
+      if (!(_digital[p].mode & 0x01)) {
+        digitalWrite(p, 0);
+      }
+    }
+  }
+
   // **** SENSOR
   int test;
 
@@ -343,6 +770,8 @@ void Process()
     _digital[IN_SENSOR_HIGH].ivalue = test;
   }
 
+  Affichage();
+
   t_backup = t;
 }
 
@@ -359,6 +788,8 @@ double SimulConsoBrown(double valeur_precedente)
   float mu = 0.01 * -((((int)t_elapsed / 30) % 2) * 2 - 1);  // Taux de croissance (1%)
   float sigma = 0.05;   // Volatilité (5%)
 
+   mvprintw(8, 40, "(µ %.1f %% ; σ %.1f %%) ", mu * 100, sigma * 100);
+
   // Nombre aléatoire compris dans [-1 +1]
   float rand_std_normal = ((double)rand() / RAND_MAX) * 2.0 - 1.0;
 
@@ -373,19 +804,234 @@ double SimulConsoBrown(double valeur_precedente)
  * Affichage dans la console
  */
 
+/**
+ * Initialisation, affichage des parties statiques
+ */
+void AffichageWindow()
+{
+  window = subwin(stdscr, 19, 62, 0, 0);
+
+  box(window, 0, 0);
+
+  // Titre
+  mvwprintw(window, 1, 2, "Château d'eau (11/2024)");
+
+  // I/O
+  // Ligne du haut
+  mvwaddch(window, 2, 0, ACS_LTEE);
+  mvwhline(window, 2, 1, 0, 60);
+  mvwaddch(window, 2, 61, ACS_RTEE);
+
+  // Ligne du bas
+  mvwaddch(window, 7, 0, ACS_LTEE);
+  mvwhline(window, 7, 1, 0, 60);
+  mvwaddch(window, 7, 61, ACS_RTEE);
+
+  // Séparation verticale
+  mvwaddch(window, 2, 18, ACS_TTEE);
+  mvwvline(window, 3, 18, 0, 4);
+  mvwaddch(window, 7, 18, ACS_BTEE);
+
+  // Input : Boutons poussoirs
+  mvwprintw(window, 3, 2, "BP 1");
+  mvwprintw(window, 4, 2, "BP 2");
+  mvwprintw(window, 5, 2, "BP 3");
+  mvwprintw(window, 6, 2, "BP 4");
+
+  // Output : Moteurs pompes
+  mvwprintw(window, 3, 20, "Pompe 1");
+  mvwprintw(window, 4, 20, "Pompe 2");
+  mvwprintw(window, 5, 20, "Pompe 3");
+  mvwprintw(window, 6, 20, "Pompe 4");
+
+  // Mesures
+  mvwprintw(window, 8, 2, "Debit en sortie");
+  mvwprintw(window, 9, 2, "Debit en entrée");
+  mvwprintw(window, 10, 2, "Volume dans le réservoir");
+
+  // Graphe
+
+  // Ligne du haut
+  mvwaddch(window, 11, 0, ACS_LTEE);
+  mvwhline(window, 11, 1, 0, 60);
+  mvwaddch(window, 11, 61, ACS_RTEE);
+
+  // Ligne du bas
+  mvwaddch(window, 13, 0, ACS_LTEE);
+  mvwhline(window, 13, 1, 0, 60);
+  mvwaddch(window, 13, 61, ACS_RTEE);
+
+  // Graduations
+  for (int i = 1 ; i < 11; i++) {
+    mvwaddch(window, 11, 6 * i, ACS_TTEE);
+    mvwaddch(window, 13, 6 * i, ACS_BTEE);
+  }
+
+  mvwaddch(window, 13, 2 * 6, ACS_PLUS);
+  mvwaddch(window, 13, 6 * 6, ACS_PLUS);
+  mvwaddch(window, 13, 7 * 6, ACS_PLUS);
+  mvwaddch(window, 13, 57, ACS_TTEE);  //9.5 x 6
+
+  mvwvline(window, 12, 6 * 10, 0, 1);
+
+  // Légende
+  mvwaddch(window, 16, 0, ACS_LTEE);
+  mvwhline(window, 16, 1, 0, 60);
+  mvwaddch(window, 16, 61, ACS_RTEE);
+
+  mvwprintw(window, 14, 2 * 6 - 1, "min");
+  mvwprintw(window, 14, 6 * 6 - 2, "low");
+  mvwprintw(window, 14, 7 * 6, "high");
+  mvwprintw(window, 14, 56, "max");
+
+  // Informations
+  mvwprintw(window, 17, 2, "Mode");
+
+  mvwaddch(window, 16, 18, ACS_TTEE);
+  mvwaddch(window, 18, 18, ACS_BTEE);
+  mvwvline(window, 17, 18, 0, 1);
+  mvwprintw(window, 17, 20, "Grafcet");
+
+  mvwaddch(window, 16, 31, ACS_TTEE);
+  mvwaddch(window, 18, 31, ACS_BTEE);
+  mvwvline(window, 17, 31, 0, 1);
+  mvwprintw(window, 17, 33, "min");
+
+  mvwprintw(window, 17, 46, "max");
+
+  wrefresh(window);
+}
+
+void Affichage()
+{
+  mvwprintw(window, 1, 50, "%5.1f s", t_elapsed);
+
+  for (int i = OUT_PUMP_1; i <= OUT_PUMP_4; i++)
+  {
+    mvwprintw(window, (short)(i - OUT_PUMP_1 + 3), 30, " %c    %5.1f s   %dx", _digital[i].mode & 0x01 ? _digital[i].ivalue ? 'M' : '.' : 'X', _digital[i].duration, _digital[i].nb);
+  }
+
+  for (int i = IN_KEYBOARD_1; i <= IN_KEYBOARD_4; i++)
+  {
+    mvwprintw(window, (short)(i + 3), 10, _digital[_keyboard[i].input].ivalue ? "1" : "0");
+  }
+
+  mvwprintw(window, 8, 28, "%3d    l/s", (int)_digital[IN_FLOW_OUT].dvalue);
+  mvwprintw(window, 9, 28, "%3d    l/s", (int)_digital[IN_FLOW_IN].dvalue);
+  mvwprintw(window, 10, 29, "%5.2lf m3    (%+d l/s)", _digital[IN_TANK_LEVEL].dvalue, (int)_digital[IN_FLOW_DIF].dvalue);
+
+  AffichageGraphe(12, 1, _digital[IN_TANK_LEVEL].dvalue * 6);
+
+/*
+  if (_digital[IN_SENSOR_LOW].mode & 0x01)
+  {
+    mvwprintw(window, 15, 1, "          (%dx)                    (%dx)        (%dx)    (%dx)", _digital[IN_SENSOR_MIN].nb, _digital[IN_SENSOR_LOW].nb, _digital[IN_SENSOR_HIGH].nb, _digital[IN_SENSOR_MAX].nb);
+  }
+  else
+  {
+    mvwprintw(window, 15, 1, "          (%dx)                  X (%dx)        (%dx)    (%dx)", _digital[IN_SENSOR_MIN].nb, _digital[IN_SENSOR_LOW].nb, _digital[IN_SENSOR_HIGH].nb, _digital[IN_SENSOR_MAX].nb);
+  }
+  */
+  mvwprintw(window, 14, 15, "%d", _digital[IN_SENSOR_MIN].ivalue);
+  mvwprintw(window, 14, 38, "%d", _digital[IN_SENSOR_LOW].ivalue);
+  mvwprintw(window, 14, 47, "%d", _digital[IN_SENSOR_HIGH].ivalue);
+  mvwprintw(window, 14, 60, "%d", _digital[IN_SENSOR_MAX].ivalue);
+
+  mvwprintw(window, 15, 11, "(%dx) %s", _digital[IN_SENSOR_MIN].nb, _digital[IN_SENSOR_MIN].mode & 0x01 ? " " : "D");
+  mvwprintw(window, 15, 34, "(%dx) %s", _digital[IN_SENSOR_LOW].nb, _digital[IN_SENSOR_LOW].mode & 0x01 ? " " : "D");
+  mvwprintw(window, 15, 42, "(%dx) %s", _digital[IN_SENSOR_HIGH].nb, _digital[IN_SENSOR_HIGH].mode & 0x01 ? " " : "D");
+  mvwprintw(window, 15, 56, "(%dx)", _digital[IN_SENSOR_MAX].nb);
+
+  // Informations complémentaires
+  mvwprintw(window, 17, 2, _digital[OUT_DISPLAY_MODE].ivalue ? "Automatique" : "Manuel     ");
+  mvwprintw(window, 17, 28, "%d", _digital[OUT_DISPLAY_GRAFCET].ivalue);
+  mvwprintw(window, 17, 38, "%4.2lf m3", _digital[IN_TANK_MIN].dvalue);
+  mvwprintw(window, 17, 51, "%4.2lf m3", _digital[IN_TANK_MAX].dvalue);
+
+  wrefresh(window);
+}
+
+void AffichageGraphe(int y, int x, double value)
+{
+  int entier = (int)(value);
+  int i;
+
+  for (i = 0; i < entier; i++)
+  {
+    mvwaddwstr(window, y, x + i, L"█"); // U+2588
+  }
+
+  int frac = (int)((value - entier) * 4);
+  if (frac > 3) // 0.75 -> 0.99
+  {
+    mvwaddwstr(window, y, x + i, L"▊"); // U+258A
+    entier += 1;
+  }
+  if (frac > 2) // 0.5 -> 0.99
+  {
+    mvwaddwstr(window, y, x + i, L"▌"); // U+258C
+    entier += 1;
+  }
+  else if (frac > 1) // 0.25 -> 0.49
+  {
+    mvwaddwstr(window, y, x + i, L"▎"); //U+258E
+    entier += 1;
+  }
+
+  for (int i = entier; i < 59; i++)
+  {
+    mvwprintw(window, y, x + i, " ");
+  }
+}
 
 /**
  * Prometheus
  */
 void InitPrometheus()
 {
+  static Exposer exposer{"0.0.0.0:8099"};
 
+  // Le registre central
+  registry = std::make_shared<Registry>();
 
-}
+  exposer.RegisterCollectable(registry);
 
-void ProcessPrometheus()
-{
+  auto& gauge_volume = BuildGauge()
+      .Name("geii_volume")
+      .Help("Volume en m3")
+      .Register(*registry);
 
+  tank_gauge = &gauge_volume.Add({});
+
+  auto& gauge_debit = BuildGauge()
+      .Name("geii_debit")
+      .Help("Débit en l/s")
+      .Register(*registry);
+
+  debit_entree = &gauge_debit.Add({{"numero", "entree"}});
+  debit_sortie = &gauge_debit.Add({{"numero", "sortie"}});
+
+  debit_p1 = &gauge_debit.Add({{"numero", "1"}});
+  debit_p2 = &gauge_debit.Add({{"numero", "2"}});
+  debit_p3 = &gauge_debit.Add({{"numero", "3"}});
+  debit_p4 = &gauge_debit.Add({{"numero", "4"}});
+
+  auto& counter_debit = BuildCounter()
+      .Name("geii_litre")
+      .Help("Volume en l")
+      .Register(*registry);
+
+  volume_p1 = &counter_debit.Add({{"numero", "1"}});
+  volume_p2 = &counter_debit.Add({{"numero", "2"}});
+  volume_p3 = &counter_debit.Add({{"numero", "3"}});
+  volume_p4 = &counter_debit.Add({{"numero", "4"}});
+
+  auto& hist_volume = BuildHistogram()
+        .Name("geii_tank")
+        .Help("volume du reservoir en m3")
+        .Register(*registry);
+
+  tank_histogram = &hist_volume.Add({}, buckets);
 }
 
 void ProcessMQTT(mqtt::async_client* client)
@@ -393,19 +1039,35 @@ void ProcessMQTT(mqtt::async_client* client)
   json obj = {
     {"entree", _digital[IN_FLOW_IN].dvalue},
     {"sortie", _digital[IN_FLOW_OUT].dvalue},
-    {"tank", _digital[IN_TANK_LEVEL].dvalue},
-    {"pompe1", _digital[IN_FLOW_1].dvalue},
-    {"pompe2", _digital[IN_FLOW_2].dvalue},
-    {"pompe3", _digital[IN_FLOW_3].dvalue},
-    {"pompe4", _digital[IN_FLOW_4].dvalue},
-    {"min",  _digital[IN_SENSOR_MIN].ivalue},
-    {"low",  _digital[IN_SENSOR_LOW].ivalue},
-    {"high",  _digital[IN_SENSOR_HIGH].ivalue},
-    {"max",  _digital[IN_SENSOR_MAX].ivalue},
+    {"p1", _digital[IN_FLOW_1].dvalue},
+    {"p2", _digital[IN_FLOW_2].dvalue},
+    {"p3", _digital[IN_FLOW_3].dvalue},
+    {"p4", _digital[IN_FLOW_4].dvalue},
+    {"level", _digital[IN_TANK_LEVEL].dvalue}
   };
 
   std::string payload = obj.dump();
+
   auto msg = mqtt::make_message("geii/telemetry", payload);
   msg->set_qos(1);
   client->publish(msg);
 }
+
+void ProcessPrometheus()
+{
+  tank_gauge->Set(_digital[IN_TANK_LEVEL].dvalue);
+  tank_histogram->Observe(_digital[IN_TANK_LEVEL].dvalue);
+
+  debit_entree->Set(_digital[IN_FLOW_IN].dvalue);
+  debit_sortie->Set(_digital[IN_FLOW_OUT].dvalue);
+
+  debit_p1->Set(_digital[IN_FLOW_1].dvalue);
+  debit_p2->Set(_digital[IN_FLOW_2].dvalue);
+  debit_p3->Set(_digital[IN_FLOW_3].dvalue);
+  debit_p4->Set(_digital[IN_FLOW_4].dvalue);
+
+  volume_p1->Increment(_digital[IN_FLOW_1].dvalue * dt);
+  volume_p2->Increment(_digital[IN_FLOW_2].dvalue * dt);
+  volume_p3->Increment(_digital[IN_FLOW_3].dvalue * dt);
+  volume_p4->Increment(_digital[IN_FLOW_4].dvalue * dt);
+}

main.hpp

@@ -1,8 +1,15 @@
 #undef timeout
 #include "mqtt/async_client.h"
 
+void ConsoleInit();
+
+void LireClavier(int ch);
+void LireEntree();
+void EvolutionGrafcet();
+void Actions();
 void RemiseZeroInput();
 
+void ProcessInitKeyboard();
 void ProcessInitIO();
 void ProcessInitValues();
 double ProcessMoteur(int i);
@@ -17,6 +24,24 @@ void ProcessMQTT(mqtt::async_client* client);
 double SimulConsoSinusoidale(long t);
 double SimulConsoBrown(double valeur_precedente);
 
+void AffichageWindow();
+void Affichage();
+void AffichageGraphe(int y, int x, double value);
+
+// KEYBOARD INPUT
+
+#define IN_KEYBOARD_1 0
+#define IN_KEYBOARD_2 1
+#define IN_KEYBOARD_3 2
+#define IN_KEYBOARD_4 3
+#define IN_KEYBOARD_A 4
+
+#define IN_KEYBOARD_X 5
+#define IN_KEYBOARD_7 6
+#define IN_KEYBOARD_8 7
+#define IN_KEYBOARD_9 8
+#define IN_KEYBOARD_0 9
+
 // DIGITAL INPUT
 
 #define IN_SENSOR_MIN 10

thread.c

@@ -0,0 +1,315 @@
+/*
+ * amqp_machine.c
+ *
+ * Exemple multithread AMQP (rabbitmq-c) + boucle machine déterministe.
+ *
+ * Compilation :
+ *   gcc amqp_machine.c -lrabbitmq -lpthread -o amqp_machine
+ *
+ * Pré-requis :
+ *   librabbitmq-dev (rabbitmq-c)
+ *
+ * Test de publication :
+ *   rabbitmqadmin publish exchange=amq.default routing_key=geii_orders payload="START"
+ *   rabbitmqadmin publish exchange=amq.default routing_key=geii_orders payload="STOP"
+ *   rabbitmqadmin publish exchange=amq.default routing_key=geii_orders payload="E_STOP"
+ *
+ */
+
+ /*
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <signal.h>
+#include <pthread.h>
+#include <stdatomic.h>
+
+#include <amqp.h>
+#include <amqp_tcp_socket.h>
+
+// Connexion AMQP
+#define HOST "localhost"
+#define PORT 5672
+#define USER "guest"
+#define PASS "guest"
+#define VHOST "/"
+#define QUEUE "geii_orders"
+
+// Boucle machine
+#define CYCLE_MS 100
+
+// Structures de la queue interne
+typedef struct node {
+    char *msg;
+    struct node *next;
+} node_t;
+
+typedef struct {
+    node_t *head;
+    node_t *tail;
+    pthread_mutex_t mtx;
+} queue_t;
+
+static void queue_init(queue_t *q) {
+    q->head = q->tail = NULL;
+    pthread_mutex_init(&q->mtx, NULL);
+}
+
+static void queue_push(queue_t *q, const char *s) {
+    node_t *n = malloc(sizeof(node_t));
+    n->msg = strdup(s);
+    n->next = NULL;
+    pthread_mutex_lock(&q->mtx);
+    if (q->tail) q->tail->next = n;
+    q->tail = n;
+    if (!q->head) q->head = n;
+    pthread_mutex_unlock(&q->mtx);
+}
+
+static char *queue_pop_all_and_get_last(queue_t *q) {
+    // Vide la queue et retourne la dernière chaine (caller doit free()).
+    // Retourne NULL si queue vide.
+    pthread_mutex_lock(&q->mtx);
+    if (!q->head) {
+        pthread_mutex_unlock(&q->mtx);
+        return NULL;
+    }
+    char *last = NULL;
+    node_t *cur = q->head;
+    while (cur) {
+        if (last) free(last);
+        last = strdup(cur->msg);
+        node_t *tmp = cur;
+        cur = cur->next;
+        free(tmp->msg);
+        free(tmp);
+    }
+    q->head = q->tail = NULL;
+    pthread_mutex_unlock(&q->mtx);
+    return last;
+}
+
+static void queue_destroy(queue_t *q) {
+    pthread_mutex_lock(&q->mtx);
+    node_t *cur = q->head;
+    while (cur) {
+        node_t *tmp = cur;
+        cur = cur->next;
+        free(tmp->msg);
+        free(tmp);
+    }
+    q->head = q->tail = NULL;
+    pthread_mutex_unlock(&q->mtx);
+    pthread_mutex_destroy(&q->mtx);
+}
+
+// Etats machine
+typedef enum {
+    STATE_STOPPED,
+    STATE_RUNNING,
+    STATE_ESTOP
+} machine_state_t;
+
+// Variables globales de contrôle
+static atomic_int running = 1;        // 0 = arrêt demandé
+static atomic_int estop_flag = 0;     // 1 = E-STOP actif
+static queue_t orders_queue;
+static machine_state_t machine_state = STATE_STOPPED;
+
+// Fonctions "appliquer" (à remplacer par actions réelles)
+static void apply_start(void) {
+    if (machine_state != STATE_RUNNING) {
+        printf("[MACHINE] -> START\n");
+        // TODO: démarrer moteurs / sorties
+        machine_state = STATE_RUNNING;
+    }
+}
+
+static void apply_stop(void) {
+    if (machine_state != STATE_STOPPED) {
+        printf("[MACHINE] -> STOP\n");
+        // TODO: couper moteurs / sorties
+        machine_state = STATE_STOPPED;
+    }
+}
+
+static void apply_estop(void) {
+    if (machine_state != STATE_ESTOP) {
+        printf("[MACHINE] -> E-STOP (arrêt d'urgence)\n");
+        // TODO: couper puissance, sécurité
+        machine_state = STATE_ESTOP;
+    }
+}
+
+// Thread machine : boucle déterministe
+void *machine_thread_fn(void *arg) {
+    (void)arg;
+    const int cycle_us = CYCLE_MS * 1000;
+    while (atomic_load(&running)) {
+        // 1) Vérifier estop immédiat
+        if (atomic_load(&estop_flag)) {
+            apply_estop();
+            // On peut décider ici de vider la queue ou de la garder
+            //   mais on ignore les autres commandes jusqu'à reset.
+            usleep(cycle_us);
+            continue;
+        }
+
+        // 2) Dépiler toute la file et ne garder que la dernière commande
+        char *last = queue_pop_all_and_get_last(&orders_queue);
+        if (last) {
+            // Normaliser message (trim)
+            if (strcmp(last, "START") == 0) {
+                apply_start();
+            } else if (strcmp(last, "STOP") == 0) {
+                apply_stop();
+            } else {
+                printf("[MACHINE] Commande inconnue reçue: '%s'\n", last);
+            }
+            free(last);
+        }
+
+        // 3) Exécuter la logique machine déterministe (boucle cycle)
+        // TODO: lire capteurs, asservissements, sorties...
+
+        usleep(cycle_us);
+    }
+
+    // Fin : safe stop
+    apply_stop();
+    return NULL;
+}
+
+// Thread AMQP : consomme messages et les place dans la queue
+//   Si message == "E_STOP" => set estop_flag immédiatement (ne pas mettre en queue)
+
+void *amqp_thread_fn(void *arg) {
+    (void)arg;
+    amqp_connection_state_t conn = amqp_new_connection();
+    amqp_socket_t *socket = amqp_tcp_socket_new(conn);
+    if (!socket) {
+        fprintf(stderr, "[AMQP] Erreur: création socket\n");
+        atomic_store(&running, 0);
+        return NULL;
+    }
+    if (amqp_socket_open(socket, HOST, PORT)) {
+        fprintf(stderr, "[AMQP] Erreur: ouverture connexion %s:%d\n", HOST, PORT);
+        atomic_store(&running, 0);
+        return NULL;
+    }
+
+    amqp_rpc_reply_t rpc_reply = amqp_login(conn, VHOST, 0, 131072, 60,
+                                            AMQP_SASL_METHOD_PLAIN, USER, PASS);
+    if (rpc_reply.reply_type != AMQP_RESPONSE_NORMAL) {
+        fprintf(stderr, "[AMQP] Erreur login\n");
+        atomic_store(&running, 0);
+        return NULL;
+    }
+
+    amqp_channel_open(conn, 1);
+    amqp_get_rpc_reply(conn);
+
+    // Déclarer la queue (idempotent)
+    amqp_queue_declare(conn, 1, amqp_cstring_bytes(QUEUE),
+                       0, 0, 0, 1, amqp_empty_table());
+    amqp_get_rpc_reply(conn);
+
+    // Démarrer la consommation
+    amqp_basic_consume(conn, 1,
+                       amqp_cstring_bytes(QUEUE),
+                       amqp_empty_bytes, // consumer tag auto
+                       0, // no_local
+                       1, // no_ack = 1 (auto-ack) ; ajuster selon besoin
+                       0, // exclusive
+                       amqp_empty_table());
+    amqp_get_rpc_reply(conn);
+
+    printf("[AMQP] En attente de messages sur '%s'...\n", QUEUE);
+
+    while (atomic_load(&running)) {
+        amqp_envelope_t envelope;
+        amqp_maybe_release_buffers(conn);
+
+        struct timeval timeout;
+        timeout.tv_sec = 0;
+        timeout.tv_usec = 500000; // 500 ms
+
+        rpc_reply = amqp_consume_message(conn, &envelope, &timeout, 0);
+
+        if (rpc_reply.reply_type == AMQP_RESPONSE_NORMAL) {
+            // Message reçu
+            size_t len = envelope.message.body.len;
+            char *body = malloc(len + 1);
+            memcpy(body, envelope.message.body.bytes, len);
+            body[len] = '\0';
+
+            printf("[AMQP] Reçu: '%s'\n", body);
+
+            if (strcmp(body, "E_STOP") == 0) {
+                // Priorité absolue : traiter immédiatement
+                atomic_store(&estop_flag, 1);
+                printf("[AMQP] E-STOP reçu : flag estop activé\n");
+                free(body);
+            } else {
+                // Push dans la queue pour traitement au prochain cycle
+                queue_push(&orders_queue, body);
+                free(body);
+            }
+
+            amqp_destroy_envelope(&envelope);
+        } else if (rpc_reply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION &&
+                   rpc_reply.library_error == AMQP_STATUS_TIMEOUT) {
+            // timeout - pas de message
+            // rien
+        } else {
+            // autre erreur - tentatives de reconnexion possibles
+            fprintf(stderr, "[AMQP] Erreur consume (reconnexion nécessaire?)\n");
+            sleep(1);
+            // Ici on pourrait tenter de reconnecter proprement ; pour cet exemple,
+            //   on continue la boucle et laisse l'admin redémarrer si nécessaire
+        }
+    }
+
+    // Fermeture
+    amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
+    amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
+    amqp_destroy_connection(conn);
+    return NULL;
+}
+
+// Handler SIGINT pour arrêt propre
+static void sigint_handler(int signum) {
+    (void)signum;
+    printf("[MAIN] SIGINT reçu : arrêt\n");
+    atomic_store(&running, 0);
+}
+
+int main(int argc, char **argv) {
+    (void)argc; (void)argv;
+    signal(SIGINT, sigint_handler);
+
+    queue_init(&orders_queue);
+
+    pthread_t th_amqp, th_machine;
+    if (pthread_create(&th_amqp, NULL, amqp_thread_fn, NULL) != 0) {
+        perror("pthread_create amqp");
+        return 1;
+    }
+    if (pthread_create(&th_machine, NULL, machine_thread_fn, NULL) != 0) {
+        perror("pthread_create machine");
+        atomic_store(&running, 0);
+        pthread_join(th_amqp, NULL);
+        return 1;
+    }
+
+    // Attendre threads
+    pthread_join(th_amqp, NULL);
+    pthread_join(th_machine, NULL);
+
+    queue_destroy(&orders_queue);
+
+    printf("[MAIN] Terminé.\n");
+    return 0;
+}
+*/