Merge pull request 'bump_tmtccmd' (#88) from bump_tmtccmd into main

Reviewed-on: #88

README.md

@@ -5,7 +5,9 @@ sat-rs
 
 This is the repository of the sat-rs framework. Its primary goal is to provide re-usable components
 to write on-board software for remote systems like rovers or satellites. It is specifically written
-for the special requirements for these systems.
+for the special requirements for these systems. You can find an overview of the project and the
+link to the [more high-level sat-rs book](https://documentation.irs.uni-stuttgart.de/projects/sat-rs/)
+at the [IRS documentation website](https://documentation.irs.uni-stuttgart.de/sat-rs.html).
 
 A lot of the architecture and general design considerations are based on the
 [FSFW](https://egit.irs.uni-stuttgart.de/fsfw/fsfw) C++ framework which has flight heritage
@@ -16,6 +18,10 @@ and [EIVE](https://www.irs.uni-stuttgart.de/en/research/satellitetechnology-and-
 
 This project currently contains following crates:
 
+* [`satrs-book`](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-book):
+   Primary information resource in addition to the API documentation, hosted
+   [here](https://documentation.irs.uni-stuttgart.de/projects/sat-rs/). It can be useful to read
+   this first before delving into the example application and the API documentation.
 * [`satrs-core`](https://egit.irs.uni-stuttgart.de/rust/satrs-launchpad/src/branch/main/satrs-core):
    Core components of sat-rs.
 * [`satrs-example`](https://egit.irs.uni-stuttgart.de/rust/satrs-launchpad/src/branch/main/satrs-example):

satrs-book/src/introduction.md

@@ -21,3 +21,9 @@ A lot of the architecture and general design considerations are based on the
 through the 2 missions [FLP](https://www.irs.uni-stuttgart.de/en/research/satellitetechnology-and-instruments/smallsatelliteprogram/flying-laptop/)
 and [EIVE](https://www.irs.uni-stuttgart.de/en/research/satellitetechnology-and-instruments/smallsatelliteprogram/EIVE/).
 
+# Getting started with the example
+
+The [`satrs-example`](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example)
+provides various practical usage examples of the `sat-rs` framework. If you are more interested in
+the practical application of `sat-rs` inside an application, it is recommended to have a look at
+the example application.

satrs-core/src/encoding/ccsds.rs

@@ -30,6 +30,12 @@ impl PacketIdLookup for [u16] {
     }
 }
 
+impl PacketIdLookup for &[u16] {
+    fn validate(&self, packet_id: u16) -> bool {
+        self.binary_search(&packet_id).is_ok()
+    }
+}
+
 #[cfg(feature = "alloc")]
 impl PacketIdLookup for Vec<PacketId> {
     fn validate(&self, packet_id: u16) -> bool {
@@ -49,6 +55,12 @@ impl PacketIdLookup for [PacketId] {
     }
 }
 
+impl PacketIdLookup for &[PacketId] {
+    fn validate(&self, packet_id: u16) -> bool {
+        self.binary_search(&PacketId::from(packet_id)).is_ok()
+    }
+}
+
 /// This function parses a given buffer for tightly packed CCSDS space packets. It uses the
 /// [PacketId] field of the CCSDS packets to detect the start of a CCSDS space packet and then
 /// uses the length field of the packet to extract CCSDS packets.

satrs-core/src/hal/std/tcp_cobs_server.rs

@@ -1,4 +1,3 @@
-use alloc::boxed::Box;
 use alloc::vec;
 use cobs::encode;
 use delegate::delegate;
@@ -29,7 +28,6 @@ impl<TmError, TcError: 'static> TcpTcParser<TmError, TcError> for CobsTcParser {
         current_write_idx: usize,
         next_write_idx: &mut usize,
     ) -> Result<(), TcpTmtcError<TmError, TcError>> {
-        // Reader vec full, need to parse for packets.
         conn_result.num_received_tcs += parse_buffer_for_cobs_encoded_packets(
             &mut tc_buffer[..current_write_idx],
             tc_receiver.upcast_mut(),
@@ -111,11 +109,23 @@ impl<TmError, TcError> TcpTmSender<TmError, TcError> for CobsTmSender {
 ///
 /// The [TCP integration tests](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-core/tests/tcp_servers.rs)
 /// test also serves as the example application for this module.
-pub struct TcpTmtcInCobsServer<TmError, TcError: 'static> {
-    generic_server: TcpTmtcGenericServer<TmError, TcError, CobsTmSender, CobsTcParser>,
+pub struct TcpTmtcInCobsServer<
+    TmError,
+    TcError: 'static,
+    TmSource: TmPacketSource<Error = TmError>,
+    TcReceiver: ReceivesTc<Error = TcError>,
+> {
+    generic_server:
+        TcpTmtcGenericServer<TmError, TcError, TmSource, TcReceiver, CobsTmSender, CobsTcParser>,
 }
 
-impl<TmError: 'static, TcError: 'static> TcpTmtcInCobsServer<TmError, TcError> {
+impl<
+        TmError: 'static,
+        TcError: 'static,
+        TmSource: TmPacketSource<Error = TmError>,
+        TcReceiver: ReceivesTc<Error = TcError>,
+    > TcpTmtcInCobsServer<TmError, TcError, TmSource, TcReceiver>
+{
     /// Create a new TCP TMTC server which exchanges TMTC packets encoded with
     /// [COBS protocol](https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing).
     ///
@@ -128,9 +138,9 @@ impl<TmError: 'static, TcError: 'static> TcpTmtcInCobsServer<TmError, TcError> {
     ///     forwarded to this TC receiver.
     pub fn new(
         cfg: ServerConfig,
-        tm_source: Box<dyn TmPacketSource<Error = TmError>>,
-        tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
-    ) -> Result<Self, TcpTmtcError<TmError, TcError>> {
+        tm_source: TmSource,
+        tc_receiver: TcReceiver,
+    ) -> Result<Self, std::io::Error> {
         Ok(Self {
             generic_server: TcpTmtcGenericServer::new(
                 cfg,
@@ -177,7 +187,7 @@ mod tests {
             ServerConfig,
         },
     };
-    use alloc::{boxed::Box, sync::Arc};
+    use alloc::sync::Arc;
     use cobs::encode;
 
     use super::TcpTmtcInCobsServer;
@@ -202,11 +212,11 @@ mod tests {
         addr: &SocketAddr,
         tc_receiver: SyncTcCacher,
         tm_source: SyncTmSource,
-    ) -> TcpTmtcInCobsServer<(), ()> {
+    ) -> TcpTmtcInCobsServer<(), (), SyncTmSource, SyncTcCacher> {
         TcpTmtcInCobsServer::new(
             ServerConfig::new(*addr, Duration::from_millis(2), 1024, 1024),
-            Box::new(tm_source),
-            Box::new(tc_receiver),
+            tm_source,
+            tc_receiver,
         )
         .expect("TCP server generation failed")
     }

satrs-core/src/hal/std/tcp_server.rs

@@ -1,6 +1,6 @@
 //! Generic TCP TMTC servers with different TMTC format flavours.
 use alloc::vec;
-use alloc::{boxed::Box, vec::Vec};
+use alloc::vec::Vec;
 use core::time::Duration;
 use socket2::{Domain, Socket, Type};
 use std::io::Read;
@@ -134,20 +134,29 @@ pub trait TcpTmSender<TmError, TcError> {
 pub struct TcpTmtcGenericServer<
     TmError,
     TcError,
-    TmHandler: TcpTmSender<TmError, TcError>,
-    TcHandler: TcpTcParser<TmError, TcError>,
+    TmSource: TmPacketSource<Error = TmError>,
+    TcReceiver: ReceivesTc<Error = TcError>,
+    TmSender: TcpTmSender<TmError, TcError>,
+    TcParser: TcpTcParser<TmError, TcError>,
 > {
-    base: TcpTmtcServerBase<TmError, TcError>,
-    tc_handler: TcHandler,
-    tm_handler: TmHandler,
+    pub(crate) listener: TcpListener,
+    pub(crate) inner_loop_delay: Duration,
+    pub(crate) tm_source: TmSource,
+    pub(crate) tm_buffer: Vec<u8>,
+    pub(crate) tc_receiver: TcReceiver,
+    pub(crate) tc_buffer: Vec<u8>,
+    tc_handler: TcParser,
+    tm_handler: TmSender,
 }
 
 impl<
         TmError: 'static,
         TcError: 'static,
+        TmSource: TmPacketSource<Error = TmError>,
+        TcReceiver: ReceivesTc<Error = TcError>,
         TmSender: TcpTmSender<TmError, TcError>,
         TcParser: TcpTcParser<TmError, TcError>,
-    > TcpTmtcGenericServer<TmError, TcError, TmSender, TcParser>
+    > TcpTmtcGenericServer<TmError, TcError, TmSource, TcReceiver, TmSender, TcParser>
 {
     /// Create a new generic TMTC server instance.
     ///
@@ -165,25 +174,38 @@ impl<
         cfg: ServerConfig,
         tc_parser: TcParser,
         tm_sender: TmSender,
-        tm_source: Box<dyn TmPacketSource<Error = TmError>>,
-        tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
-    ) -> Result<TcpTmtcGenericServer<TmError, TcError, TmSender, TcParser>, std::io::Error> {
+        tm_source: TmSource,
+        tc_receiver: TcReceiver,
+    ) -> Result<Self, std::io::Error> {
+        // Create a TCP listener bound to two addresses.
+        let socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
+        socket.set_reuse_address(cfg.reuse_addr)?;
+        #[cfg(unix)]
+        socket.set_reuse_port(cfg.reuse_port)?;
+        let addr = (cfg.addr).into();
+        socket.bind(&addr)?;
+        socket.listen(128)?;
         Ok(Self {
-            base: TcpTmtcServerBase::new(cfg, tm_source, tc_receiver)?,
             tc_handler: tc_parser,
             tm_handler: tm_sender,
+            listener: socket.into(),
+            inner_loop_delay: cfg.inner_loop_delay,
+            tm_source,
+            tm_buffer: vec![0; cfg.tm_buffer_size],
+            tc_receiver,
+            tc_buffer: vec![0; cfg.tc_buffer_size],
         })
     }
 
     /// Retrieve the internal [TcpListener] class.
     pub fn listener(&mut self) -> &mut TcpListener {
-        self.base.listener()
+        &mut self.listener
     }
 
     /// Can be used to retrieve the local assigned address of the TCP server. This is especially
     /// useful if using the port number 0 for OS auto-assignment.
     pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
-        self.base.local_addr()
+        self.listener.local_addr()
     }
 
     /// This call is used to handle the next connection to a client. Right now, it performs
@@ -205,20 +227,20 @@ impl<
         let mut connection_result = ConnectionResult::default();
         let mut current_write_idx;
         let mut next_write_idx = 0;
-        let (mut stream, addr) = self.base.listener.accept()?;
+        let (mut stream, addr) = self.listener.accept()?;
         stream.set_nonblocking(true)?;
         connection_result.addr = Some(addr);
         current_write_idx = next_write_idx;
         loop {
-            let read_result = stream.read(&mut self.base.tc_buffer[current_write_idx..]);
+            let read_result = stream.read(&mut self.tc_buffer[current_write_idx..]);
             match read_result {
                 Ok(0) => {
                     // Connection closed by client. If any TC was read, parse for complete packets.
                     // After that, break the outer loop.
                     if current_write_idx > 0 {
                         self.tc_handler.handle_tc_parsing(
-                            &mut self.base.tc_buffer,
-                            self.base.tc_receiver.as_mut(),
+                            &mut self.tc_buffer,
+                            &mut self.tc_receiver,
                             &mut connection_result,
                             current_write_idx,
                             &mut next_write_idx,
@@ -229,10 +251,10 @@ impl<
                 Ok(read_len) => {
                     current_write_idx += read_len;
                     // TC buffer is full, we must parse for complete packets now.
-                    if current_write_idx == self.base.tc_buffer.capacity() {
+                    if current_write_idx == self.tc_buffer.capacity() {
                         self.tc_handler.handle_tc_parsing(
-                            &mut self.base.tc_buffer,
-                            self.base.tc_receiver.as_mut(),
+                            &mut self.tc_buffer,
+                            &mut self.tc_receiver,
                             &mut connection_result,
                             current_write_idx,
                             &mut next_write_idx,
@@ -245,8 +267,8 @@ impl<
                     // both UNIX and Windows.
                     std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut => {
                         self.tc_handler.handle_tc_parsing(
-                            &mut self.base.tc_buffer,
-                            self.base.tc_receiver.as_mut(),
+                            &mut self.tc_buffer,
+                            &mut self.tc_receiver,
                             &mut connection_result,
                             current_write_idx,
                             &mut next_write_idx,
@@ -254,14 +276,14 @@ impl<
                         current_write_idx = next_write_idx;
 
                         if !self.tm_handler.handle_tm_sending(
-                            &mut self.base.tm_buffer,
-                            self.base.tm_source.as_mut(),
+                            &mut self.tm_buffer,
+                            &mut self.tm_source,
                             &mut connection_result,
                             &mut stream,
                         )? {
                             // No TC read, no TM was sent, but the client has not disconnected.
                             // Perform an inner delay to avoid burning CPU time.
-                            thread::sleep(self.base.inner_loop_delay);
+                            thread::sleep(self.inner_loop_delay);
                         }
                     }
                     _ => {
@@ -271,8 +293,8 @@ impl<
             }
         }
         self.tm_handler.handle_tm_sending(
-            &mut self.base.tm_buffer,
-            self.base.tm_source.as_mut(),
+            &mut self.tm_buffer,
+            &mut self.tm_source,
             &mut connection_result,
             &mut stream,
         )?;
@@ -280,47 +302,6 @@ impl<
     }
 }
 
-pub(crate) struct TcpTmtcServerBase<TmError, TcError> {
-    pub(crate) listener: TcpListener,
-    pub(crate) inner_loop_delay: Duration,
-    pub(crate) tm_source: Box<dyn TmPacketSource<Error = TmError>>,
-    pub(crate) tm_buffer: Vec<u8>,
-    pub(crate) tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
-    pub(crate) tc_buffer: Vec<u8>,
-}
-
-impl<TmError, TcError> TcpTmtcServerBase<TmError, TcError> {
-    pub(crate) fn new(
-        cfg: ServerConfig,
-        tm_source: Box<dyn TmPacketSource<Error = TmError>>,
-        tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
-    ) -> Result<Self, std::io::Error> {
-        // Create a TCP listener bound to two addresses.
-        let socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
-        socket.set_reuse_address(cfg.reuse_addr)?;
-        socket.set_reuse_port(cfg.reuse_port)?;
-        let addr = (cfg.addr).into();
-        socket.bind(&addr)?;
-        socket.listen(128)?;
-        Ok(Self {
-            listener: socket.into(),
-            inner_loop_delay: cfg.inner_loop_delay,
-            tm_source,
-            tm_buffer: vec![0; cfg.tm_buffer_size],
-            tc_receiver,
-            tc_buffer: vec![0; cfg.tc_buffer_size],
-        })
-    }
-
-    pub(crate) fn listener(&mut self) -> &mut TcpListener {
-        &mut self.listener
-    }
-
-    pub(crate) fn local_addr(&self) -> std::io::Result<SocketAddr> {
-        self.listener.local_addr()
-    }
-}
-
 #[cfg(test)]
 pub(crate) mod tests {
     use std::sync::Mutex;

satrs-core/src/hal/std/tcp_spacepackets_server.rs

@@ -88,16 +88,31 @@ impl<TmError, TcError> TcpTmSender<TmError, TcError> for SpacepacketsTmSender {
 /// [spacepackets::PacketId]s as part of the server configuration for that purpose.
 ///
 /// ## Example
-///
 /// The [TCP server integration tests](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-core/tests/tcp_servers.rs)
 /// also serves as the example application for this module.
-pub struct TcpSpacepacketsServer<TmError, TcError: 'static> {
-    generic_server:
-        TcpTmtcGenericServer<TmError, TcError, SpacepacketsTmSender, SpacepacketsTcParser>,
+pub struct TcpSpacepacketsServer<
+    TmError,
+    TcError: 'static,
+    TmSource: TmPacketSource<Error = TmError>,
+    TcReceiver: ReceivesTc<Error = TcError>,
+> {
+    generic_server: TcpTmtcGenericServer<
+        TmError,
+        TcError,
+        TmSource,
+        TcReceiver,
+        SpacepacketsTmSender,
+        SpacepacketsTcParser,
+    >,
 }
 
-impl<TmError: 'static, TcError: 'static> TcpSpacepacketsServer<TmError, TcError> {
-    /// Create a new TCP TMTC server which exchanges CCSDS space packets.
+impl<
+        TmError: 'static,
+        TcError: 'static,
+        TmSource: TmPacketSource<Error = TmError>,
+        TcReceiver: ReceivesTc<Error = TcError>,
+    > TcpSpacepacketsServer<TmError, TcError, TmSource, TcReceiver>
+{
     ///
     /// ## Parameter
     ///
@@ -110,10 +125,10 @@ impl<TmError: 'static, TcError: 'static> TcpSpacepacketsServer<TmError, TcError>
     ///     parsing. This mechanism is used to have a start marker for finding CCSDS packets.
     pub fn new(
         cfg: ServerConfig,
-        tm_source: Box<dyn TmPacketSource<Error = TmError>>,
-        tc_receiver: Box<dyn ReceivesTc<Error = TcError>>,
+        tm_source: TmSource,
+        tc_receiver: TcReceiver,
         packet_id_lookup: Box<dyn PacketIdLookup + Send>,
-    ) -> Result<Self, TcpTmtcError<TmError, TcError>> {
+    ) -> Result<Self, std::io::Error> {
         Ok(Self {
             generic_server: TcpTmtcGenericServer::new(
                 cfg,
@@ -179,11 +194,11 @@ mod tests {
         tc_receiver: SyncTcCacher,
         tm_source: SyncTmSource,
         packet_id_lookup: HashSet<PacketId>,
-    ) -> TcpSpacepacketsServer<(), ()> {
+    ) -> TcpSpacepacketsServer<(), (), SyncTmSource, SyncTcCacher> {
         TcpSpacepacketsServer::new(
             ServerConfig::new(*addr, Duration::from_millis(2), 1024, 1024),
-            Box::new(tm_source),
-            Box::new(tc_receiver),
+            tm_source,
+            tc_receiver,
             Box::new(packet_id_lookup),
         )
         .expect("TCP server generation failed")

satrs-core/tests/tcp_servers.rs

@@ -94,8 +94,8 @@ fn test_cobs_server() {
     tm_source.add_tm(&INVERTED_PACKET);
     let mut tcp_server = TcpTmtcInCobsServer::new(
         ServerConfig::new(AUTO_PORT_ADDR, Duration::from_millis(2), 1024, 1024),
-        Box::new(tm_source),
-        Box::new(tc_receiver.clone()),
+        tm_source,
+        tc_receiver.clone(),
     )
     .expect("TCP server generation failed");
     let dest_addr = tcp_server
@@ -176,8 +176,8 @@ fn test_ccsds_server() {
     packet_id_lookup.insert(TEST_PACKET_ID_0);
     let mut tcp_server = TcpSpacepacketsServer::new(
         ServerConfig::new(AUTO_PORT_ADDR, Duration::from_millis(2), 1024, 1024),
-        Box::new(tm_source),
-        Box::new(tc_receiver.clone()),
+        tm_source,
+        tc_receiver.clone(),
         Box::new(packet_id_lookup),
     )
     .expect("TCP server generation failed");

satrs-example/Cargo.toml

@@ -19,9 +19,9 @@ num_enum = "0.7"
 thiserror = "1"
 
 [dependencies.satrs-core]
-# version = "0.1.0-alpha.0"
+# version = "0.1.0-alpha.1"
 path = "../satrs-core"
 
-
 [dependencies.satrs-mib]
-path = "../satrs-mib"
+version = "0.1.0-alpha.1"
+# path = "../satrs-mib"

satrs-example/README.md

@@ -17,7 +17,7 @@ cargo run --bin simpleclient
 This repository also contains a more complex client using the
 [Python tmtccmd](https://github.com/robamu-org/tmtccmd) module.
 
-# Using the tmtccmd Python client
+# <a id="tmtccmd"></a> Using the tmtccmd Python client
 
 The python client requires a valid installation of the
 [tmtccmd package](https://github.com/robamu-org/tmtccmd).
@@ -51,3 +51,26 @@ the `simpleclient`:
 
 You can also simply call the script without any arguments to view a list of services (`-s` flag)
 and corresponding op codes (`-o` flag) for each service.
+
+# Structure of the example project
+
+The example project contains components which could also be expected to be part of a production
+On-Board Software.
+
+1. A UDP and TCP server to receive telecommands and poll telemetry from. This might be an optional
+   component for an OBSW which is only used during the development phase on ground. The TCP
+   server parses space packets by using the CCSDS space packet ID as the packet start delimiter.
+2. A PUS service stack which exposes some functionality conformant with the ECSS PUS service. This
+   currently includes the following services:
+   - Service 1 for telecommand verification.
+   - Service 3 for housekeeping telemetry handling.
+   - Service 5 for management and downlink of on-board events.
+   - Service 8 for handling on-board actions.
+   - Service 11 for scheduling telecommands to be released at a specific time.
+   - Service 17 for test purposes (pings)
+3. An event manager component which handles the event IPC mechanism.
+4. A TC source component which demultiplexes and routes telecommands based on parameters like
+   packet APID or PUS service and subservice type.
+5. A TM sink sink component which is the target of all sent telemetry and sends it to downlink
+   handlers like the UDP and TCP server.
+6. An AOCS example task which can also process some PUS commands.

satrs-example/pyclient/common.py

@@ -4,7 +4,11 @@ import dataclasses
 import enum
 import struct
 
+from spacepackets.ecss.tc import PacketId, PacketType
+
 EXAMPLE_PUS_APID = 0x02
+EXAMPLE_PUS_PACKET_ID_TM = PacketId(PacketType.TM, True, EXAMPLE_PUS_APID)
+TM_PACKET_IDS = [EXAMPLE_PUS_PACKET_ID_TM]
 
 
 class EventSeverity(enum.IntEnum):

satrs-example/pyclient/main.py

@@ -14,7 +14,7 @@ from spacepackets.ccsds.time import CdsShortTimestamp
 from tmtccmd import CcsdsTmtcBackend, TcHandlerBase, ProcedureParamsWrapper
 from tmtccmd.core.base import BackendRequest
 from tmtccmd.pus import VerificationWrapper
-from tmtccmd.tm import CcsdsTmHandler, SpecificApidHandlerBase
+from tmtccmd.tmtc import CcsdsTmHandler, SpecificApidHandlerBase
 from tmtccmd.com import ComInterface
 from tmtccmd.config import (
     default_json_path,
@@ -30,7 +30,7 @@ from tmtccmd.logging.pus import (
     RawTmtcTimedLogWrapper,
     TimedLogWhen,
 )
-from tmtccmd.tc import (
+from tmtccmd.tmtc import (
     TcQueueEntryType,
     ProcedureWrapper,
     TcProcedureType,
@@ -45,7 +45,7 @@ from tmtccmd.util.obj_id import ObjectIdDictT
 
 import pus_tc
 import tc_definitions
-from common import EXAMPLE_PUS_APID, EventU32
+from common import EXAMPLE_PUS_APID, TM_PACKET_IDS, EventU32
 
 _LOGGER = logging.getLogger()
 
@@ -63,7 +63,7 @@ class SatRsConfigHook(HookBase):
         cfg = create_com_interface_cfg_default(
             com_if_key=com_if_key,
             json_cfg_path=self.cfg_path,
-            space_packet_ids=None,
+            space_packet_ids=TM_PACKET_IDS,
         )
         return create_com_interface_default(cfg)
 
@@ -128,6 +128,7 @@ class PusHandler(SpecificApidHandlerBase):
                 if len(pus_tm.source_data) < 8:
                     raise ValueError("No addressable ID in HK packet")
                 json_str = pus_tm.source_data[8:]
+                _LOGGER.info(json_str)
             dedicated_handler = True
         if service == 5:
             tm_packet = PusTelemetry.unpack(

satrs-example/pyclient/pus_tc.py

@@ -3,9 +3,9 @@ import datetime
 from spacepackets.ccsds import CdsShortTimestamp
 from spacepackets.ecss import PusTelecommand
 from tmtccmd.config import CoreServiceList
-from tmtccmd.tc import DefaultPusQueueHelper
-from tmtccmd.tc.pus_11_tc_sched import create_time_tagged_cmd
-from tmtccmd.tc.pus_3_fsfw_hk import create_request_one_hk_command
+from tmtccmd.tmtc import DefaultPusQueueHelper
+from tmtccmd.pus.s11_tc_sched import create_time_tagged_cmd
+from tmtccmd.pus.tc.s3_fsfw_hk import create_request_one_hk_command
 
 from common import (
     EXAMPLE_PUS_APID,

satrs-example/pyclient/requirements.txt

@@ -1,2 +1,2 @@
-tmtccmd == 5.0.0rc0
+tmtccmd == 7.0.0
 # -e git+https://github.com/robamu-org/tmtccmd@97e5e51101a08b21472b3ddecc2063359f7e307a#egg=tmtccmd

satrs-example/pyclient/tmtc_conf.json

@@ -1,6 +1,8 @@
 {
-    "com_if": "udp",
+    "com_if": "tcp",
     "tcpip_udp_ip_addr": "127.0.0.1",
     "tcpip_udp_port": 7301,
-    "tcpip_udp_recv_max_size": 1500
-}
+    "tcpip_udp_recv_max_size": 1500,
+    "tcpip_tcp_ip_addr": "127.0.0.1",
+    "tcpip_tcp_port": 7301
+}

satrs-example/src/ccsds.rs

@@ -3,6 +3,7 @@ use satrs_core::spacepackets::{CcsdsPacket, SpHeader};
 use satrs_core::tmtc::{CcsdsPacketHandler, ReceivesCcsdsTc};
 use satrs_example::PUS_APID;
 
+#[derive(Clone)]
 pub struct CcsdsReceiver {
     pub tc_source: PusTcSource,
 }

satrs-example/src/main.rs

@@ -3,10 +3,15 @@ mod hk;
 mod logging;
 mod pus;
 mod requests;
+mod tcp;
 mod tmtc;
+mod udp;
 
 use log::{info, warn};
+use satrs_core::hal::std::tcp_server::ServerConfig;
+use satrs_core::hal::std::udp_server::UdpTcServer;
 
+use crate::ccsds::CcsdsReceiver;
 use crate::hk::AcsHkIds;
 use crate::logging::setup_logger;
 use crate::pus::action::{Pus8Wrapper, PusService8ActionHandler};
@@ -14,9 +19,11 @@ use crate::pus::event::Pus5Wrapper;
 use crate::pus::hk::{Pus3Wrapper, PusService3HkHandler};
 use crate::pus::scheduler::Pus11Wrapper;
 use crate::pus::test::Service17CustomWrapper;
-use crate::pus::PusTcMpscRouter;
+use crate::pus::{PusReceiver, PusTcMpscRouter};
 use crate::requests::{Request, RequestWithToken};
-use crate::tmtc::{core_tmtc_task, PusTcSource, TcArgs, TcStore, TmArgs, TmFunnel};
+use crate::tcp::{SyncTcpTmSource, TcpTask};
+use crate::tmtc::{PusTcSource, TcArgs, TcStore, TmArgs, TmFunnel, TmtcTask};
+use crate::udp::UdpTmtcServer;
 use satrs_core::event_man::{
     EventManagerWithMpscQueue, MpscEventReceiver, MpscEventU32SendProvider, SendEventProvider,
 };
@@ -43,7 +50,7 @@ use satrs_core::spacepackets::{
     SpHeader,
 };
 use satrs_core::tmtc::tm_helper::SharedTmStore;
-use satrs_core::tmtc::{AddressableId, TargetId};
+use satrs_core::tmtc::{AddressableId, CcsdsDistributor, TargetId};
 use satrs_core::ChannelId;
 use satrs_example::{
     RequestTargetId, TcReceiverId, TmSenderId, OBSW_SERVER_ADDR, PUS_APID, SERVER_PORT,
@@ -139,7 +146,7 @@ fn main() {
     let tm_args = TmArgs {
         tm_store: shared_tm_store.clone(),
         tm_sink_sender: tm_funnel_tx.clone(),
-        tm_server_rx,
+        tm_udp_server_rx: tm_server_rx,
     };
 
     let aocs_tm_funnel = tm_funnel_tx.clone();
@@ -266,11 +273,50 @@ fn main() {
     );
     let mut pus_3_wrapper = Pus3Wrapper { pus_3_handler };
 
-    info!("Starting TMTC task");
-    let jh0 = thread::Builder::new()
-        .name("TMTC".to_string())
+    let ccsds_receiver = CcsdsReceiver {
+        tc_source: tc_args.tc_source.clone(),
+    };
+    let mut tmtc_task = TmtcTask::new(tc_args, PusReceiver::new(verif_reporter, pus_router));
+
+    let udp_ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver.clone()));
+    let udp_tc_server = UdpTcServer::new(sock_addr, 2048, Box::new(udp_ccsds_distributor))
+        .expect("creating UDP TMTC server failed");
+    let mut udp_tmtc_server = UdpTmtcServer {
+        udp_tc_server,
+        tm_rx: tm_args.tm_udp_server_rx,
+        tm_store: tm_args.tm_store.clone_backing_pool(),
+    };
+
+    info!("Starting TMTC and UDP task");
+    let jh_udp_tmtc = thread::Builder::new()
+        .name("TMTC and UDP".to_string())
         .spawn(move || {
-            core_tmtc_task(sock_addr, tc_args, tm_args, verif_reporter, pus_router);
+            info!("Running UDP server on port {SERVER_PORT}");
+            loop {
+                udp_tmtc_server.periodic_operation();
+                tmtc_task.periodic_operation();
+                thread::sleep(Duration::from_millis(400));
+            }
+        })
+        .unwrap();
+
+    let tcp_ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver));
+    let tcp_server_cfg = ServerConfig::new(sock_addr, Duration::from_millis(400), 4096, 8192);
+    let mut sync_tm_tcp_source = SyncTcpTmSource::new(200);
+    let mut tcp_server = TcpTask::new(
+        tcp_server_cfg,
+        sync_tm_tcp_source.clone(),
+        tcp_ccsds_distributor,
+    )
+    .expect("tcp server creation failed");
+    info!("Starting TCP task");
+    let jh_tcp = thread::Builder::new()
+        .name("TCP".to_string())
+        .spawn(move || {
+            info!("Running TCP server on port {SERVER_PORT}");
+            loop {
+                tcp_server.periodic_operation();
+            }
         })
         .unwrap();
 
@@ -311,6 +357,7 @@ fn main() {
                         .tm_server_tx
                         .send(addr)
                         .expect("Sending TM to server failed");
+                    sync_tm_tcp_source.add_tm(tm_raw);
                 }
             }
         })
@@ -382,6 +429,7 @@ fn main() {
             let mut timestamp: [u8; 7] = [0; 7];
             let mut time_provider = TimeProvider::new_with_u16_days(0, 0);
             loop {
+                // TODO: Move this into a separate function/task/module..
                 match acs_thread_rx.try_recv() {
                     Ok(request) => {
                         info!(
@@ -481,7 +529,12 @@ fn main() {
             thread::sleep(Duration::from_millis(200));
         })
         .unwrap();
-    jh0.join().expect("Joining UDP TMTC server thread failed");
+    jh_udp_tmtc
+        .join()
+        .expect("Joining UDP TMTC server thread failed");
+    jh_tcp
+        .join()
+        .expect("Joining TCP TMTC server thread failed");
     jh1.join().expect("Joining TM Funnel thread failed");
     jh2.join().expect("Joining Event Manager thread failed");
     jh3.join().expect("Joining AOCS thread failed");

satrs-example/src/tcp.rs

@@ -0,0 +1,115 @@
+use std::{
+    collections::VecDeque,
+    sync::{Arc, Mutex},
+};
+
+use log::{info, warn};
+use satrs_core::{
+    hal::std::tcp_server::{ServerConfig, TcpSpacepacketsServer},
+    spacepackets::PacketId,
+    tmtc::{CcsdsDistributor, CcsdsError, TmPacketSourceCore},
+};
+use satrs_example::PUS_APID;
+
+use crate::tmtc::MpscStoreAndSendError;
+
+pub const PACKET_ID_LOOKUP: &[PacketId] = &[PacketId::const_tc(true, PUS_APID)];
+
+#[derive(Default, Clone)]
+pub struct SyncTcpTmSource {
+    tm_queue: Arc<Mutex<VecDeque<Vec<u8>>>>,
+    max_packets_stored: usize,
+    pub silent_packet_overwrite: bool,
+}
+
+impl SyncTcpTmSource {
+    pub fn new(max_packets_stored: usize) -> Self {
+        Self {
+            tm_queue: Arc::default(),
+            max_packets_stored,
+            silent_packet_overwrite: true,
+        }
+    }
+
+    pub fn add_tm(&mut self, tm: &[u8]) {
+        let mut tm_queue = self.tm_queue.lock().expect("locking tm queue failec");
+        if tm_queue.len() > self.max_packets_stored {
+            if !self.silent_packet_overwrite {
+                warn!("TPC TM source is full, deleting oldest packet");
+            }
+            tm_queue.pop_front();
+        }
+        tm_queue.push_back(tm.to_vec());
+    }
+}
+
+impl TmPacketSourceCore for SyncTcpTmSource {
+    type Error = ();
+
+    fn retrieve_packet(&mut self, buffer: &mut [u8]) -> Result<usize, Self::Error> {
+        let mut tm_queue = self.tm_queue.lock().expect("locking tm queue failed");
+        if !tm_queue.is_empty() {
+            let next_vec = tm_queue.front().unwrap();
+            if buffer.len() < next_vec.len() {
+                panic!(
+                    "provided buffer too small, must be at least {} bytes",
+                    next_vec.len()
+                );
+            }
+            let next_vec = tm_queue.pop_front().unwrap();
+            buffer[0..next_vec.len()].copy_from_slice(&next_vec);
+            if next_vec.len() > 9 {
+                let service = next_vec[7];
+                let subservice = next_vec[8];
+                info!("Sending PUS TM[{service},{subservice}]")
+            } else {
+                info!("Sending PUS TM");
+            }
+            return Ok(next_vec.len());
+        }
+        Ok(0)
+    }
+}
+
+pub struct TcpTask {
+    server: TcpSpacepacketsServer<
+        (),
+        CcsdsError<MpscStoreAndSendError>,
+        SyncTcpTmSource,
+        CcsdsDistributor<MpscStoreAndSendError>,
+    >,
+}
+
+impl TcpTask {
+    pub fn new(
+        cfg: ServerConfig,
+        tm_source: SyncTcpTmSource,
+        tc_receiver: CcsdsDistributor<MpscStoreAndSendError>,
+    ) -> Result<Self, std::io::Error> {
+        Ok(Self {
+            server: TcpSpacepacketsServer::new(
+                cfg,
+                tm_source,
+                tc_receiver,
+                Box::new(PACKET_ID_LOOKUP),
+            )?,
+        })
+    }
+
+    pub fn periodic_operation(&mut self) {
+        loop {
+            let result = self.server.handle_next_connection();
+            match result {
+                Ok(conn_result) => {
+                    info!(
+                        "Served {} TMs and {} TCs for client {:?}",
+                        conn_result.num_sent_tms, conn_result.num_received_tcs, conn_result.addr
+                    );
+                }
+                Err(e) => {
+                    warn!("TCP server error: {e:?}");
+                }
+            }
+        }
+    }
+}

satrs-example/src/tmtc.rs

@@ -1,26 +1,21 @@
-use log::{info, warn};
-use satrs_core::hal::std::udp_server::{ReceiveResult, UdpTcServer};
-use std::net::SocketAddr;
+use log::warn;
+use satrs_core::pus::ReceivesEcssPusTc;
+use satrs_core::spacepackets::SpHeader;
 use std::sync::mpsc::{Receiver, SendError, Sender, TryRecvError};
-use std::thread;
-use std::time::Duration;
 use thiserror::Error;
 
-use crate::ccsds::CcsdsReceiver;
-use crate::pus::{PusReceiver, PusTcMpscRouter};
+use crate::pus::PusReceiver;
 use satrs_core::pool::{SharedPool, StoreAddr, StoreError};
-use satrs_core::pus::verification::StdVerifReporterWithSender;
-use satrs_core::pus::{ReceivesEcssPusTc, TcAddrWithToken};
+use satrs_core::pus::TcAddrWithToken;
 use satrs_core::spacepackets::ecss::tc::PusTcReader;
 use satrs_core::spacepackets::ecss::PusPacket;
-use satrs_core::spacepackets::SpHeader;
 use satrs_core::tmtc::tm_helper::SharedTmStore;
-use satrs_core::tmtc::{CcsdsDistributor, CcsdsError, ReceivesCcsdsTc};
+use satrs_core::tmtc::ReceivesCcsdsTc;
 
 pub struct TmArgs {
     pub tm_store: SharedTmStore,
     pub tm_sink_sender: Sender<StoreAddr>,
-    pub tm_server_rx: Receiver<StoreAddr>,
+    pub tm_udp_server_rx: Receiver<StoreAddr>,
 }
 
 pub struct TcArgs {
@@ -64,12 +59,6 @@ pub struct TmFunnel {
     pub tm_server_tx: Sender<StoreAddr>,
 }
 
-pub struct UdpTmtcServer {
-    udp_tc_server: UdpTcServer<CcsdsError<MpscStoreAndSendError>>,
-    tm_rx: Receiver<StoreAddr>,
-    tm_store: SharedPool,
-}
-
 #[derive(Clone)]
 pub struct PusTcSource {
     pub tc_source: Sender<StoreAddr>,
@@ -98,131 +87,60 @@ impl ReceivesCcsdsTc for PusTcSource {
     }
 }
 
-pub fn core_tmtc_task(
-    socket_addr: SocketAddr,
-    mut tc_args: TcArgs,
-    tm_args: TmArgs,
-    verif_reporter: StdVerifReporterWithSender,
-    pus_router: PusTcMpscRouter,
-) {
-    let mut pus_receiver = PusReceiver::new(verif_reporter, pus_router);
-
-    let ccsds_receiver = CcsdsReceiver {
-        tc_source: tc_args.tc_source.clone(),
-    };
-
-    let ccsds_distributor = CcsdsDistributor::new(Box::new(ccsds_receiver));
-
-    let udp_tc_server = UdpTcServer::new(socket_addr, 2048, Box::new(ccsds_distributor))
-        .expect("creating UDP TMTC server failed");
-
-    let mut udp_tmtc_server = UdpTmtcServer {
-        udp_tc_server,
-        tm_rx: tm_args.tm_server_rx,
-        tm_store: tm_args.tm_store.clone_backing_pool(),
-    };
-
-    let mut tc_buf: [u8; 4096] = [0; 4096];
-    loop {
-        core_tmtc_loop(
-            &mut udp_tmtc_server,
-            &mut tc_args,
-            &mut tc_buf,
-            &mut pus_receiver,
-        );
-        thread::sleep(Duration::from_millis(400));
-    }
+pub struct TmtcTask {
+    tc_args: TcArgs,
+    tc_buf: [u8; 4096],
+    pus_receiver: PusReceiver,
 }
 
-fn core_tmtc_loop(
-    udp_tmtc_server: &mut UdpTmtcServer,
-    tc_args: &mut TcArgs,
-    tc_buf: &mut [u8],
-    pus_receiver: &mut PusReceiver,
-) {
-    while poll_tc_server(udp_tmtc_server) {}
-    match tc_args.tc_receiver.try_recv() {
-        Ok(addr) => {
-            let pool = tc_args
-                .tc_source
-                .tc_store
-                .pool
-                .read()
-                .expect("locking tc pool failed");
-            let data = pool.read(&addr).expect("reading pool failed");
-            tc_buf[0..data.len()].copy_from_slice(data);
-            drop(pool);
-            match PusTcReader::new(tc_buf) {
-                Ok((pus_tc, _)) => {
-                    pus_receiver
-                        .handle_tc_packet(addr, pus_tc.service(), &pus_tc)
-                        .ok();
-                }
-                Err(e) => {
-                    warn!("error creating PUS TC from raw data: {e}");
-                    warn!("raw data: {tc_buf:x?}");
-                }
-            }
-        }
-        Err(e) => {
-            if let TryRecvError::Disconnected = e {
-                warn!("tmtc thread: sender disconnected")
-            }
+impl TmtcTask {
+    pub fn new(tc_args: TcArgs, pus_receiver: PusReceiver) -> Self {
+        Self {
+            tc_args,
+            tc_buf: [0; 4096],
+            pus_receiver,
         }
     }
-    if let Some(recv_addr) = udp_tmtc_server.udp_tc_server.last_sender() {
-        core_tm_handling(udp_tmtc_server, &recv_addr);
-    }
-}
 
-fn poll_tc_server(udp_tmtc_server: &mut UdpTmtcServer) -> bool {
-    match udp_tmtc_server.udp_tc_server.try_recv_tc() {
-        Ok(_) => true,
-        Err(e) => match e {
-            ReceiveResult::ReceiverError(e) => match e {
-                CcsdsError::ByteConversionError(e) => {
-                    warn!("packet error: {e:?}");
-                    true
+    pub fn periodic_operation(&mut self) {
+        //while self.poll_tc() {}
+        self.poll_tc();
+    }
+
+    pub fn poll_tc(&mut self) -> bool {
+        match self.tc_args.tc_receiver.try_recv() {
+            Ok(addr) => {
+                let pool = self
+                    .tc_args
+                    .tc_source
+                    .tc_store
+                    .pool
+                    .read()
+                    .expect("locking tc pool failed");
+                let data = pool.read(&addr).expect("reading pool failed");
+                self.tc_buf[0..data.len()].copy_from_slice(data);
+                drop(pool);
+                match PusTcReader::new(&self.tc_buf) {
+                    Ok((pus_tc, _)) => {
+                        self.pus_receiver
+                            .handle_tc_packet(addr, pus_tc.service(), &pus_tc)
+                            .ok();
+                        true
+                    }
+                    Err(e) => {
+                        warn!("error creating PUS TC from raw data: {e}");
+                        warn!("raw data: {:x?}", self.tc_buf);
+                        true
+                    }
                 }
-                CcsdsError::CustomError(e) => {
-                    warn!("mpsc store and send error {e:?}");
-                    true
+            }
+            Err(e) => match e {
+                TryRecvError::Empty => false,
+                TryRecvError::Disconnected => {
+                    warn!("tmtc thread: sender disconnected");
+                    false
                 }
             },
-            ReceiveResult::IoError(e) => {
-                warn!("IO error {e}");
-                false
-            }
-            ReceiveResult::NothingReceived => false,
-        },
-    }
-}
-
-fn core_tm_handling(udp_tmtc_server: &mut UdpTmtcServer, recv_addr: &SocketAddr) {
-    while let Ok(addr) = udp_tmtc_server.tm_rx.try_recv() {
-        let store_lock = udp_tmtc_server.tm_store.write();
-        if store_lock.is_err() {
-            warn!("Locking TM store failed");
-            continue;
-        }
-        let mut store_lock = store_lock.unwrap();
-        let pg = store_lock.read_with_guard(addr);
-        let read_res = pg.read();
-        if read_res.is_err() {
-            warn!("Error reading TM pool data");
-            continue;
-        }
-        let buf = read_res.unwrap();
-        if buf.len() > 9 {
-            let service = buf[7];
-            let subservice = buf[8];
-            info!("Sending PUS TM[{service},{subservice}]")
-        } else {
-            info!("Sending PUS TM");
-        }
-        let result = udp_tmtc_server.udp_tc_server.socket.send_to(buf, recv_addr);
-        if let Err(e) = result {
-            warn!("Sending TM with UDP socket failed: {e}")
         }
     }
 }

satrs-example/src/udp.rs

@@ -0,0 +1,76 @@
+use std::{net::SocketAddr, sync::mpsc::Receiver};
+
+use log::{info, warn};
+use satrs_core::{
+    hal::std::udp_server::{ReceiveResult, UdpTcServer},
+    pool::{SharedPool, StoreAddr},
+    tmtc::CcsdsError,
+};
+
+use crate::tmtc::MpscStoreAndSendError;
+
+pub struct UdpTmtcServer {
+    pub udp_tc_server: UdpTcServer<CcsdsError<MpscStoreAndSendError>>,
+    pub tm_rx: Receiver<StoreAddr>,
+    pub tm_store: SharedPool,
+}
+impl UdpTmtcServer {
+    pub fn periodic_operation(&mut self) {
+        while self.poll_tc_server() {}
+        if let Some(recv_addr) = self.udp_tc_server.last_sender() {
+            self.send_tm_to_udp_client(&recv_addr);
+        }
+    }
+
+    fn poll_tc_server(&mut self) -> bool {
+        match self.udp_tc_server.try_recv_tc() {
+            Ok(_) => true,
+            Err(e) => match e {
+                ReceiveResult::ReceiverError(e) => match e {
+                    CcsdsError::ByteConversionError(e) => {
+                        warn!("packet error: {e:?}");
+                        true
+                    }
+                    CcsdsError::CustomError(e) => {
+                        warn!("mpsc store and send error {e:?}");
+                        true
+                    }
+                },
+                ReceiveResult::IoError(e) => {
+                    warn!("IO error {e}");
+                    false
+                }
+                ReceiveResult::NothingReceived => false,
+            },
+        }
+    }
+
+    fn send_tm_to_udp_client(&mut self, recv_addr: &SocketAddr) {
+        while let Ok(addr) = self.tm_rx.try_recv() {
+            let store_lock = self.tm_store.write();
+            if store_lock.is_err() {
+                warn!("Locking TM store failed");
+                continue;
+            }
+            let mut store_lock = store_lock.unwrap();
+            let pg = store_lock.read_with_guard(addr);
+            let read_res = pg.read();
+            if read_res.is_err() {
+                warn!("Error reading TM pool data");
+                continue;
+            }
+            let buf = read_res.unwrap();
+            if buf.len() > 9 {
+                let service = buf[7];
+                let subservice = buf[8];
+                info!("Sending PUS TM[{service},{subservice}]")
+            } else {
+                info!("Sending PUS TM");
+            }
+            let result = self.udp_tc_server.socket.send_to(buf, recv_addr);
+            if let Err(e) = result {
+                warn!("Sending TM with UDP socket failed: {e}")
+            }
+        }
+    }
+}

satrs-mib/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "satrs-mib"
-version = "0.1.0-alpha.0"
+version = "0.1.0-alpha.1"
 edition = "2021"
 rust-version = "1.61"
 authors = ["Robin Mueller <muellerr@irs.uni-stuttgart.de>"]
@@ -23,14 +23,14 @@ version = "1"
 optional = true
 
 [dependencies.satrs-core]
-# version = "0.1.0-alpha.1"
-git = "https://egit.irs.uni-stuttgart.de/rust/sat-rs.git"
-branch = "main"
+version = "0.1.0-alpha.1"
+# git = "https://egit.irs.uni-stuttgart.de/rust/sat-rs.git"
+# branch = "main"
 # rev = "35e1f7a983f6535c5571186e361fe101d4306b89"
 
 [dependencies.satrs-mib-codegen]
 path = "codegen"
-version = "0.1.0-alpha.0"
+version = "0.1.0-alpha.1"
 
 [dependencies.serde]
 version = "1"

satrs-mib/codegen/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "satrs-mib-codegen"
-version = "0.1.0-alpha.0"
+version = "0.1.0-alpha.1"
 edition = "2021"
 description = "satrs-mib proc macro implementation"
 homepage = "https://egit.irs.uni-stuttgart.de/rust/sat-rs"
@@ -20,9 +20,9 @@ quote = "1"
 proc-macro2 = "1"
 
 [dependencies.satrs-core]
-# version = "0.1.0-alpha.1"
-git = "https://egit.irs.uni-stuttgart.de/rust/sat-rs.git"
-branch = "main"
+version = "0.1.0-alpha.1"
+# git = "https://egit.irs.uni-stuttgart.de/rust/sat-rs.git"
+# branch = "main"
 # rev = "35e1f7a983f6535c5571186e361fe101d4306b89"
 
 [dev-dependencies]