<commit_before>package org.yamcs;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.yamcs.StreamConfig.StandardStreamType;
import org.yamcs.StreamConfig.StreamConfigEntry;
import org.yamcs.archive.PacketWithTime;
import org.yamcs.xtce.SequenceContainer;
import org.yamcs.xtce.XtceDb;
import org.yamcs.xtceproc.XtceDbFactory;
import org.yamcs.yarch.Stream;
import org.yamcs.yarch.StreamSubscriber;
import org.yamcs.yarch.Tuple;
import org.yamcs.yarch.YarchDatabase;
import org.yamcs.yarch.YarchDatabaseInstance;

import com.google.common.util.concurrent.AbstractService;

/**
 * Receives packets from yamcs streams and sends them to the Processor/TmProcessor for extraction of parameters.
 * 
 * Can read from multiple streams, each with its own root container used as start of XTCE packet processing
 * 
 * @author nm
 *
 */
public class StreamTmPacketProvider extends AbstractService implements TmPacketProvider {
    Stream stream;
    TmProcessor tmProcessor;
    volatile boolean disabled = false;
    volatile long lastPacketTime;

    List<StreamReader> readers = new ArrayList<>();
    String yamcsInstance;
    YConfiguration config;

    public StreamTmPacketProvider(String yamcsInstance, YConfiguration config) throws ConfigurationException {
        this.yamcsInstance = yamcsInstance;

        if (!config.containsKey("streams")) {
            throw new ConfigurationException("Cannot find key 'streams' in StreamTmPacketProvider");
        }
        this.config = config;
    }

    @Override
    public void init(Processor proc) {
        this.tmProcessor = proc.getTmProcessor();
        readStreamConfig(proc.getName());
        proc.setPacketProvider(this);

    }

    /**
     * add to readers all the streams specifically specified in the streams config (in processor.yaml) or those that
     * have configured the processor to this processor in yamcs.instance.yaml
     */
    private void readStreamConfig(String procName) {
        YarchDatabaseInstance ydb = YarchDatabase.getInstance(yamcsInstance);
        XtceDb xtcedb = XtceDbFactory.getInstance(yamcsInstance);

        StreamConfig streamConfig = StreamConfig.getInstance(yamcsInstance);

        Set<String> streams = new HashSet<>();
        streams.addAll(config.getList("streams"));

        for (StreamConfigEntry sce : streamConfig.getEntries(StandardStreamType.tm)) {
            if(procName.equals(sce.getProcessor())) {
                streams.add(sce.getName());
            }
        }

        for (String streamName : streams) {
            StreamConfigEntry sce = streamConfig.getEntry(StandardStreamType.tm, streamName);
            SequenceContainer rootContainer;
            if (sce.getRootContainer() != null) {
                rootContainer = sce.getRootContainer();
            } else {
                rootContainer = xtcedb.getRootSequenceContainer();
                if (rootContainer == null) {<commit_msg>avoid NPE in case of wrong config<commit_after>package org.yamcs;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.yamcs.StreamConfig.StandardStreamType;
import org.yamcs.StreamConfig.StreamConfigEntry;
import org.yamcs.archive.PacketWithTime;
import org.yamcs.xtce.SequenceContainer;
import org.yamcs.xtce.XtceDb;
import org.yamcs.xtceproc.XtceDbFactory;
import org.yamcs.yarch.Stream;
import org.yamcs.yarch.StreamSubscriber;
import org.yamcs.yarch.Tuple;
import org.yamcs.yarch.YarchDatabase;
import org.yamcs.yarch.YarchDatabaseInstance;

import com.google.common.util.concurrent.AbstractService;

/**
 * Receives packets from yamcs streams and sends them to the Processor/TmProcessor for extraction of parameters.
 * 
 * Can read from multiple streams, each with its own root container used as start of XTCE packet processing
 * 
 * @author nm
 *
 */
public class StreamTmPacketProvider extends AbstractService implements TmPacketProvider {
    Stream stream;
    TmProcessor tmProcessor;
    volatile boolean disabled = false;
    volatile long lastPacketTime;

    List<StreamReader> readers = new ArrayList<>();
    String yamcsInstance;
    YConfiguration config;

    public StreamTmPacketProvider(String yamcsInstance, YConfiguration config) throws ConfigurationException {
        this.yamcsInstance = yamcsInstance;

        if (!config.containsKey("streams")) {
            throw new ConfigurationException("Cannot find key 'streams' in StreamTmPacketProvider");
        }
        this.config = config;
    }

    @Override
    public void init(Processor proc) {
        this.tmProcessor = proc.getTmProcessor();
        readStreamConfig(proc.getName());
        proc.setPacketProvider(this);

    }

    /**
     * add to readers all the streams specifically specified in the streams config (in processor.yaml) or those that
     * have configured the processor to this processor in yamcs.instance.yaml
     */
    private void readStreamConfig(String procName) {
        YarchDatabaseInstance ydb = YarchDatabase.getInstance(yamcsInstance);
        XtceDb xtcedb = XtceDbFactory.getInstance(yamcsInstance);

        StreamConfig streamConfig = StreamConfig.getInstance(yamcsInstance);

        Set<String> streams = new HashSet<>();
        streams.addAll(config.getList("streams"));

        for (StreamConfigEntry sce : streamConfig.getEntries(StandardStreamType.tm)) {
            if(procName.equals(sce.getProcessor())) {
                streams.add(sce.getName());
            }
        }

        for (String streamName : streams) {
            StreamConfigEntry sce = streamConfig.getEntry(StandardStreamType.tm, streamName);
            SequenceContainer rootContainer;
            
            if (sce!=null && sce.getRootContainer() != null) {
                rootContainer = sce.getRootContainer();
            } else {
                rootContainer = xtcedb.getRootSequenceContainer();
                if (rootContainer == null) {
{'fraction_non_alphanumeric': 0.08657361249584579, 'fraction_numerical': 0.0, 'mean_word_length': 3.336455331412104, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '6646504', 'n_tokens_mistral': 1676, 'n_tokens_neox': 1642, 'n_words': 401}