using UnityEngine; using MQTTnet; using MQTTnet.Client; using System; using System.Text; using System.IO; using System.Threading; using System.Threading.Tasks; using System.Collections.Concurrent; [Serializable] public class SensorData { public float hr; public float gsr; } public class ExperimentManager : MonoBehaviour { public static ExperimentManager Instance; [Header("MQTT Settings")] public string brokerIp = "127.0.0.1"; public int port = 1883; public string topic = "m5stick/state"; [Header("Watchdog Settings")] public float timeoutSeconds = 5f; // Czas w sekundach do włączenia alarmu private IMqttClient _mqttClient; private string _subjectFolderPath; private string _currentCsvFilePath; private ConcurrentQueue _dataQueue = new ConcurrentQueue(); // Zmienne do pilnowania czasu private float _timeSinceLastMessage = 0f; private bool _isDataTimeoutWarningShown = false; void Awake() { if (Instance == null) { Instance = this; DontDestroyOnLoad(gameObject); } else { Destroy(gameObject); } } public async void InitializeSession(string subjectId) { string dataPath = Path.Combine(Application.dataPath, "Data"); _subjectFolderPath = Path.Combine(dataPath, subjectId); Directory.CreateDirectory(_subjectFolderPath); Directory.CreateDirectory(Path.Combine(_subjectFolderPath, "Neutral")); Directory.CreateDirectory(Path.Combine(_subjectFolderPath, "Experimental")); Debug.Log($"[ExperimentManager] Folders created for subject: {subjectId}"); await ConnectToMqttAsync(); } public void SetPhase(string phaseName) { string fileName = $"{phaseName}_Data_{DateTime.Now:yyyyMMdd_HHmmss}.csv"; _currentCsvFilePath = Path.Combine(_subjectFolderPath, phaseName, fileName); if (!File.Exists(_currentCsvFilePath)) { File.WriteAllText(_currentCsvFilePath, "Timestamp;HR;GSR\n"); } Debug.Log($"[ExperimentManager] Now logging to: {_currentCsvFilePath}"); } public string GetCurrentFolderPath() { if (string.IsNullOrEmpty(_currentCsvFilePath)) return null; return Path.GetDirectoryName(_currentCsvFilePath); } public void StopLogging() { _currentCsvFilePath = null; Debug.Log("[ExperimentManager] Logging stopped. Data collection finished."); } private async Task ConnectToMqttAsync() { var factory = new MqttFactory(); _mqttClient = factory.CreateMqttClient(); var options = new MqttClientOptionsBuilder() .WithTcpServer(brokerIp, port) .Build(); _mqttClient.ApplicationMessageReceivedAsync += e => { string payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment); try { SensorData data = JsonUtility.FromJson(payload); _dataQueue.Enqueue(data); } catch (Exception ex) { Debug.LogWarning($"[MQTT] Parse error: {ex.Message}"); } return Task.CompletedTask; }; try { await _mqttClient.ConnectAsync(options, CancellationToken.None); var subOptions = new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(topic) .Build(); await _mqttClient.SubscribeAsync(subOptions, CancellationToken.None); Debug.Log("[MQTT] Connected successfully!"); } catch (Exception ex) { Debug.LogError($"[MQTT] Connection Error: {ex.Message}"); } } void Update() { bool receivedNewDataThisFrame = false; // Opróżnianie kolejki i zapis while (_dataQueue.TryDequeue(out SensorData newData)) { receivedNewDataThisFrame = true; if (!string.IsNullOrEmpty(_currentCsvFilePath)) { string timestamp = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"); string csvLine = $"{timestamp};{newData.hr};{newData.gsr}\n"; File.AppendAllText(_currentCsvFilePath, csvLine); } } // --- SYSTEM WATCHDOG (Pilnowanie transmisji) --- if (receivedNewDataThisFrame) { _timeSinceLastMessage = 0f; // Resetujemy stoper if (_isDataTimeoutWarningShown) { // Jeśli wcześniej wywaliliśmy błąd, a teraz dane wróciły, dajemy znać, że już jest OK Debug.Log("[ExperimentManager] Transmisja odzyskana! M5Stick znów nadaje."); _isDataTimeoutWarningShown = false; } } else if (_mqttClient != null && _mqttClient.IsConnected) { // Czas leci tylko wtedy, gdy Unity myśli, że MQTT jest połączone _timeSinceLastMessage += Time.deltaTime; if (_timeSinceLastMessage >= timeoutSeconds && !_isDataTimeoutWarningShown) { Debug.LogError($"[UWAGA MQTT!] Brak danych od {timeoutSeconds} sekund! Sprawdź M5Stick!"); _isDataTimeoutWarningShown = true; } } } private async void OnApplicationQuit() { if (_mqttClient != null && _mqttClient.IsConnected) { await _mqttClient.DisconnectAsync(); } } }