<commit_before>package org.yamcs.tctm.ccsds;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.yamcs.cmdhistory.CommandHistoryPublisher.AckStatus;
import org.yamcs.commanding.PreparedCommand;
import org.yamcs.parameter.AggregateValue;
import org.yamcs.parameter.ParameterValue;
import org.yamcs.parameter.SystemParametersCollector;
import org.yamcs.protobuf.Clcw;
import org.yamcs.protobuf.Commanding.CommandHistoryAttribute;
import org.yamcs.protobuf.Cop1Config;
import org.yamcs.protobuf.Cop1State;
import org.yamcs.protobuf.Cop1Status;
import org.yamcs.protobuf.TimeoutType;
import org.yamcs.tctm.AbstractTcDataLink;
import org.yamcs.tctm.ccsds.Cop1Monitor.AlertType;
import org.yamcs.tctm.ccsds.TcManagedParameters.TcVcManagedParameters;
import org.yamcs.utils.TimeEncoding;
import org.yamcs.utils.ValueUtility;
import org.yamcs.xtce.util.AggregateMemberNames;

import static org.yamcs.cmdhistory.CommandHistoryPublisher.*;

/**
 * Assembles TC packets into TC frames as per CCSDS 232.0-B-3 and sends them out via FOP1
 * 
 * <p>
 * Implements the FOP (transmitter) part of the Communications Operations Procedure-1
 * CCSDS 232.1-B-2 September 2010
 * <p>
 * The FOP1 implementation is a little different than the standard: the "Initiate AD service with CLCW check" will wait
 * for the first CLCW and immediately set the vS to the nR in the CLCW. The standard specifies that the vS has somehow
 * to be set manually to an CLCW observed value before calling the "Initiate AD with CLCW check" directive.
 * 
 * @author nm
 *
 */
public class Cop1TcPacketHandler extends AbstractTcDataLink implements VcUplinkHandler {

    static final int OUT_QUEUE_SIZE = 20;

    // the frames to be sent out are placed here
    BlockingQueue<QueuedFrame> outQueue;

    /**
     * If this is true, we transform the BD packets to frames and put them directly on the outQueue
     * <p>
     * if it's false, we put the BD packets in the inQueue to be taken out in between AD packets
     * <p>
     * Note that if the fop1 state is 4,5 or 6 (i.e. initialising) then the BD packets are also put directly in the
     * outQueue
     */
    boolean bdAbsolutePriority;

    /**
     * If this is false, all frames will be sent directly without passing through the FOP1 state machine
     */
    boolean cop1Active = true;
    /**
     * Used if cop1Active = false, if set all the TC packets will be sent in frames with the bypass flag set
     */
    boolean bypassAll = true;

    TcVcManagedParameters vmp;
    TcFrameFactory frameFactory;

    static final public String CMD_ATTRIBUTE_BYPASS = "cop1Bypass";

    // used to signal to the master channel when data is available on this VC
    private Semaphore dataAvailableSemaphore;
    final ScheduledThreadPoolExecutor executor;

    protected ArrayDeque<PreparedCommand> waitQueue = new ArrayDeque<>();

    static final int INVALID_CLCW = -1;
    /**
     * 1 - Active
     * 2 - Retransmit without wait
     * 3 - Retransmit with wait
     * 4 - Initialising without BC Frame
     * 5 - Initialising with BC Frame
     * 6 - Initial
     */
    int state = 6;

    // _state is seen by external threads
    volatile int externalState = state;

    // V(S) - Transmitter Frame Sequence Number;
    int vS;

    // Sent_Queue;
    QueuedFrame[] sentQueue = new QueuedFrame[256];

    boolean adOutReady = true;
    boolean bcOutReady = true;

    // NN(R) - Expected Acknowledgement Frame Sequence Number - the value of nR from the previous CLCW
    int nnR;

    // Timer_Initial_Value (also known as ‘T1_Initial’) in milliseconds
    private long t1Initial = 2000;

    // Transmission_Limit;
    int txLimit = 3;

    // Transmission_Count;
    int txCount;

    // FOP_Sliding_Window_Width (also known as ‘K’);
    int windowWidth = 10;

    // Timeout_Type (TT);
    // 1-> SUSPEND, 0 -> go to uninitialized state
    int timeoutType = 1;

    // Suspend_State (SS).
    // 0 = not suspended
    // 1-4 = state in which has been suspended (see state above)
    int suspendState;
    AtomicInteger _clcw = new AtomicInteger(INVALID_CLCW);

    // these are received in the CLCW
    byte clcwLockout;
    byte clcwWait;
    byte clcwRetransmit;

    // timestamp of the reception of the last clcw
    long clcwTimestamp = TimeEncoding.INVALID_INSTANT;

    // used at startup for the initial CLCW wait time
    long initialClcwWait;

    /** N(R) - The Next Expected Frame Sequence Number as received in the last CLCW */
    int nR;

    ScheduledFuture<?> timer;

    private QueuedFrame pendingBCFrame;

    CopyOnWriteArrayList<Cop1Monitor> monitors = new CopyOnWriteArrayList<>();

    final int vcId;
    String clcwStreamName;
    ClcwStreamHelper clcwHelper;

    protected String sv_cop1Status_id;
    private volatile ParameterValue cop1Status;
    final static AggregateMemberNames cop1StatusMembers = AggregateMemberNames.get(new String[] { "cop1Active",
            "state", "waitQueueNumTC", "sentQueueNumFrames", "vS", "nnR" });

    public Cop1TcPacketHandler(String yamcsInstance, String linkName,
            TcVcManagedParameters vmp, ScheduledThreadPoolExecutor executor) {
        super(yamcsInstance, linkName, vmp.config);

        this.frameFactory = vmp.getFrameFactory();
        this.executor = executor;
        this.vmp = vmp;
        this.vcId = vmp.vcId;
        outQueue = new ArrayBlockingQueue<>(OUT_QUEUE_SIZE);
        clcwStreamName = vmp.config.getString("clcwStream");
        this.initialClcwWait = 1000 * vmp.config.getInt("initialClcwWait", -1);
    }

    public void addMonitor(Cop1Monitor monitor) {
        monitors.add(monitor);
    }

    public void removeMonitor(Cop1Monitor monitor) {
        monitors.remove(monitor);
    }

    @Override
    public void sendTc(PreparedCommand pc) {
        if (disabled) {
            log.debug("TC disabled, ignoring command {}", pc.getCommandId());
            if (failCommandOnDisabled) {
                failedCommand(pc.getCommandId(), "Link " + name + " disabled");
            }
            return;
        }
        log.trace("state: {} new TC {}", externalState, pc);
        if (!cop1Active) {
            if (bypassAll) {
                sendSingleTc(pc, true);
            } else {
                sendSingleTc(pc, isBypass(pc));<commit_msg>allow to provide t1 in the config file<commit_after>package org.yamcs.tctm.ccsds;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.yamcs.cmdhistory.CommandHistoryPublisher.AckStatus;
import org.yamcs.commanding.PreparedCommand;
import org.yamcs.parameter.AggregateValue;
import org.yamcs.parameter.ParameterValue;
import org.yamcs.parameter.SystemParametersCollector;
import org.yamcs.protobuf.Clcw;
import org.yamcs.protobuf.Commanding.CommandHistoryAttribute;
import org.yamcs.protobuf.Cop1Config;
import org.yamcs.protobuf.Cop1State;
import org.yamcs.protobuf.Cop1Status;
import org.yamcs.protobuf.TimeoutType;
import org.yamcs.tctm.AbstractTcDataLink;
import org.yamcs.tctm.ccsds.Cop1Monitor.AlertType;
import org.yamcs.tctm.ccsds.TcManagedParameters.TcVcManagedParameters;
import org.yamcs.utils.TimeEncoding;
import org.yamcs.utils.ValueUtility;
import org.yamcs.xtce.util.AggregateMemberNames;

import static org.yamcs.cmdhistory.CommandHistoryPublisher.*;

/**
 * Assembles TC packets into TC frames as per CCSDS 232.0-B-3 and sends them out via FOP1
 * 
 * <p>
 * Implements the FOP (transmitter) part of the Communications Operations Procedure-1
 * CCSDS 232.1-B-2 September 2010
 * <p>
 * The FOP1 implementation is a little different than the standard: the "Initiate AD service with CLCW check" will wait
 * for the first CLCW and immediately set the vS to the nR in the CLCW. The standard specifies that the vS has somehow
 * to be set manually to an CLCW observed value before calling the "Initiate AD with CLCW check" directive.
 * 
 * @author nm
 *
 */
public class Cop1TcPacketHandler extends AbstractTcDataLink implements VcUplinkHandler {

    static final int OUT_QUEUE_SIZE = 20;

    // the frames to be sent out are placed here
    BlockingQueue<QueuedFrame> outQueue;

    /**
     * If this is true, we transform the BD packets to frames and put them directly on the outQueue
     * <p>
     * if it's false, we put the BD packets in the inQueue to be taken out in between AD packets
     * <p>
     * Note that if the fop1 state is 4,5 or 6 (i.e. initialising) then the BD packets are also put directly in the
     * outQueue
     */
    boolean bdAbsolutePriority;

    /**
     * If this is false, all frames will be sent directly without passing through the FOP1 state machine
     */
    boolean cop1Active = true;
    /**
     * Used if cop1Active = false, if set all the TC packets will be sent in frames with the bypass flag set
     */
    boolean bypassAll = true;

    TcVcManagedParameters vmp;
    TcFrameFactory frameFactory;

    static final public String CMD_ATTRIBUTE_BYPASS = "cop1Bypass";

    // used to signal to the master channel when data is available on this VC
    private Semaphore dataAvailableSemaphore;
    final ScheduledThreadPoolExecutor executor;

    protected ArrayDeque<PreparedCommand> waitQueue = new ArrayDeque<>();

    static final int INVALID_CLCW = -1;
    /**
     * 1 - Active
     * 2 - Retransmit without wait
     * 3 - Retransmit with wait
     * 4 - Initialising without BC Frame
     * 5 - Initialising with BC Frame
     * 6 - Initial
     */
    int state = 6;

    // _state is seen by external threads
    volatile int externalState = state;

    // V(S) - Transmitter Frame Sequence Number;
    int vS;

    // Sent_Queue;
    QueuedFrame[] sentQueue = new QueuedFrame[256];

    boolean adOutReady = true;
    boolean bcOutReady = true;

    // NN(R) - Expected Acknowledgement Frame Sequence Number - the value of nR from the previous CLCW
    int nnR;

    // Timer_Initial_Value (also known as ‘T1_Initial’) in milliseconds
    private long t1Initial;

    // Transmission_Limit;
    int txLimit = 3;

    // Transmission_Count;
    int txCount;

    // FOP_Sliding_Window_Width (also known as ‘K’);
    int windowWidth = 10;

    // Timeout_Type (TT);
    // 1-> SUSPEND, 0 -> go to uninitialized state
    int timeoutType = 1;

    // Suspend_State (SS).
    // 0 = not suspended
    // 1-4 = state in which has been suspended (see state above)
    int suspendState;
    AtomicInteger _clcw = new AtomicInteger(INVALID_CLCW);

    // these are received in the CLCW
    byte clcwLockout;
    byte clcwWait;
    byte clcwRetransmit;

    // timestamp of the reception of the last clcw
    long clcwTimestamp = TimeEncoding.INVALID_INSTANT;

    // used at startup for the initial CLCW wait time
    long initialClcwWait;

    /** N(R) - The Next Expected Frame Sequence Number as received in the last CLCW */
    int nR;

    ScheduledFuture<?> timer;

    private QueuedFrame pendingBCFrame;

    CopyOnWriteArrayList<Cop1Monitor> monitors = new CopyOnWriteArrayList<>();

    final int vcId;
    String clcwStreamName;
    ClcwStreamHelper clcwHelper;

    protected String sv_cop1Status_id;
    private volatile ParameterValue cop1Status;
    final static AggregateMemberNames cop1StatusMembers = AggregateMemberNames.get(new String[] { "cop1Active",
            "state", "waitQueueNumTC", "sentQueueNumFrames", "vS", "nnR" });

    public Cop1TcPacketHandler(String yamcsInstance, String linkName,
            TcVcManagedParameters vmp, ScheduledThreadPoolExecutor executor) {
        super(yamcsInstance, linkName, vmp.config);

        this.frameFactory = vmp.getFrameFactory();
        this.executor = executor;
        this.vmp = vmp;
        this.vcId = vmp.vcId;
        outQueue = new ArrayBlockingQueue<>(OUT_QUEUE_SIZE);
        clcwStreamName = vmp.config.getString("clcwStream");
        this.initialClcwWait = 1000 * vmp.config.getInt("initialClcwWait", -1);
        this.t1Initial =  1000 * vmp.config.getInt("t1", 3);
    }

    public void addMonitor(Cop1Monitor monitor) {
        monitors.add(monitor);
    }

    public void removeMonitor(Cop1Monitor monitor) {
        monitors.remove(monitor);
    }

    @Override
    public void sendTc(PreparedCommand pc) {
        if (disabled) {
            log.debug("TC disabled, ignoring command {}", pc.getCommandId());
            if (failCommandOnDisabled) {
                failedCommand(pc.getCommandId(), "Link " + name + " disabled");
            }
            return;
        }
        log.trace("state: {} new TC {}", externalState, pc);
        if (!cop1Active) {
            if (bypassAll) {
                sendSingleTc(pc, true);
            } else {
                sendSingleTc(pc, isBypass(pc));
{'fraction_non_alphanumeric': 0.08366562749220618, 'fraction_numerical': 0.010730080475603566, 'mean_word_length': 3.208053691275168, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 29, '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': '2278007', 'n_tokens_mistral': 4342, 'n_tokens_neox': 4094, 'n_words': 1243}