<commit_before>package uk.ac.abdn.iotstreams.csparql;

import java.time.ZonedDateTime;
import java.util.function.Consumer;
import java.util.function.Function;

import uk.ac.abdn.iotstreams.util.IotStreamsException;
import uk.ac.abdn.iotstreams.util.Logging;

import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;

import eu.larkc.csparql.cep.api.RdfQuadruple;
import eu.larkc.csparql.cep.api.RdfStream;
import eu.larkc.csparql.common.config.Config;
import eu.larkc.csparql.core.engine.CsparqlEngineImpl;

/**
 * 
 * @author nhc
 *
 * An IotStreamsEngine is a specific CsparqlEngine with methods
 * and queries for this project.
 */
public final class IotStreamsEngine
    extends CsparqlEngineImpl 
    implements Function<ZonedDateTime, Consumer<Model>> {
    
    /** This engine's sole stream */
    private final RdfStream rdfStream = new RdfStream("http://iotstreams");
    
    /** Object to which all inferences will be passed */
    private final Consumer<Model> persistentModel;
    
    /** Number of quadruples put on stream so far */
    private long numQuads = 0;
    
    /**
     * Verifies that C-SPARQL is configured for using recorded data,
     * then returns an IotStreamsEngine.
     * @param inferredDataConsumer Object to which all inferences will be passed
     * @return A fresh instance of IotStreamsEngine
     */
    public static IotStreamsEngine forRecordedData(final Consumer<Model> inferredDataConsumer) {
        if (Config.INSTANCE.isEsperUsingExternalTimestamp()) {
            return new IotStreamsEngine(inferredDataConsumer);
        } else {
            throw IotStreamsException.configurationError(
                    "To use recorded data, csparql.properties must contain the line "
                    + "'esper.externaltime.enabled=true'");
        }
    }
    
    /**
     * Verifies that C-SPARQL is configured for using live data,
     * then returns an IotStreamsEngine.
     * @param inferredDataConsumer Object to which all inferences will be passed
     * @return A fresh instance of IotStreamsEngine
     */
    public static IotStreamsEngine forLiveData(final Consumer<Model> inferredDataConsumer) {
        if (Config.INSTANCE.isEsperUsingExternalTimestamp()) {
            throw IotStreamsException.configurationError(
                    "When using live data, csparql.properties must not contain the line "
                    + "'esper.externaltime.enabled=true'");
        } else {
            return new IotStreamsEngine(inferredDataConsumer);
        }
    }
    
    /**
     * Initializes this engine.
     * @param persistentModel Object to which all inferences will be passed
     */
    private IotStreamsEngine(final Consumer<Model> persistentModel) {
        this.persistentModel = persistentModel;
        this.initialize();
        this.registerStream(this.rdfStream);
        new Configurator(this, this.persistentModel);
    }
    
    /**
     * Logs a status message
     */
    public void log() {
        Logging.info(String.format("%d quadruples put on stream", this.numQuads));
    }
    
    /**
     * To a Jena Model for a given timestamp, use engine.apply(t).accept(model)
     * This is equivalent to engine.add(t, m)
     */
    @Override
    public Consumer<Model> apply(final ZonedDateTime t) {
        return m -> this.add(t, m);
    }

    /**<commit_msg>Help writing engine for live data<commit_after>package uk.ac.abdn.iotstreams.csparql;

import java.time.ZonedDateTime;
import java.util.function.Consumer;
import java.util.function.Function;

import uk.ac.abdn.iotstreams.util.IotStreamsException;
import uk.ac.abdn.iotstreams.util.Logging;

import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;

import eu.larkc.csparql.cep.api.RdfQuadruple;
import eu.larkc.csparql.cep.api.RdfStream;
import eu.larkc.csparql.common.config.Config;
import eu.larkc.csparql.core.engine.CsparqlEngineImpl;

/**
 * 
 * @author nhc
 *
 * An IotStreamsEngine is a specific CsparqlEngine with methods
 * and queries for this project.
 */
public final class IotStreamsEngine
    extends CsparqlEngineImpl 
    implements 
        Function<ZonedDateTime, Consumer<Model>>,
        Consumer<Model> {
    
    /** This engine's sole stream */
    private final RdfStream rdfStream = new RdfStream("http://iotstreams");
    
    /** Object to which all inferences will be passed */
    private final Consumer<Model> persistentModel;
    
    /** Number of quadruples put on stream so far */
    private long numQuads = 0;
    
    /**
     * Verifies that C-SPARQL is configured for using recorded data,
     * then returns an IotStreamsEngine.
     * @param inferredDataConsumer Object to which all inferences will be passed
     * @return A fresh instance of IotStreamsEngine
     */
    public static IotStreamsEngine forRecordedData(final Consumer<Model> inferredDataConsumer) {
        if (Config.INSTANCE.isEsperUsingExternalTimestamp()) {
            return new IotStreamsEngine(inferredDataConsumer);
        } else {
            throw IotStreamsException.configurationError(
                    "To use recorded data, csparql.properties must contain the line "
                    + "'esper.externaltime.enabled=true'");
        }
    }
    
    /**
     * Verifies that C-SPARQL is configured for using live data,
     * then returns an IotStreamsEngine.
     * @param inferredDataConsumer Object to which all inferences will be passed
     * @return A fresh instance of IotStreamsEngine
     */
    public static IotStreamsEngine forLiveData(final Consumer<Model> inferredDataConsumer) {
        if (Config.INSTANCE.isEsperUsingExternalTimestamp()) {
            throw IotStreamsException.configurationError(
                    "When using live data, csparql.properties must not contain the line "
                    + "'esper.externaltime.enabled=true'");
        } else {
            return new IotStreamsEngine(inferredDataConsumer);
        }
    }
    
    /**
     * Initializes this engine.
     * @param persistentModel Object to which all inferences will be passed
     */
    private IotStreamsEngine(final Consumer<Model> persistentModel) {
        this.persistentModel = persistentModel;
        this.initialize();
        this.registerStream(this.rdfStream);
        new Configurator(this, this.persistentModel);
    }
    
    /**
     * Logs a status message
     */
    public void log() {
        Logging.info(String.format("%d quadruples put on stream", this.numQuads));
    }
    
    /**
     * To add a Jena Model for a given timestamp, use engine.apply(t).accept(model)
     */
    @Override
    public Consumer<Model> apply(final ZonedDateTime t) {
        return m -> this.add(t, m);
    }
    
    /**
     * To add a Jena Model for the current time (i.e. NOW), use engine.accept(model)
     */
    @Override
    public void accept(final Model m) {
        this.add(ZonedDateTime.now(), m);
    }
    

    /**
{'fraction_non_alphanumeric': 0.09243462452803804, 'fraction_numerical': 0.00027968116347364006, 'mean_word_length': 2.7661927330173777, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '5563413', 'n_tokens_mistral': 2151, 'n_tokens_neox': 2109, 'n_words': 534}