Proposed SatnogsTransferService to perform deduplication on Satnogs Telemetry API Result Field

pom.xml

@@ -38,6 +38,21 @@
       <artifactId>commons-codec</artifactId>
       <version>1.15</version>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+      <version>2.15.2</version>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <version>2.15.2</version>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-annotations</artifactId>
+      <version>2.15.2</version>
+    </dependency>
   </dependencies>
 
   <build>

src/main/java/org/oresat/uniclogs/links/SatnogsTmLink.java

@@ -1,9 +1,23 @@
 package org.oresat.uniclogs.links;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.yamcs.ConfigurationException;
+import org.yamcs.YConfiguration;
+import org.yamcs.YamcsServer;
 import org.yamcs.tctm.AbstractTmDataLink;
+import org.oresat.uniclogs.services.SatnogsTransferService;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+
 
 public class SatnogsTmLink extends AbstractTmDataLink {
 
+    SatnogsTransferService tsfrService;
+
     @Override
     protected Status connectionStatus() {
         // TODO Auto-generated method stub
@@ -12,14 +26,54 @@ protected Status connectionStatus() {
 
     @Override
     protected void doStart() {
-        // TODO Auto-generated method stub
-
+
+        try {
+
+            URL urlArtifacts = new URL("https://db.satnogs.org/api/telemetry/?format=json&satellite=98867");
+            HttpURLConnection conArtifacts = (HttpURLConnection) urlArtifacts.openConnection();
+            conArtifacts.setRequestProperty("Authorization", "Bearer " + System.getenv("SATNOGSDB_AUTH_TOKEN"));
+            conArtifacts.setRequestMethod("GET");
+            conArtifacts.setRequestProperty("contentType", "application/json; charset=utf-8");
+
+            BufferedReader inArtifacts = new BufferedReader(
+                    new InputStreamReader(conArtifacts.getInputStream()));
+            String inputLineArtifacts;
+            StringBuilder contentArtifacts = new StringBuilder();
+            while ((inputLineArtifacts = inArtifacts.readLine()) != null) {
+                contentArtifacts.append(inputLineArtifacts);
+            }
+            inArtifacts.close();
+
+            ObjectMapper objectMapper = new ObjectMapper();
+            JsonNode jsonNode = objectMapper.readTree(String.valueOf(contentArtifacts));
+
+            // Access specific fields from the JSON structure
+            String contentArtifactsSet = jsonNode.get("results").asText();
+
+            tsfrService.getDeduplication(contentArtifactsSet);
+
+        } catch (IOException e) {
+            this.log.error(e.getMessage());
+        }
     }
 
     @Override
     protected void doStop() {
         // TODO Auto-generated method stub
-
+
+    }
+
+    public SatnogsTmLink(String instanceName, YConfiguration config) {
+        String envName = config.getString("envService");
+        this.tsfrService = YamcsServer.getServer().getInstance(instanceName).getService(SatnogsTransferService.class, envName);
+
+        if (this.tsfrService == null) {
+            throw new ConfigurationException("Service " + envName + " does not exist or is not of class SatnogsTransferService.");
+        }
+    }
+
+    public SatnogsTmLink(String instanceName) {
+        this(instanceName, YConfiguration.emptyConfig());
     }
-    
+
 }

src/main/java/org/oresat/uniclogs/services/SatnogsTransferService.java

@@ -0,0 +1,122 @@
+package org.oresat.uniclogs.services;
+
+import org.yamcs.AbstractYamcsService;
+import org.yamcs.InitException;
+import org.yamcs.YConfiguration;
+import org.yamcs.utils.ByteArrayUtils;
+import org.yamcs.yarch.Bucket;
+import org.yamcs.yarch.YarchDatabase;
+import org.yamcs.yarch.YarchDatabaseInstance;
+import java.util.HashSet;
+import java.util.Set;
+
+import java.io.IOException;
+
+public class SatnogsTransferService extends AbstractYamcsService {
+    static final String sequenceNumberId = "seqNum";
+    static final String hmacKeyId = "hmacKey";
+    String hmacEnvVar;
+    byte[] hmacKey;
+    Integer seqNum;
+    Bucket db;
+
+    private byte[] loadHmacFromEnv(String varName) {
+        return System.getenv(varName).getBytes();
+    }
+
+    private Set<String> processedData;
+    // processedData set keeps track of unique data
+
+    @Override
+    protected void doStart() {
+        try {
+            this.hmacKey = this.loadHmacKey();
+            this.seqNum = this.loadSeqNum();
+            this.log.info("Loaded Sequence Number: " + this.seqNum);
+            processedData = new HashSet<>();
+        } catch (IOException e) {
+            this.log.error(e.getMessage());
+        }
+        this.notifyStarted();
+    }
+
+    @Override
+    protected void doStop() {
+        try {
+            this.saveSeqNum(this.seqNum);
+            this.saveHmacKey(this.hmacKey);
+        } catch (IOException e) {
+            this.log.error(e.getMessage());
+        }
+        this.notifyStopped();
+    }
+
+    public Integer getSeqNum() {
+        Integer num = this.seqNum;
+        this.seqNum++;
+        return num;
+    }
+
+    public Boolean getDeduplication(String contentArtifactsSet) {
+        // Check if the data is unique (not already processed)
+        if (!processedData.contains(contentArtifactsSet)){
+            System.out.println("Processing unique data: " + contentArtifactsSet);
+            // Add the processed data to the set
+            this.log.info("DB Satnogs Artifacts: " + processedData.add(contentArtifactsSet));
+            return true;
+        } else {
+            System.out.println("Duplicate data received: " + contentArtifactsSet);
+            // Handle duplicate data (If the text is present, it won't be added and the method returns false)
+            return false;
+        }
+    }
+
+    public byte[] getHmacKey() {
+        return this.hmacKey;
+    }
+
+    @Override
+    public void init(String yamcsInstance, String serviceName, YConfiguration config) throws InitException {
+        super.init(yamcsInstance, serviceName, config);
+        this.hmacEnvVar = config.getString("hmacEnvVar");
+        log.info(this.hmacEnvVar);
+        YarchDatabaseInstance instanceDb = YarchDatabase.getInstance(yamcsInstance);
+        try {
+            this.db = instanceDb.getBucket("env");
+            if (this.db == null) {
+                this.log.info(String.format("Env not found for %s, creating new env...", yamcsInstance));
+                this.db = instanceDb.createBucket("env");
+            }
+        } catch (IOException e) {
+            throw new InitException(e.getMessage());
+        }
+    }
+
+    private Integer loadSeqNum() throws IOException {
+        byte[] numBytes = db.getObject(sequenceNumberId);
+        if (numBytes == null) {
+            this.log.info(String.format("Sequence Number not found for %s, Sequence Number set to 1", yamcsInstance));
+            this.saveSeqNum(1);
+            return 1;
+        }
+        return ByteArrayUtils.decodeInt(db.getObject(sequenceNumberId), 0);
+    }
+
+    private byte[] loadHmacKey() throws IOException {
+        byte[] hmac = this.db.getObject(hmacKeyId);
+        if (hmac == null) {
+            this.log.info(String.format("Hmac Key not in %s. Loading from env var: %s", yamcsInstance, this.hmacEnvVar));
+            hmac = this.loadHmacFromEnv(this.hmacEnvVar);
+            this.saveHmacKey(hmac);
+        }
+        return hmac;
+    }
+
+    private void saveSeqNum(Integer seq) throws IOException {
+        db.putObject(sequenceNumberId, "Integer", null, ByteArrayUtils.encodeInt(seq));
+    }
+
+    private void saveHmacKey(byte[] hmacKey) throws IOException {
+        db.putObject(hmacKeyId, "bytes", null, hmacKey);
+    }
+}

src/main/java/org/oresat/uniclogs/tctm/SatnogsCommandPostprocessor.java

@@ -0,0 +1,47 @@
+package org.oresat.uniclogs.tctm;
+import org.oresat.uniclogs.services.SatnogsTransferService;
+import org.yamcs.ConfigurationException;
+import org.yamcs.YConfiguration;
+import org.yamcs.YamcsServer;
+import org.yamcs.cmdhistory.CommandHistoryPublisher;
+import org.yamcs.commanding.PreparedCommand;
+import org.yamcs.tctm.CommandPostprocessor;
+
+
+public class SatnogsCommandPostprocessor implements CommandPostprocessor {
+    protected CommandHistoryPublisher cmdHistory;
+    SatnogsTransferService env;
+
+    @Override
+    public byte[] process(PreparedCommand pc) {
+        Integer seqNum = this.env.getSeqNum();
+        byte[] hmacKey = this.env.getHmacKey();
+
+        EDLPacket cmd = new EDLPacket(pc.getBinary(), seqNum, hmacKey);
+        byte[] cmdBin = cmd.getBinary();
+
+        pc.setBinary(cmdBin);
+        this.cmdHistory.publish(pc.getCommandId(), "edl-seqnum", seqNum);
+        this.cmdHistory.publish(pc.getCommandId(), PreparedCommand.CNAME_BINARY, cmdBin);
+        return cmdBin;
+    }
+
+    @Override
+    public void setCommandHistoryPublisher(CommandHistoryPublisher commandHistoryListener) {
+        this.cmdHistory = commandHistoryListener;
+    }
+
+    public SatnogsCommandPostprocessor(String instanceName, YConfiguration config) {
+        String envName = config.getString("envService");
+        this.env = YamcsServer.getServer().getInstance(instanceName).getService(SatnogsTransferService.class, envName);
+
+        if (this.env == null) {
+            throw new ConfigurationException("Service " + envName + " does not exist or is not of class SatnogsTransferService.");
+        }
+    }
+
+    public SatnogsCommandPostprocessor(String instanceName) {
+        this(instanceName, YConfiguration.emptyConfig());
+    }
+
+}

src/main/java/org/oresat/uniclogs/tctm/SatnogsPacket.java

@@ -0,0 +1,53 @@
+package org.oresat.uniclogs.tctm;
+
+import java.util.Arrays;
+
+import org.apache.commons.codec.digest.HmacAlgorithms;
+import org.apache.commons.codec.digest.HmacUtils;
+import org.yamcs.TmPacket;
+import org.yamcs.xtce.util.HexUtils;
+
+
+public class SatnogsPacket extends Packet {
+    private static final Integer SEQ_NUM_OFFSET = 7;
+
+    public SatnogsPacket(byte[] packet, Integer seqNum, byte[] hmacSecret) {
+        //sequence number offset of 7
+        super(packet, packet.length+36, seqNum, SEQ_NUM_OFFSET);
+
+        // set sequence number in packet
+        this.encodeSeqNum();
+
+
+        // set frame length in packet: C = (Total Number of Octets in the Transfer Frame) − 1
+        // CRC adds 4, HMAC adds 32 -> (size + (36 - 1))
+        this.encodeFrameLength(35, 4);
+        this.addHmac(hmacSecret);
+
+        // Add CRC data to packet
+        this.encodeCrc();
+
+    }
+
+    private void addHmac(byte[] hmacSecret) {
+        byte[] hmac = new HmacUtils(HmacAlgorithms.HMAC_SHA_256, hmacSecret).hmac(this.data.array());
+        log.info(String.format("Hmac bytes: %s", Arrays.toString(hmac)));
+        this.data.put(hmac);
+        log.info(String.format("HMAC_SHA_256 (%s) added to packet (seqNum: %d).", HexUtils.hex(hmac), this.sequenceNumber));
+    }
+
+    protected void encodeCrc() {
+        int crc = this.calcCrc(this.data.array());
+        log.info(String.format("CRC_32 (%d) added to packet (seqNum: %d).", crc, this.sequenceNumber));
+        this.data.putInt(crc);
+    }
+
+    public SatnogsPacket(TmPacket tmPacket) {
+        super(tmPacket.getPacket(), tmPacket.getPacket().length, getSequenceNumber(tmPacket.getPacket(), SEQ_NUM_OFFSET), SEQ_NUM_OFFSET);
+    }
+
+    @Override
+    boolean validCrc() {
+        return crc32(0, this.data.array().length);
+    }
+}

src/main/java/org/oresat/uniclogs/tctm/SatnogsPacketPreprocessor.java

@@ -0,0 +1,31 @@
+package org.oresat.uniclogs.tctm;
+
+import java.util.Date;
+
+import org.yamcs.TmPacket;
+import org.yamcs.YConfiguration;
+import org.yamcs.tctm.AbstractPacketPreprocessor;
+
+public class SatnogsPacketPreprocessor extends AbstractPacketPreprocessor {
+
+    public SatnogsPacketPreprocessor(String yamcsInstance, YConfiguration config) {
+        super(yamcsInstance, config);
+    }
+
+    public SatnogsPacketPreprocessor(String yamcsInstance) {
+        super(yamcsInstance, YConfiguration.emptyConfig());
+    }
+
+    @Override
+    public TmPacket process(TmPacket tmPacket) {
+        SatnogsPacket packet = new SatnogsPacket(tmPacket);
+        tmPacket.setSequenceCount(packet.getSeqNum());
+        if (!packet.validCrc()) {
+            tmPacket.setInvalid();
+            this.eventProducer.sendWarning("PACKET_CORRUPT", "Satnogs Response Packet Corrupted");
+        }
+        tmPacket.setGenerationTime(new Date().getTime());
+        tmPacket.setLocalGenTimeFlag();
+        return tmPacket;
+    }
+}

src/main/yamcs/etc/processor.yaml

@@ -2,7 +2,7 @@ realtime:
   services:
     - class: org.yamcs.StreamTmPacketProvider
       args:
-        streams: ["edl_tm_realtime", "beacon_tm_realtime", "tm_dump"]
+        streams: ["edl_tm_realtime", "beacon_tm_realtime", "tm_dump", "satnogs_tm_realtime"]
     - class: org.yamcs.StreamTcCommandReleaser
     - class: org.yamcs.tctm.StreamParameterProvider
     - class: org.yamcs.algorithms.AlgorithmManager