SimpleSceneIntegration.java

package ipos.project.DataModellntegration.SimpleSceneIntegration;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.protobuf.InvalidProtocolBufferException;
import ipos.models.SimpleScene;
import ipos.project.DataModellntegration.SimpleSceneIntegration.api.MqttRequestHandler;
import ipos.project.DataModellntegration.SimpleSceneIntegration.service.SimpleSceneTransformer;
import ipos.project.DataModellntegration.SimpleSceneIntegration.service.impl.ExternalPubServiceImpl;
import ipos.project.DataModellntegration.iPos_Datamodel.*;
import ipos.project.DataModellntegration.iPos_Datamodel.impl.PositionEventImpl;
import ipos.project.UseCaseController.Administration;
import ipos.project.UseCaseController.PositionMonitoring;
import ipos.project.mapper.ProtoJsonMap;
import org.apache.logging.log4j.LogManager;
import org.eclipse.emf.common.util.EList;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Component
public class SimpleSceneIntegration {
    // private static final Logger LOG = LoggerFactory.getLogger(SimpleSceneIntegration.class);
    private static org.apache.logging.log4j.Logger LOG = LogManager.getLogger();

    //@Autowired
    public static ExternalPubServiceImpl mqttService;

    public static void init(){
        Administration.initialize();
    }

    public static void init(String path_to_init_data_file){
        Administration.initialize(); // clears WorldModel
        handleMessageFile(path_to_init_data_file);
    }

    public static void handleMessageFile(String path_to_message_file) {
        File initFile = new File(path_to_message_file);
        try {
            for (String line : readLines(initFile)) {
                SimpleScene.IposConfigWrapper iposConfigWrapper = ProtoJsonMap.fromJson(line, SimpleScene.IposConfigWrapper.class);
                MqttRequestHandler.handleConfigWrapper(iposConfigWrapper);
            }
        }catch (InvalidProtocolBufferException e) {
            LOG.error("Error trying to read JSON into protobuf-objects: ");
            e.printStackTrace();
        }
    }

    /**
     * As opposed to the Stream<String>-approach using this approach reads the lines in order
     * @param initFile
     * @return
     */
    private static List<String> readLines(File initFile) {
        List<String> lines = new ArrayList<>();
        try(BufferedReader br = new BufferedReader(new FileReader(initFile))) { // br is automatically closed at the end of try-scope
            for (String line; (line = br.readLine()) != null; ){
                lines.add(line);
            }
        } catch (IOException e) {
            LOG.error("Error while reading the text file: ");
            e.printStackTrace();
        }
        return lines;
    }

    @Autowired
    public SimpleSceneIntegration(ExternalPubServiceImpl mqttService_){
        mqttService = mqttService_;
    }

    /*
    @JmsListener(destination = "/positions", containerFactory = "jmsListenFactory")
    public static void receiveMessage(IposPosition pos) {
        LOG.trace("Data integration get: " + pos);
        // we can translate a class into a string using several methods: 1) `.toString()` 2) `JsonFormat` in `ProtoJsonMap`
        String jsonPos = ProtoJsonMap.toJson(pos);
        if (jsonPos != null) {
            this.mqttService.publish("test51/subscribe/positions", jsonPos, 0, false);
        }
    }*/

    public static void receiveMessage(DataStorageQueryResponse dsQueryResponse, String trackingTaskId, String serializationType){
        SimpleScene.IposQueryResponse queryResponse_proto = SimpleSceneTransformer.queryResp_Internal2SScene(dsQueryResponse, trackingTaskId);
        MqttMessage mqttMessage = mqttService.createMqttMsg(queryResponse_proto, 0, false);
        LOG.info("INDFRO: Publishing DataStorageQueryResponse on topic " + trackingTaskId + ": " + ProtoJsonMap.toJson(queryResponse_proto) + "; serializationType: " + serializationType);
        publishRespectingSerializationType(trackingTaskId, serializationType, ProtoJsonMap.toJson(queryResponse_proto), mqttMessage);
    }

    public static void receiveMessage(PositionEvent positionEvent, String monitoringTaskId, String notificationType, String serializationType) {
        //String jsonPos = ProtoJsonMap.toJson(positionEvent); // we can translate a class into a string using several methods: 1) `.toString()` 2) `JsonFormat` in `ProtoJsonMap`
        if (positionEvent != null) {

            // verwenden: DataServices.create-Methoden verwenden. Für ZoneDeskriptoren neue create-Funktionen schreiben
            // PositionEvent posEvent_copy = duplicatePositionEvent(positionEvent);
            // scale_position_frontend_app(posEvent_copy, 15);
            SimpleScene.IposPositionEvent posEvent_proto = SimpleSceneTransformer.posEvent_internal2Proto(positionEvent, notificationType);
            MqttMessage mqttMessage = mqttService.createMqttMsg(posEvent_proto, 0, false);
            logging_sdf(monitoringTaskId, serializationType, posEvent_proto);
            logging_indfro(positionEvent, monitoringTaskId, serializationType);

            publishRespectingSerializationType(monitoringTaskId, serializationType, ProtoJsonMap.toJson(posEvent_proto), mqttMessage);

        }else{
            LOG.warn("Warning: SimpleScene: Received empty PositionEvent");
        }
    }

    private static PositionEvent duplicatePositionEvent(PositionEvent positionEvent) {
        PositionEventImpl posEvent_copy = (PositionEventImpl) positionEvent;
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            posEvent_copy = objectMapper.readValue(objectMapper.writeValueAsString(positionEvent), PositionEventImpl.class);
        } catch (JsonProcessingException e) {
            LOG.warn("could not duplicate PositionEvent");
            e.printStackTrace();
        }
        return posEvent_copy;
    }

    private static void logging_indfro(PositionEvent positionEvent, String monitoringTaskId, String serializationType) {
        LOG.info("INDFRO: Publishing PositionEvent on topic " + monitoringTaskId + ": ZoneDescriptors:" + logZoneDescr(positionEvent.getZonedescriptors()) + "; serializationType: " + serializationType);
    }

    private static String logZoneDescr(EList<ZoneDescriptor> zonedescriptors) {
        String logString = "";
        for (ZoneDescriptor zoneDescriptor : zonedescriptors){
            logString += "zoneId: " + zoneDescriptor.getZoneId() + "; ";
            logString += "notificationType: " + zoneDescriptor.getNotificationType() + "    ";
        }
        logString += "_______";
        return logString;
    }

    private static void publishRespectingSerializationType(String topic, String serializationType, String jsonString, MqttMessage mqttMessage) {
        // publishes protobuf over MQTT
        if (PositionMonitoring.PROTOBUF_SERIALIZATION_TYPE.equals(serializationType)){
            mqttService.publish(topic, mqttMessage);
            LOG.info("publishing protobuf mqttMessage. JSON-representation:" + jsonString + " on topic: " + topic);
        }

        // publishes JSON over MQTT
        if (PositionMonitoring.JSON_SERIALIZATION_TYPE.equals(serializationType)){
            String fakeTopic = "ipos/client/position";
            mqttService.publish(fakeTopic, jsonString, 0, false);
            LOG.info("publishing JSON mqttMessage:" + jsonString + " on topic: " + fakeTopic);
        }
    }

    private static PositionEvent scale_position_frontend_app(PositionEvent internalPositionEvent, int factor) {
        Point3D point = (Point3D) internalPositionEvent.getPlacing().getPosition().getPoint();
        point.setX(factor*point.getX());
        point.setY(factor*point.getY());
        point.setZ(factor*point.getZ());
        return internalPositionEvent;
    }

    private static void logging_sdf(String monitoringTaskId, String serializationType, SimpleScene.IposPositionEvent posEvent_proto) {
        for (SimpleScene.IposObject obj : posEvent_proto.getObjectsList()){
            String agentId = obj.getId();
            String sensorId = obj.getSensorId();
            String sensorType = obj.getSensorType();
            String timestamp = obj.getLastPosUpdate();
            String topic = monitoringTaskId;
            float accuracy = obj.getPosition().getAccuracy();
            String positionString = toPositionString(obj.getPosition());
            LOG.info("SDF-PUB: Publishing: sensorType: " + sensorType + "; timestamp: " + timestamp + "; accuracy: " + accuracy + "position: " + positionString + "; sensorId: " + sensorId + "; agentId: " + agentId + "; topic: " + monitoringTaskId + "; serializationType: " + serializationType);
            LOG.info("SDF: ");
            LOG.info("SDF: ");
            LOG.info("SDF: ");
        }
    }

    private static String toPositionString(SimpleScene.IposPosition position) {
        String posString = "(";
        posString += "x: " + position.getPoint().getX();
        posString += "y: " + position.getPoint().getY();
        posString += "z: " + position.getPoint().getZ();
        posString += ")";
        return posString;
    }

}