Universite/c_pompes
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: Universite:main
Branches Tags
Universite:main Universite:dev
..
compare: Universite:dev
Branches Tags
Universite:main Universite:dev

4 Commits
main ... dev

Author SHA1 Message Date
medina5
3ce3cecc34 JSON 2025-12-07 16:48:16 +01:00
medina5
70aa2687d0 ProcessMQTT 2025-12-07 16:04:44 +01:00
medina5
9d29b94a96 MQTT 2025-12-07 15:51:48 +01:00
medina5
8c8dc55db4 Exception 2025-12-07 15:51:07 +01:00
13 changed files with 1256 additions and 308 deletions
Expand all files Collapse all files

15
.devcontainer/Dockerfile
View File

@@ -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"]

54
.devcontainer/compose.yml
View File

@@ -1,54 +0,0 @@
services:
dev:
build:
context: .
dockerfile: Dockerfile
volumes:
- ..:/workspace:cached
command: sleep infinity
networks:
- dev_net
nodered:
build: ./nodered
container_name: nodered
ports:
- "1880:1880"
networks:
- dev_net
environment:
TZ: Europe/Paris
volumes:
- nodered:/data
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:
nodered:
rabbitmq:

12
.devcontainer/devcontainer.json
View File

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

7
.devcontainer/nodered/Dockerfile
View File

@@ -1,7 +0,0 @@
FROM nodered/node-red:4.1
# Installer FlowFuse Dashboard
RUN npm install --unsafe-perm @flowfuse/node-red-dashboard
# Copier votre flux si nécessaire
COPY flows.json /data/flows.json

147
.devcontainer/nodered/flows.json
View File

@@ -1,147 +0,0 @@
[
{
"id": "1ba681875b5ebc3a",
"type": "ui-gauge",
"z": "41526b8c80d5a5f7",
"name": "",
"group": "5628ce0b7dc12431",
"order": 6,
"value": "$round(payload.pompe4,1)",
"valueType": "jsonata",
"width": 3,
"height": 3,
"gtype": "gauge-half",
"gstyle": "needle",
"title": "Pompe 4",
"alwaysShowTitle": false,
"floatingTitlePosition": "top-left",
"units": "l/s",
"icon": "",
"prefix": "",
"suffix": "",
"segments": [
{
"from": "0",
"color": "#ffc800",
"text": "",
"textType": "label"
},
{
"from": "65",
"color": "#5cd65c",
"text": "",
"textType": "label"
}
],
"min": 0,
"max": "80",
"sizeThickness": 16,
"sizeGap": 4,
"sizeKeyThickness": 8,
"styleRounded": true,
"styleGlow": false,
"className": "",
"x": 360,
"y": 1560,
"wires": [
[]
]
},
{
"id": "5628ce0b7dc12431",
"type": "ui-group",
"name": "Pompes",
"page": "5e9c1f56ed1b2af5",
"width": 6,
"height": 1,
"order": 2,
"showTitle": true,
"className": "",
"visible": "true",
"disabled": "false",
"groupType": "default"
},
{
"id": "5e9c1f56ed1b2af5",
"type": "ui-page",
"name": "Tableau de bord",
"ui": "15f54ff2469ec593",
"path": "/page1",
"icon": "home",
"layout": "grid",
"theme": "64cdf126ed34d37d",
"breakpoints": [
{
"name": "Default",
"px": "0",
"cols": "3"
},
{
"name": "Tablet",
"px": "576",
"cols": "6"
},
{
"name": "Small Desktop",
"px": "768",
"cols": "9"
},
{
"name": "Desktop",
"px": "1024",
"cols": "12"
}
],
"order": 1,
"className": "",
"visible": true,
"disabled": false
},
{
"id": "15f54ff2469ec593",
"type": "ui-base",
"name": "My Dashboard",
"path": "/dashboard",
"appIcon": "",
"includeClientData": true,
"acceptsClientConfig": [
"ui-notification",
"ui-control"
],
"showPathInSidebar": false,
"headerContent": "page",
"navigationStyle": "default",
"titleBarStyle": "default",
"showReconnectNotification": true,
"notificationDisplayTime": 1,
"showDisconnectNotification": true,
"allowInstall": false
},
{
"id": "64cdf126ed34d37d",
"type": "ui-theme",
"name": "Default Theme",
"colors": {
"surface": "#ffffff",
"primary": "#813d9c",
"bgPage": "#eeeeee",
"groupBg": "#ffffff",
"groupOutline": "#cccccc"
},
"sizes": {
"density": "default",
"pagePadding": "12px",
"groupGap": "12px",
"groupBorderRadius": "4px",
"widgetGap": "12px"
}
},
{
"id": "2b9a5dc0f1a61045",
"type": "global-config",
"env": [],
"modules": {
"@flowfuse/node-red-dashboard": "1.29.0"
}
}
]

5
.vscode/launch.json vendored
View File

@@ -2,10 +2,10 @@
"version": "0.2.0",
"configurations": [
{
"name": "Debug pompes",
"name": "Debug geii_exporter",
"type": "cppdbg",
"request": "launch",
"program": "${workspaceFolder}/build/pompes",
"program": "${workspaceFolder}/build/geii_exporter",
"args": [],
"stopAtEntry": false,
"cwd": "${workspaceFolder}",
@@ -23,3 +23,4 @@
}
]
}

26
AutomForArduino.cpp
View File

@@ -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
}

6
CMakeLists.txt
View File

@@ -1,5 +1,5 @@
cmake_minimum_required(VERSION 3.15)
project(pompes LANGUAGES CXX)
project(geii_exporter LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
@@ -9,7 +9,7 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -Wpedantic")
# Executable principal
add_executable(pompes
add_executable(geii_exporter
main.cpp
)
@@ -45,7 +45,7 @@ find_package(prometheus-cpp REQUIRED)
# -------------------------------
# Lien des bibliothèques
# -------------------------------
target_link_libraries(pompes
target_link_libraries(geii_exporter
prometheus-cpp::core
prometheus-cpp::pull
${CURL_LIBRARIES}

4
main-mqtt.cpp
View File

@@ -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);

163
main-mqtt2.cpp Normal file
View File

@@ -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;
}

785
main.cpp
View File

@@ -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,11 +28,13 @@
#include "mqtt/async_client.h"
#include <nlohmann/json.hpp>
using namespace std::chrono_literals;
using json = nlohmann::json;
// Constantes de fonctionnement
#define LEVEL_MIN 2
#define FLOW_PER_PUMP 75
#define FLOW_PER_PUMP 150
/* Configuration MQTT */
const std::string ADDRESS = "tcp://rabbitmq:1883";
@@ -42,61 +43,185 @@ 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)
unsigned short pompe1_old, pompe2_old, pompe3_old, pompe4_old;
unsigned short sensor_max, sensor_high, sensor_low, sensor_min;
float TankInitalValue = 5;
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";
}
}
*/
try {
json j = json::parse(payload);
// Ne rien faire si l'objet JSON est vide
if (j.empty()) return;
if (j.contains("p1")) pompe1 = j["p1"].get<int>() != 0;
if (j.contains("p2")) pompe2 = j["p2"].get<int>() != 0;
if (j.contains("p3")) pompe3 = j["p3"].get<int>() != 0;
if (j.contains("p4")) pompe4 = j["p4"].get<int>() != 0;
std::cout << "Pompes : " << pompe1 << " " << pompe2 << " " << pompe3 << " " << pompe4 << std::endl;
}
catch (const json::parse_error& e) {
std::cerr << "Erreur JSON : " << e.what() << "\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);
@@ -104,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();
@@ -114,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);
@@ -135,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);
@@ -215,8 +604,8 @@ void ProcessInitValues()
_digital[OUT_FLOW_OUT_AMPLITUDE].dvalue = 100.0;
_digital[OUT_LEVEL_MIN].dvalue = LEVEL_MIN;
_digital[OUT_LEVEL_LOW].dvalue = 4.3;
_digital[OUT_LEVEL_HIGH].dvalue = 6.1;
_digital[OUT_LEVEL_LOW].dvalue = 6;
_digital[OUT_LEVEL_HIGH].dvalue = 7;
_digital[OUT_LEVEL_MAX].dvalue = 9.5;
_digital[IN_FLOW_OUT].dvalue = 100.0;
@@ -267,8 +656,9 @@ double ProcessMoteur(int i)
void ProcessException()
{
if (t_elapsed > 30) {
//_digital[OUT_PUMP_1].mode = 0;
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;
}
@@ -319,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;
@@ -362,6 +770,8 @@ void Process()
_digital[IN_SENSOR_HIGH].ivalue = test;
}
Affichage();
t_backup = t;
}
@@ -378,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;
@@ -392,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)
@@ -412,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);
}

25
main.hpp
View File

@@ -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

315
thread.c Normal file
View File

@@ -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;
}
*/