<commit_before>/**
 * Copyright (C) 2004
 * by 52 North Initiative for Geospatial Open Source Software GmbH
 *
 * Contact: <NAME>
 * 52 North Initiative for Geospatial Open Source Software GmbH
 * Martin-Luther-King-Weg 24
 * 48155 Muenster, Germany
 * <EMAIL>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.n52.cario.gep.inbound;

import java.io.IOException;
import java.io.InputStream;
import java.text.DateFormat;
import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.n52.cario.gep.commons.definitions.TrackDefinition;
import org.n52.cario.gep.commons.inbound.AbstractInboundAdapter;
import org.n52.cario.gep.commons.json.JsonUtil;
import org.n52.cario.parser.sensor.SensorImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.esri.ges.adapter.AdapterDefinition;
import com.esri.ges.core.component.ComponentException;
import com.esri.ges.core.geoevent.FieldException;
import com.esri.ges.core.geoevent.GeoEvent;
import com.esri.ges.core.geoevent.GeoEventDefinition;
import com.esri.ges.messaging.MessagingException;
import com.esri.ges.spatial.Geometry;
import com.fasterxml.jackson.databind.util.ISO8601DateFormat;

public class TrackInboundAdapter extends AbstractInboundAdapter {

	private Queue<GeoEvent> eventQueue = new ConcurrentLinkedQueue<GeoEvent>();

	private static DateFormat isoDateTimeFormat = new ISO8601DateFormat();

	private static final Logger logger = LoggerFactory
			.getLogger(TrackInboundAdapter.class);

	private static final String FEATURES_KEY = "features";
	private static final String PROPERTIES_KEY = "properties";
	private static final String PHENOMENONS_KEY = "phenomenons";
	private static final String VALUE_KEY = "value";
	private static final String GEOMETRY_KEY = "geometry";
	private static final String TIME_KEY = "time";
	private static final String ID_KEY = "id";
	private static final String TYPE_KEY = "type";

	public static final String ENVIROCAR_TRACK_VIEW_LINK = "https://envirocar.org/route.php?id=";
	
	/**
	 * Mapping cancidates in the enviroCar datastructure
	 * to fields in the GeoEventDefinition
	 */
	private static Map<String[], String> phenomenonFields;
	
	static {
		phenomenonFields = new HashMap<String[], String>();
		phenomenonFields.put(new String[] {"Consumption"}, TrackDefinition.CONSUMPTION_KEY);
		phenomenonFields.put(new String[] {"CO2"}, TrackDefinition.CO2_KEY);
		phenomenonFields.put(new String[] {"Speed"}, TrackDefinition.SPEED_KEY);
		phenomenonFields.put(new String[] {"GPS Bearing"}, TrackDefinition.BEARING_KEY);
	}

	public TrackInboundAdapter(AdapterDefinition definition)
			throws ComponentException {
		super(definition);
	}

	@Override
	protected synchronized boolean allDataConsumed() {
		return this.eventQueue.isEmpty();
	}

	@Override
	protected synchronized GeoEvent createGeoEvent(InputStream input) {
		try {
			Map<?, ?> json = JsonUtil.createJson(input);
			appendFeaturesToQueue(json);
			return this.eventQueue.poll();
		} catch (RuntimeException e) {
			logger.warn(e.getMessage(), e);
			return null;
		} catch (IOException e) {
			logger.warn(e.getMessage(), e);
			return null;
		}
	}

	@Override
	protected synchronized GeoEvent createNextRemainingGeoEvent() {
		return this.eventQueue.poll();
	}

	private void appendFeaturesToQueue(Map<?, ?> json) {
		if (!json.containsKey(FEATURES_KEY)
				|| !json.containsKey(PROPERTIES_KEY)
				|| !json.containsKey(TrackDefinition.TYPE_KEY))
			return;
		
		String type = (String) json.get(TrackDefinition.TYPE_KEY);
		if (!type.equals("FeatureCollection"))
			return;

		List<?> features = (List<?>) json.get(FEATURES_KEY);
		Map<?, ?> props = (Map<?, ?>) json.get(PROPERTIES_KEY);
		
		String sensor = parseSensor(props);
		
		String id = parseId(props);

		for (Object feature : features) {
			if (feature instanceof Map<?, ?>) {
				Map<?, ?> mapFeature = (Map<?, ?>) feature;
				try {
					GeoEvent geoEvent = prepareGeoEvent(sensor, id);
					fillGeoEvent(geoEvent, mapFeature);
					logger.info("Adding {} to eventQueue tail.", geoEvent);
					this.eventQueue.add(geoEvent);
				} catch (MessagingException e) {<commit_msg>add logging statements for debugging purposes<commit_after>/**
 * Copyright (C) 2004
 * by 52 North Initiative for Geospatial Open Source Software GmbH
 *
 * Contact: <NAME>
 * 52 North Initiative for Geospatial Open Source Software GmbH
 * Martin-Luther-King-Weg 24
 * 48155 Muenster, Germany
 * <EMAIL>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.n52.cario.gep.inbound;

import java.io.IOException;
import java.io.InputStream;
import java.text.DateFormat;
import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.n52.cario.gep.commons.definitions.TrackDefinition;
import org.n52.cario.gep.commons.inbound.AbstractInboundAdapter;
import org.n52.cario.gep.commons.json.JsonUtil;
import org.n52.cario.parser.sensor.SensorImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.esri.ges.adapter.AdapterDefinition;
import com.esri.ges.core.component.ComponentException;
import com.esri.ges.core.geoevent.FieldException;
import com.esri.ges.core.geoevent.GeoEvent;
import com.esri.ges.core.geoevent.GeoEventDefinition;
import com.esri.ges.messaging.MessagingException;
import com.esri.ges.spatial.Geometry;
import com.fasterxml.jackson.databind.util.ISO8601DateFormat;

public class TrackInboundAdapter extends AbstractInboundAdapter {

	private Queue<GeoEvent> eventQueue = new ConcurrentLinkedQueue<GeoEvent>();

	private static DateFormat isoDateTimeFormat = new ISO8601DateFormat();

	private static final Logger logger = LoggerFactory
			.getLogger(TrackInboundAdapter.class);

	private static final String FEATURES_KEY = "features";
	private static final String PROPERTIES_KEY = "properties";
	private static final String PHENOMENONS_KEY = "phenomenons";
	private static final String VALUE_KEY = "value";
	private static final String GEOMETRY_KEY = "geometry";
	private static final String TIME_KEY = "time";
	private static final String ID_KEY = "id";
	private static final String TYPE_KEY = "type";

	public static final String ENVIROCAR_TRACK_VIEW_LINK = "https://envirocar.org/route.php?id=";
	
	/**
	 * Mapping cancidates in the enviroCar datastructure
	 * to fields in the GeoEventDefinition
	 */
	private static Map<String[], String> phenomenonFields;
	
	static {
		phenomenonFields = new HashMap<String[], String>();
		phenomenonFields.put(new String[] {"Consumption"}, TrackDefinition.CONSUMPTION_KEY);
		phenomenonFields.put(new String[] {"CO2"}, TrackDefinition.CO2_KEY);
		phenomenonFields.put(new String[] {"Speed"}, TrackDefinition.SPEED_KEY);
		phenomenonFields.put(new String[] {"GPS Bearing"}, TrackDefinition.BEARING_KEY);
	}

	public TrackInboundAdapter(AdapterDefinition definition)
			throws ComponentException {
		super(definition);
	}

	@Override
	protected synchronized boolean allDataConsumed() {
		return this.eventQueue.isEmpty();
	}

	@Override
	protected synchronized GeoEvent createGeoEvent(InputStream input) {
		try {
			logger.info("Queue size before: {}", this.eventQueue.size());
			Map<?, ?> json = JsonUtil.createJson(input);
			appendFeaturesToQueue(json);
			logger.info("Queue size after: {}", this.eventQueue.size());
			return this.eventQueue.poll();
		} catch (RuntimeException e) {
			logger.warn(e.getMessage(), e);
			return null;
		} catch (IOException e) {
			logger.warn(e.getMessage(), e);
			return null;
		}
	}

	@Override
	protected synchronized GeoEvent createNextRemainingGeoEvent() {
		return this.eventQueue.poll();
	}

	private void appendFeaturesToQueue(Map<?, ?> json) {
		if (!json.containsKey(FEATURES_KEY)
				|| !json.containsKey(PROPERTIES_KEY)
				|| !json.containsKey(TrackDefinition.TYPE_KEY))
			return;
		
		String type = (String) json.get(TrackDefinition.TYPE_KEY);
		if (!type.equals("FeatureCollection"))
			return;

		List<?> features = (List<?>) json.get(FEATURES_KEY);
		logger.info("# of features in track's JSON: {}", features.size());
		Map<?, ?> props = (Map<?, ?>) json.get(PROPERTIES_KEY);
		
		String sensor = parseSensor(props);
		
		String id = parseId(props);

		for (Object feature : features) {
			if (feature instanceof Map<?, ?>) {
				Map<?, ?> mapFeature = (Map<?, ?>) feature;
				try {
					GeoEvent geoEvent = prepareGeoEvent(sensor, id);
					fillGeoEvent(geoEvent, mapFeature);
					logger.info("Adding {} to eventQueue tail.", geoEvent);
					this.eventQueue.add(geoEvent);
				} catch (MessagingException e) {
{'fraction_non_alphanumeric': 0.11211997551520098, 'fraction_numerical': 0.008365639665374413, 'mean_word_length': 4.912545235223161, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4745875', 'n_tokens_mistral': 3233, 'n_tokens_neox': 2940, 'n_words': 778}