Merge pull request #48 from Mes-Open/develop

Author: jakub-przepiora

README.md

@@ -22,6 +22,7 @@
 
 [![Discord](https://img.shields.io/badge/Discord-Join%20us-5865F2?logo=discord&logoColor=white)](https://discord.gg/fw3fG78pZj)
 
+
 </div>
 
 ---

backend/app/Console/Commands/ModbusPollCommand.php

@@ -0,0 +1,81 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Models\MachineConnection;
+use App\Services\Machine\MachineSignalIngestor;
+use App\Services\Machine\Modbus\ModbusReader;
+use Illuminate\Console\Command;
+
+/**
+ * Long-running Modbus TCP poller. One process per machine_connection: connects,
+ * then every poll_interval_ms reads all active tags and feeds each into the
+ * MachineSignalIngestor. Reconnects with backoff on transport errors.
+ *
+ *   php artisan modbus:poll --connection=3
+ */
+class ModbusPollCommand extends Command
+{
+    protected $signature = 'modbus:poll {--connection= : machine_connection id} {--once : single poll cycle then exit (for testing)}';
+
+    protected $description = 'Poll a Modbus TCP device and ingest machine signals';
+
+    public function handle(MachineSignalIngestor $ingestor, \App\Services\Machine\RuntimeMonitor $runtime): int
+    {
+        $connection = MachineConnection::with(['modbusConnection', 'activeTags.workstation'])
+            ->find($this->option('connection'));
+
+        if (! $connection || ! $connection->modbusConnection) {
+            $this->error('Modbus connection not found.');
+
+            return self::FAILURE;
+        }
+
+        $modbus = $connection->modbusConnection;
+        $tags = $connection->activeTags;
+        $intervalUs = max(100, $modbus->poll_interval_ms) * 1000;
+        $once = (bool) $this->option('once');
+
+        $this->info("Polling {$connection->name} ({$modbus->host}:{$modbus->port}), {$tags->count()} tags");
+
+        do {
+            $reader = new ModbusReader($modbus);
+            try {
+                $reader->connect();
+                $connection->markConnected();
+
+                do {
+                    $cycleStart = microtime(true);
+                    $runtime->heartbeat($connection->protocol, $connection->id);
+                    foreach ($tags as $tag) {
+                        try {
+                            $value = $reader->readTag($tag);
+                            $ingestor->ingest($tag, $value);
+                            $connection->increment('messages_received');
+                        } catch (\Throwable $e) {
+                            $this->warn("tag {$tag->name}: {$e->getMessage()}");
+                        }
+                    }
+
+                    if ($once) {
+                        break 2;
+                    }
+
+                    $elapsed = (int) ((microtime(true) - $cycleStart) * 1_000_000);
+                    usleep(max(0, $intervalUs - $elapsed));
+                } while (true);
+            } catch (\Throwable $e) {
+                $connection->markError($e->getMessage());
+                $this->error("connection error: {$e->getMessage()}");
+                if ($once) {
+                    return self::FAILURE;
+                }
+                sleep(5); // backoff before reconnect
+            } finally {
+                $reader->close();
+            }
+        } while (! $once);
+
+        return self::SUCCESS;
+    }
+}

backend/app/Console/Commands/ModbusSimulateCommand.php

@@ -0,0 +1,150 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+
+/**
+ * Minimal Modbus TCP *server* that simulates a machine, so the real
+ * modbus:poll daemon has something to read end-to-end without hardware.
+ *
+ * Register map (holding & input registers share the same backing store):
+ *   0 → state        (1=RUNNING, 2=IDLE, 3=FAULT)
+ *   1 → good counter (increments each second while RUNNING)
+ *   2 → reject counter
+ *   3 → temperature  (telemetry, °C ×10)
+ *
+ * The machine cycles RUNNING → IDLE → RUNNING → FAULT … on a timer.
+ *
+ *   php artisan modbus:simulate --port=5020
+ */
+class ModbusSimulateCommand extends Command
+{
+    protected $signature = 'modbus:simulate {--port=5020} {--host=0.0.0.0}';
+
+    protected $description = 'Run a simulated Modbus TCP machine (for testing the poller)';
+
+    /** @var array<int,int> holding/input registers */
+    private array $reg = [0 => 1, 1 => 0, 2 => 0, 3 => 220];
+
+    public function handle(): int
+    {
+        $host = $this->option('host');
+        $port = (int) $this->option('port');
+
+        $server = @stream_socket_server("tcp://{$host}:{$port}", $errno, $errstr);
+        if (! $server) {
+            $this->error("Cannot bind {$host}:{$port}: {$errstr}");
+
+            return self::FAILURE;
+        }
+        stream_set_blocking($server, false);
+        $this->info("Modbus simulator listening on {$host}:{$port} (Ctrl+C to stop)");
+
+        $clients = [];
+        $lastTick = microtime(true);
+        // Scripted state timeline (state, seconds)
+        $script = [[1, 15], [2, 5], [1, 20], [3, 8], [1, 15], [2, 4]];
+        $phase = 0;
+        $phaseElapsed = 0.0;
+
+        while (true) {
+            $read = array_merge([$server], $clients);
+            $write = $except = null;
+            if (@stream_select($read, $write, $except, 1) === false) {
+                break;
+            }
+
+            foreach ($read as $sock) {
+                if ($sock === $server) {
+                    $client = @stream_socket_accept($server, 0);
+                    if ($client) {
+                        stream_set_blocking($client, false);
+                        $clients[(int) $client] = $client;
+                    }
+
+                    continue;
+                }
+
+                $data = @fread($sock, 1024);
+                if ($data === '' || $data === false) {
+                    fclose($sock);
+                    unset($clients[(int) $sock]);
+
+                    continue;
+                }
+                $response = $this->handleRequest($data);
+                if ($response !== null) {
+                    @fwrite($sock, $response);
+                }
+            }
+
+            // Advance simulation on ~1s ticks.
+            $now = microtime(true);
+            $dt = $now - $lastTick;
+            if ($dt >= 1.0) {
+                $lastTick = $now;
+                $phaseElapsed += $dt;
+                [$state, $duration] = $script[$phase];
+                $this->reg[0] = $state;
+                if ($state === 1) {            // RUNNING
+                    $this->reg[1] += 2;        // good parts
+                    if (random_int(1, 10) === 1) {
+                        $this->reg[2] += 1;    // occasional reject
+                    }
+                    $this->reg[3] = 220 + random_int(-5, 15);
+                }
+                if ($phaseElapsed >= $duration) {
+                    $phaseElapsed = 0;
+                    $phase = ($phase + 1) % count($script);
+                }
+                $label = [1 => 'RUNNING', 2 => 'IDLE', 3 => 'FAULT'][$this->reg[0]];
+                $this->line(sprintf('[sim] state=%s good=%d reject=%d temp=%.1f', $label, $this->reg[1], $this->reg[2], $this->reg[3] / 10));
+            }
+        }
+
+        return self::SUCCESS;
+    }
+
+    /**
+     * Parse a Modbus TCP frame and build a response. Supports FC 0x03/0x04
+     * (read holding/input registers) and 0x01/0x02 (read coils/discretes).
+     */
+    private function handleRequest(string $data): ?string
+    {
+        if (strlen($data) < 12) {
+            return null;
+        }
+        $mbap = unpack('ntid/nprot/nlen/Cunit', substr($data, 0, 7));
+        $fc = ord($data[7]);
+        $body = substr($data, 8);
+        $req = unpack('naddr/nqty', $body);
+        $addr = $req['addr'];
+        $qty = max(1, $req['qty']);
+
+        if (in_array($fc, [0x03, 0x04], true)) {
+            $payload = '';
+            for ($i = 0; $i < $qty; $i++) {
+                $val = $this->reg[$addr + $i] ?? 0;
+                $payload .= pack('n', $val & 0xFFFF);
+            }
+            $pdu = chr($fc).chr(strlen($payload)).$payload;
+        } elseif (in_array($fc, [0x01, 0x02], true)) {
+            $byteCount = intdiv($qty + 7, 8);
+            $bits = 0;
+            for ($i = 0; $i < $qty; $i++) {
+                if (($this->reg[$addr + $i] ?? 0) > 0) {
+                    $bits |= (1 << $i);
+                }
+            }
+            $pdu = chr($fc).chr($byteCount).pack('C', $bits & 0xFF);
+        } else {
+            // Illegal function exception
+            $pdu = chr($fc | 0x80).chr(0x01);
+        }
+
+        $len = strlen($pdu) + 1; // + unit id
+
+        return pack('nnn', $mbap['tid'], 0, $len).chr($mbap['unit']).$pdu;
+    }
+}

backend/app/Events/Machine/WorkstationStateChanged.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace App\Events\Machine;
+
+use App\Models\Workstation;
+use App\Models\WorkstationState;
+use Illuminate\Foundation\Events\Dispatchable;
+use Illuminate\Queue\SerializesModels;
+
+class WorkstationStateChanged
+{
+    use Dispatchable;
+    use SerializesModels;
+
+    public function __construct(
+        public Workstation $workstation,
+        public ?string $from,
+        public string $to,
+        public WorkstationState $state,
+    ) {}
+}

backend/app/Http/Controllers/Api/V1/MachineGatewayController.php

@@ -0,0 +1,107 @@
+<?php
+
+namespace App\Http\Controllers\Api\V1;
+
+use App\Http\Controllers\Controller;
+use App\Models\MachineConnection;
+use App\Models\MachineTag;
+use App\Services\Machine\MachineSignalIngestor;
+use App\Services\Machine\RuntimeMonitor;
+use Illuminate\Http\JsonResponse;
+use Illuminate\Http\Request;
+
+/**
+ * Bridge endpoint for external protocol gateways (OPC UA sidecar, custom REST
+ * pushers). The gateway fetches its config (which nodes/tags to read), then
+ * posts normalized readings back here; each reading flows through the same
+ * MachineSignalIngestor as Modbus/MQTT. Posting also refreshes the runtime
+ * heartbeat so the UI knows the gateway is alive.
+ */
+class MachineGatewayController extends Controller
+{
+    public function __construct(
+        private readonly MachineSignalIngestor $ingestor,
+        private readonly RuntimeMonitor $runtime,
+    ) {}
+
+    /**
+     * Config the gateway needs to connect and subscribe.
+     */
+    public function config(MachineConnection $machineConnection): JsonResponse
+    {
+        $machineConnection->load(['opcuaConnection', 'activeTags']);
+        $opc = $machineConnection->opcuaConnection;
+
+        return response()->json([
+            'connection' => [
+                'id' => $machineConnection->id,
+                'name' => $machineConnection->name,
+                'protocol' => $machineConnection->protocol,
+                'is_active' => $machineConnection->is_active,
+            ],
+            'opcua' => $opc ? [
+                'endpoint_url' => $opc->endpoint_url,
+                'security_policy' => $opc->security_policy,
+                'security_mode' => $opc->security_mode,
+                'auth_mode' => $opc->auth_mode,
+                'username' => $opc->username,
+                'publishing_interval_ms' => $opc->publishing_interval_ms,
+            ] : null,
+            'tags' => $machineConnection->activeTags->map(fn (MachineTag $t) => [
+                'id' => $t->id,
+                'name' => $t->name,
+                'node_id' => $t->address,
+                'signal_type' => $t->signal_type,
+                'data_type' => $t->data_type,
+            ])->values(),
+        ]);
+    }
+
+    /**
+     * Receive normalized readings from the gateway and ingest them.
+     *
+     * Body: { readings: [ { tag_id?, node_id?, value, ts? }, ... ] }
+     */
+    public function ingest(Request $request, MachineConnection $machineConnection): JsonResponse
+    {
+        $data = $request->validate([
+            'readings' => ['required', 'array', 'min:1'],
+            'readings.*.tag_id' => ['nullable', 'integer'],
+            'readings.*.node_id' => ['nullable', 'string'],
+            'readings.*.value' => ['present'],
+            'readings.*.ts' => ['nullable', 'date'],
+        ]);
+
+        // Heartbeat: a posting gateway is, by definition, alive.
+        $this->runtime->heartbeat($machineConnection->protocol, $machineConnection->id);
+        $machineConnection->markConnected();
+
+        $tags = $machineConnection->activeTags()->get()->keyBy('id');
+        $byAddress = $machineConnection->activeTags()->get()->keyBy('address');
+
+        $accepted = 0;
+        foreach ($data['readings'] as $r) {
+            $tag = isset($r['tag_id']) ? $tags->get($r['tag_id']) : null;
+            $tag ??= isset($r['node_id']) ? $byAddress->get($r['node_id']) : null;
+            if (! $tag) {
+                continue;
+            }
+            $at = isset($r['ts']) ? \Illuminate\Support\Carbon::parse($r['ts']) : null;
+            $this->ingestor->ingest($tag, $r['value'], $at);
+            $machineConnection->increment('messages_received');
+            $accepted++;
+        }
+
+        return response()->json(['accepted' => $accepted]);
+    }
+
+    /**
+     * Standalone heartbeat (gateway connected but no readings to push yet).
+     */
+    public function heartbeat(MachineConnection $machineConnection): JsonResponse
+    {
+        $this->runtime->heartbeat($machineConnection->protocol, $machineConnection->id);
+
+        return response()->json(['ok' => true]);
+    }
+}

backend/app/Http/Controllers/Api/V1/MaterialLotController.php

@@ -112,7 +112,9 @@ public function forwardGenealogy(MaterialLot $materialLot): JsonResponse
      */
     public function backwardGenealogy(MaterialLot $materialLot): JsonResponse
     {
-        $sourceBatchId = data_get($materialLot->extra_data, 'source_batch_id');
+        // Prefer the formal FK column; fall back to the legacy extra_data hint.
+        $sourceBatchId = $materialLot->source_batch_id
+            ?? data_get($materialLot->extra_data, 'source_batch_id');
         $upstream = collect();
 
         if ($sourceBatchId) {