Feature/frb 145

src/main/java/com/factoreal/backend/dto/SensorDto.java

@@ -15,6 +15,7 @@ public class SensorDto { // BE -> FE 용 DTO
     private String equipId;
     private Double sensorThres; // 임계치
     private Double allowVal;     // 허용치
+    private Integer isZone;
 
     public static SensorDto fromEntity(Sensor sensor) {
         if (sensor == null) return null;
@@ -26,6 +27,7 @@ public static SensorDto fromEntity(Sensor sensor) {
                 .equipId(sensor.getEquip().getEquipId())
                 .sensorThres(sensor.getSensorThres())
                 .allowVal(sensor.getAllowVal())
+                .isZone(sensor.getIsZone())
                 .build();
     }
 }

src/main/java/com/factoreal/backend/dto/SensorKafkaDto.java

@@ -9,12 +9,16 @@
 @Setter
 @NoArgsConstructor
 @AllArgsConstructor
-// Kafka mapper용 DTO
+/**
+ * Kafka Sensor 데이터 mapper용 DTO
+ */
 public class SensorKafkaDto {
-    private String zoneId;
-    private String equipId;
-    private String sensorId;
-    private String sensorType;
-    private Double val;  // 측정값 단위
-    private String time; // 센서 생성시간
+    private String zoneId;     // 공간 ID
+    private String equipId;    // 설비 ID
+    private String sensorId;   // 센서 ID
+    private String sensorType; // 센서 타입
+    private Double val;        // 센서 측정값
+    private String Category;   // 센서 카테고리 {ENVIRONMENT, EQUIPMENT}
+    private String time;       // 센서 탐지 시간
+    private int dangerLevel;   // 위험도
 }

src/main/java/com/factoreal/backend/entity/Sensor.java

@@ -1,6 +1,6 @@
 package com.factoreal.backend.entity;
 
-import com.factoreal.backend.strategy.enums.SensorType;
+import com.factoreal.backend.kafka.strategy.enums.SensorType;
 import jakarta.persistence.*;
 import lombok.*;
 import java.time.LocalDateTime;

src/main/java/com/factoreal/backend/consumer/kafka/KafkaConsumer.java → src/main/java/com/factoreal/backend/kafka/KafkaConsumerD.java

@@ -1,19 +1,17 @@
-package com.factoreal.backend.consumer.kafka;
+package com.factoreal.backend.kafka;
 
 import com.factoreal.backend.dto.SensorKafkaDto;
 import com.factoreal.backend.dto.abnormalLog.LogType;
 import com.factoreal.backend.entity.AbnormalLog;
 import com.factoreal.backend.dto.SystemLogDto;
-import com.factoreal.backend.entity.Zone;
 import com.factoreal.backend.sender.WebSocketSender;
 import com.factoreal.backend.service.ZoneService;
-import com.factoreal.backend.entity.AbnormalLog;
 import com.factoreal.backend.service.AbnormalLogService;
-import com.factoreal.backend.strategy.NotificationStrategy;
-import com.factoreal.backend.strategy.NotificationStrategyFactory;
-import com.factoreal.backend.strategy.enums.AlarmEventDto;
-import com.factoreal.backend.strategy.enums.RiskLevel;
-import com.factoreal.backend.strategy.enums.SensorType;
+import com.factoreal.backend.kafka.strategy.alarmList.NotificationStrategy;
+import com.factoreal.backend.kafka.strategy.NotificationStrategyFactory;
+import com.factoreal.backend.kafka.strategy.enums.AlarmEventDto;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.SensorType;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
@@ -22,7 +20,6 @@
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import com.factoreal.backend.service.SensorService;
@@ -31,19 +28,15 @@
 import java.time.ZoneId;
 
 import java.time.Instant;
-import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Stream;
-import java.util.Objects;
 
 @Service
 @Slf4j
 @RequiredArgsConstructor
-public class KafkaConsumer {
+public class KafkaConsumerD {
 
     private final ObjectMapper objectMapper;
     private final WebSocketSender webSocketSender;
@@ -69,8 +62,8 @@ public class KafkaConsumer {
 
     // @KafkaListener(topics = {"EQUIPMENT", "ENVIRONMENT"}, groupId =
     // "monitory-consumer-group-1")
-    @KafkaListener(topics = { "EQUIPMENT",
-            "ENVIRONMENT" }, groupId = "${spring.kafka.consumer.group-id:danger-alert-group}")
+//    @KafkaListener(topics = { "EQUIPMENT",
+//            "ENVIRONMENT" }, groupId = "${spring.kafka.consumer.group-id:danger-alert-group}")
     public void consume(String message) {
 
         log.info("💡수신한 Kafka 메시지 : " + message);
@@ -87,6 +80,7 @@ public void consume(String message) {
                 // 비동기 ES 저장
                 // #################################
                 saveToElasticsearch(dto);
+
                 log.info("▶︎ 위험도 감지 start");
                 int dangerLevel = getDangerLevel(dto.getSensorType(), dto.getVal());
                 log.info("⚠️ 위험도 {} 센서 타입 : {} 감지됨. Zone: {}", dangerLevel, dto.getSensorType(), dto.getZoneId());

src/main/java/com/factoreal/backend/kafka/consumer/KafkaConsumer.java

@@ -0,0 +1,52 @@
+package com.factoreal.backend.kafka.consumer;
+
+import com.factoreal.backend.dto.SensorKafkaDto;
+import com.factoreal.backend.kafka.processor.SensorEventProcessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * KafkaConsumer 클래스는 Kafka 토픽으로부터 메시지를 수신하고,
+ * 해당 메시지를 SensorEventProcessor.java 전달
+ */
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class KafkaConsumer {
+
+    private final ObjectMapper objectMapper;
+    private final SensorEventProcessor sensorEventProcessor;
+
+    // 설비 센서 관련 Kafka 메시지 처리
+    // Todo : 설비 머신러닝 끝나고 수정 예정
+//    @KafkaListener(topics = "EQUIPMENT", groupId = "equipment-consumer-group")
+    public void consumeEquipment(String message) {
+        log.info("📩 [EQUIPMENT] Kafka 메시지 수신: {}", message);
+        handleMessage(message, "EQUIPMENT");
+    }
+
+    // 공간 센서 관련 Kafka 메시지 처리
+    @KafkaListener(topics = "ENVIRONMENT", groupId = "environment-consumer-group")
+    public void consumeEnvironment(String message) {
+        log.info("📩 [ENVIRONMENT] Kafka 메시지 수신: {}", message);
+        handleMessage(message, "ENVIRONMENT");
+    }
+
+    // 공통 메시지 파싱 및 처리 전달
+    private void handleMessage(String message, String topic) {
+        try {
+            SensorKafkaDto dto = objectMapper.readValue(message, SensorKafkaDto.class);
+            log.info("✅ Kafka 메시지 파싱 완료: sensorId={}, zoneId={}, val={}",
+                    dto.getSensorId(), dto.getZoneId(), dto.getVal());
+            sensorEventProcessor.process(dto, topic);
+            log.info("✅ Kafka 메시지 처리 위임 완료: topic={}", topic);
+        } catch (Exception e) {
+            log.error("❌ Kafka 메시지 파싱 또는 처리 실패: {}", message, e);
+        }
+    }
+
+}

src/main/java/com/factoreal/backend/kafka/processor/SensorEventProcessor.java

@@ -0,0 +1,113 @@
+package com.factoreal.backend.kafka.processor;
+
+import com.factoreal.backend.dto.SensorKafkaDto;
+import com.factoreal.backend.dto.abnormalLog.LogType;
+import com.factoreal.backend.entity.AbnormalLog;
+import com.factoreal.backend.sender.WebSocketSender;
+import com.factoreal.backend.service.AbnormalLogService;
+import com.factoreal.backend.service.AlarmEventService;
+import com.factoreal.backend.service.AutoControlService;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.SensorType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * SensorEventProcessor 클래스는 Kafka로부터 전달받은 센서 데이터를 분석 및 처리하는 클래스입니다.
+ */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class SensorEventProcessor {
+
+    private final AutoControlService autoControlService;
+    private final AbnormalLogService abnormalLogService;
+    private final AlarmEventService alarmEventService;
+    private final WebSocketSender webSocketSender;
+
+    /**
+     * 센서 Kafka 메시지 처리
+     * @param dto 센서 데이터
+     * @param topic Kafka 토픽명 (EQUIPMENT, ENVIRONMENT)
+     */
+    public void process(SensorKafkaDto dto, String topic) {
+        try {
+
+            // 유효성 검사: zoneId와 sensorId는 필수
+            if (dto.getZoneId() == null || dto.getZoneId().isBlank()) {
+                log.warn("⚠️ 유효하지 않은 zoneId: null 또는 빈 문자열");
+                return;
+            }
+            if (dto.getSensorId() == null || dto.getSensorId().isBlank()) {
+                log.warn("⚠️ 유효하지 않은 sensorId: null 또는 빈 문자열");
+                return;
+            }
+
+            // ENVIRONMENT 토픽인 경우에만 아래 처리 로직 수행
+            if ("ENVIRONMENT".equalsIgnoreCase(topic)) {
+                if (dto.getEquipId() == null || !dto.getEquipId().equals(dto.getZoneId())) {
+                    log.warn("⚠️ ENVIRONMENT 토픽이지만 equipId와 zoneId가 일치하지 않음: equipId={}, zoneId={}",
+                            dto.getEquipId(), dto.getZoneId());
+                    return;
+                }
+
+                // 위험도 계산
+                int dangerLevel = dto.getDangerLevel();
+                SensorType sensorType = SensorType.getSensorType(dto.getSensorType());
+                RiskLevel riskLevel = RiskLevel.fromPriority(dangerLevel);
+                LogType logType = topicToLogType(topic);
+
+                // 자동 제어 메시지 판단 (Todo - 진행중)
+                autoControlService.evaluate(dto, dangerLevel);
+
+                // 이상 로그 저장
+                AbnormalLog abnLog = abnormalLogService.saveAbnormalLogFromKafkaDto(
+                        dto, sensorType, riskLevel, logType);
+
+
+                // 읽지 않은 알림 수 조회
+                Long count = abnormalLogService.readRequired();
+
+                // WebSocket 알림 전송
+                // 1. 히트맵 전송
+                webSocketSender.sendDangerLevel(dto.getZoneId(), dto.getSensorType(), dangerLevel);
+                // 2. 위험 알림 전송 -> 위험도별 Websocket + wearable + Slack(SMS 대체)
+                // Todo : (As-is) 전략 기반 startAlarm() 메서드 담당자 확인 필요
+//                webSocketSender.sendDangerAlarm(abnLog.toAlarmEventDto());
+                alarmEventService.startAlarm(dto,abnLog,dangerLevel);
+                // 3. 읽지 않은 수 전송
+                webSocketSender.sendUnreadCount(count);
+
+
+
+                log.info("✅ 센서 이벤트 처리 완료: sensorId={}, zoneId={}, level={} ({} topic)",
+                        dto.getSensorId(), dto.getZoneId(), dangerLevel, topic);
+            }
+        } catch (Exception e) {
+            log.error("❌ 센서 이벤트 처리 실패: sensorId={}, zoneId={}", dto.getSensorId(), dto.getZoneId(), e);
+        }
+    }
+
+    // 공간에 위험도 분기 로직
+    // Todo : Flink에서 적용으로 변경되어 삭제 예정
+    private int getDangerLevel(String sensorType, double val) {
+        switch (sensorType.toLowerCase()) {
+            case "temp": return val > 40 || val < -35 ? 2 : (val > 30 || val < 25 ? 1 : 0);
+            case "humid": return val >= 80 ? 2 : (val >= 60 ? 1 : 0);
+            case "vibration": return val > 7.1 ? 2 : (val > 2.8 ? 1 : 0);
+            case "current": return val >= 30 ? 2 : (val >= 7 ? 1 : 0);
+            case "dust": return val >= 150 ? 2 : (val >= 75 ? 1 : 0);
+            default: return 0;
+        }
+    }
+
+    // topic enum 변경하기
+    private LogType topicToLogType(String topic) {
+        return switch (topic.toUpperCase()) {
+            case "EQUIPMENT" -> LogType.Equip;
+            case "ENVIRONMENT" -> LogType.Sensor;
+            default -> throw new IllegalArgumentException("지원하지 않는 Kafka 토픽: " + topic);
+        };
+    }
+}

src/main/java/com/factoreal/backend/strategy/NotificationStrategyFactory.java → src/main/java/com/factoreal/backend/kafka/strategy/NotificationStrategyFactory.java

@@ -1,6 +1,7 @@
-package com.factoreal.backend.strategy;
+package com.factoreal.backend.kafka.strategy;
 
-import com.factoreal.backend.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.alarmList.NotificationStrategy;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
 import org.springframework.stereotype.Component;
 
 import java.util.List;

src/main/java/com/factoreal/backend/strategy/AppPushNotificationStrategy.java → src/main/java/com/factoreal/backend/kafka/strategy/alarmList/AppPushNotificationStrategy.java

@@ -1,15 +1,15 @@
-package com.factoreal.backend.strategy;
+package com.factoreal.backend.kafka.strategy.alarmList;
 
-import com.factoreal.backend.strategy.enums.AlarmEventDto;
-import com.factoreal.backend.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.AlarmEventDto;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component("APP")
 public class AppPushNotificationStrategy implements NotificationStrategy {
 
-
+    // TODO FCM 전송 로직
     @Override
     public void send(AlarmEventDto alarmEventDto) {
         log.info("📲 App Push Notification Strategy.");

src/main/java/com/factoreal/backend/kafka/strategy/alarmList/NotificationStrategy.java

@@ -0,0 +1,12 @@
+package com.factoreal.backend.kafka.strategy.alarmList;
+
+import com.factoreal.backend.kafka.strategy.enums.AlarmEventDto;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
+
+public interface NotificationStrategy {
+    void send(AlarmEventDto alarmEventDto);
+    // 이 인터페이스를 상속받는 객체가 동작할 최소 위험 레벨을 설정
+    // Kafka로 받아온 센서 데이터와 비교하기 위함.
+    // 센서의 Danger Level보다 같거나 낮은 Strategy들이 반환되기 위함.
+    RiskLevel getSupportedLevel();
+}

src/main/java/com/factoreal/backend/strategy/SmsNotificationStrategy.java → src/main/java/com/factoreal/backend/kafka/strategy/alarmList/SmsNotificationStrategy.java

@@ -1,8 +1,8 @@
-package com.factoreal.backend.strategy;
+package com.factoreal.backend.kafka.strategy.alarmList;
 
-import com.factoreal.backend.strategy.enums.AlarmEventDto;
+import com.factoreal.backend.kafka.strategy.enums.AlarmEventDto;
 import com.factoreal.backend.entity.Worker;
-import com.factoreal.backend.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
 import com.factoreal.backend.repository.WorkerRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -22,6 +22,7 @@ public class SmsNotificationStrategy implements NotificationStrategy {
     private final WorkerRepository workerRepository;
 
     private static final String userId = "alarm-test";
+    // TODO Seoul 리전에 SMS 지원이 안되기에 Slack으로 변경
     @Override
     public void send(AlarmEventDto alarmEventDto) {
         log.info("📬 SMS Notification Strategy.");

src/main/java/com/factoreal/backend/strategy/WebSocketNotificationStrategy.java → src/main/java/com/factoreal/backend/kafka/strategy/alarmList/WebSocketNotificationStrategy.java

@@ -1,20 +1,21 @@
-package com.factoreal.backend.strategy;
+package com.factoreal.backend.kafka.strategy.alarmList;
 
 import com.factoreal.backend.sender.WebSocketSender;
-import com.factoreal.backend.strategy.enums.AlarmEventDto;
-import com.factoreal.backend.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.AlarmEventDto;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 @RequiredArgsConstructor
 @Component
 @Slf4j
-public class WebSocketNotificationStrategy implements  NotificationStrategy{
+public class WebSocketNotificationStrategy implements NotificationStrategy {
     // SimpMessagingTemplate은 WebSocketConfig.java에 EnableWebSocketMessageBroker 어노테이션에 의해 빈이 등록됨.
     private final WebSocketSender webSocketSender;
 
     private static final String userId = "alarm-test";
+
     @Override
     public void send(AlarmEventDto alarmEventDto) {
         log.info("🌐WebSocket Notification Strategy");

src/main/java/com/factoreal/backend/strategy/DefaultRiskMessageProvider.java → src/main/java/com/factoreal/backend/kafka/strategy/alarmMessage/DefaultRiskMessageProvider.java

@@ -1,7 +1,7 @@
-package com.factoreal.backend.strategy;
+package com.factoreal.backend.kafka.strategy.alarmMessage;
 
-import com.factoreal.backend.strategy.enums.RiskLevel;
-import com.factoreal.backend.strategy.enums.SensorType;
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.SensorType;
 import org.springframework.stereotype.Component;
 
 @Component

src/main/java/com/factoreal/backend/kafka/strategy/alarmMessage/RiskMessageProvider.java

@@ -0,0 +1,8 @@
+package com.factoreal.backend.kafka.strategy.alarmMessage;
+
+import com.factoreal.backend.kafka.strategy.enums.RiskLevel;
+import com.factoreal.backend.kafka.strategy.enums.SensorType;
+
+public interface RiskMessageProvider {
+    String getMessage(SensorType sensorType, RiskLevel riskLevel);
+}