Compare commits

...

4 Commits

Author SHA1 Message Date
Robin Mueller
3f4c0e7d51 try to make it compile again 2025-11-27 18:11:14 +01:00
Robin Mueller
b3a7f3f9d2 this is a big refactoring 2025-11-27 16:58:55 +01:00
Robin Mueller
e1bf81e804 Merge remote-tracking branch 'origin/main' into move-to-ccsds-and-serde 2025-11-27 16:16:26 +01:00
fef8d6a792 start moving to CCSDS + serde 2025-11-17 11:13:49 +01:00
18 changed files with 396 additions and 381 deletions

View File

@@ -23,23 +23,17 @@ derive-new = "0.7"
cfg-if = "1"
arbitrary-int = "2"
bitbybit = "1.4"
postcard = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
[dependencies.satrs]
path = "../satrs"
features = ["test_util"]
[dependencies.satrs-minisim]
path = "../satrs-minisim"
[dependencies.satrs-mib]
version = "0.1.1"
path = "../satrs-mib"
satrs = { path = "../satrs", features = ["test_util"] }
satrs-minisim = { path = "../satrs-minisim" }
satrs-mib = { path = "../satrs-mib" }
[features]
default = ["heap_tmtc"]
heap_tmtc = []
# default = ["heap_tmtc"]
# heap_tmtc = []
[dev-dependencies]
env_logger = "0.11"

View File

@@ -1,3 +1,4 @@
/*
use derive_new::new;
use satrs::hk::{HkRequest, HkRequestVariant};
use satrs::mode_tree::{ModeChild, ModeNode};
@@ -717,3 +718,4 @@ mod tests {
assert!(mgm_set.valid);
}
}
*/

View File

@@ -22,7 +22,7 @@ use satrs::{
use satrs_example::{
config::components::NO_SENDER,
ids::{eps::PCDU, generic_pus::PUS_MODE},
DeviceMode, TimestampHelper,
CcsdsTmPacketOwned, DeviceMode, TimestampHelper,
};
use satrs_minisim::{
eps::{
@@ -32,12 +32,7 @@ use satrs_minisim::{
};
use serde::{Deserialize, Serialize};
use crate::{
hk::PusHkHelper,
pus::hk::{HkReply, HkReplyVariant},
requests::CompositeRequest,
tmtc::sender::TmTcSender,
};
use crate::{hk::PusHkHelper, requests::CompositeRequest};
pub trait SerialInterface {
type Error: core::fmt::Debug;
@@ -210,9 +205,10 @@ pub struct PcduHandler<ComInterface: SerialInterface> {
dev_str: &'static str,
mode_node: ModeRequestHandlerMpscBounded,
composite_request_rx: mpsc::Receiver<GenericMessage<CompositeRequest>>,
hk_reply_tx: mpsc::SyncSender<GenericMessage<HkReply>>,
//hk_reply_tx: mpsc::SyncSender<GenericMessage<HkReply>>,
hk_tx: std::sync::mpsc::SyncSender<CcsdsTmPacketOwned>,
switch_request_rx: mpsc::Receiver<GenericMessage<SwitchRequest>>,
tm_sender: TmTcSender,
tm_tx: mpsc::SyncSender<CcsdsTmPacketOwned>,
pub com_interface: ComInterface,
shared_switch_map: Arc<Mutex<SwitchSet>>,
#[new(value = "PusHkHelper::new(id)")]
@@ -273,7 +269,7 @@ impl<ComInterface: SerialInterface> PcduHandler<ComInterface> {
}
}
pub fn handle_hk_request(&mut self, requestor_info: &MessageMetadata, hk_request: &HkRequest) {
pub fn handle_hk_request(&mut self, _requestor_info: &MessageMetadata, hk_request: &HkRequest) {
match hk_request.variant {
HkRequestVariant::OneShot => {
if hk_request.unique_id == SetId::SwitcherSet as u32 {
@@ -300,8 +296,10 @@ impl<ComInterface: SerialInterface> PcduHandler<ComInterface> {
},
&mut self.tm_buf,
) {
// TODO: Fix
/*
self.tm_sender
.send_tm(self.id.id(), PusTmVariant::Direct(hk_tm))
.send(self.id.id(), PusTmVariant::Direct(hk_tm))
.expect("failed to send HK TM");
self.hk_reply_tx
.send(GenericMessage::new(
@@ -309,6 +307,7 @@ impl<ComInterface: SerialInterface> PcduHandler<ComInterface> {
HkReply::new(hk_request.unique_id, HkReplyVariant::Ack),
))
.expect("failed to send HK reply");
*/
}
}
}
@@ -507,9 +506,7 @@ mod tests {
use std::sync::mpsc;
use arbitrary_int::u21;
use satrs::{
mode::ModeRequest, power::SwitchStateBinary, request::GenericMessage, tmtc::PacketAsVec,
};
use satrs::{mode::ModeRequest, power::SwitchStateBinary, request::GenericMessage};
use satrs_example::ids::{self, Apid};
use satrs_minisim::eps::SwitchMapBinary;
@@ -555,8 +552,9 @@ mod tests {
pub mode_reply_rx_to_pus: mpsc::Receiver<GenericMessage<ModeReply>>,
pub mode_reply_rx_to_parent: mpsc::Receiver<GenericMessage<ModeReply>>,
pub composite_request_tx: mpsc::Sender<GenericMessage<CompositeRequest>>,
pub hk_reply_rx: mpsc::Receiver<GenericMessage<HkReply>>,
pub tm_rx: mpsc::Receiver<PacketAsVec>,
//pub hk_reply_rx: mpsc::Receiver<GenericMessage<HkReply>>,
pub hk_rx: std::sync::mpsc::Receiver<CcsdsTmPacketOwned>,
pub tm_rx: mpsc::Receiver<CcsdsTmPacketOwned>,
pub switch_request_tx: mpsc::Sender<GenericMessage<SwitchRequest>>,
pub handler: PcduHandler<SerialInterfaceTest>,
}
@@ -568,8 +566,8 @@ mod tests {
let (mode_reply_tx_to_parent, mode_reply_rx_to_parent) = mpsc::sync_channel(5);
let mode_node = ModeRequestHandlerMpscBounded::new(PCDU.into(), mode_request_rx);
let (composite_request_tx, composite_request_rx) = mpsc::channel();
let (hk_reply_tx, hk_reply_rx) = mpsc::sync_channel(10);
let (tm_tx, tm_rx) = mpsc::sync_channel::<PacketAsVec>(5);
let (hk_tx, hk_rx) = mpsc::sync_channel(10);
let (tm_tx, tm_rx) = mpsc::sync_channel(5);
let (switch_request_tx, switch_reqest_rx) = mpsc::channel();
let shared_switch_map = Arc::new(Mutex::new(SwitchSet::default()));
let mut handler = PcduHandler::new(
@@ -577,9 +575,10 @@ mod tests {
"TEST_PCDU",
mode_node,
composite_request_rx,
hk_reply_tx,
hk_tx,
switch_reqest_rx,
TmTcSender::Heap(tm_tx.clone()),
tm_tx.clone(),
//TmTcSender::Normal(tm_tx.clone()),
SerialInterfaceTest::default(),
shared_switch_map,
);
@@ -590,7 +589,7 @@ mod tests {
mode_reply_rx_to_pus,
mode_reply_rx_to_parent,
composite_request_tx,
hk_reply_rx,
hk_rx,
tm_rx,
switch_request_tx,
handler,

View File

@@ -1,6 +1,6 @@
use std::sync::mpsc::{self};
use crate::pus::create_verification_reporter;
//use crate::pus::create_verification_reporter;
use arbitrary_int::traits::Integer as _;
use arbitrary_int::u11;
use satrs::event_man_legacy::{EventMessageU32, EventRoutingError};
@@ -34,6 +34,7 @@ impl EventTmHook for EventApidSetter {
}
}
/*
/// The PUS event handler subscribes for all events and converts them into ECSS PUS 5 event
/// packets. It also handles the verification completion of PUS event service requests.
pub struct PusEventHandler<TmSender: EcssTmSender> {
@@ -292,3 +293,4 @@ mod tests {
// TODO: Add test.
}
}
*/

View File

@@ -1,22 +1,22 @@
#![allow(dead_code)]
use std::collections::VecDeque;
use std::net::{SocketAddr, UdpSocket};
use std::sync::mpsc;
use std::sync::{mpsc, Arc, Mutex};
use log::{info, warn};
use satrs::hal::std::udp_server::{ReceiveResult, UdpTcServer};
use satrs::pus::HandlingStatus;
use satrs::queue::GenericSendError;
use satrs::tmtc::PacketAsVec;
use satrs::pool::{PoolProviderWithGuards, SharedStaticMemoryPool};
use satrs::tmtc::PacketInPool;
use satrs_example::CcsdsTmPacketOwned;
use crate::tmtc::sender::TmTcSender;
pub trait UdpTmHandler {
pub trait UdpTmHandlerProvider {
fn send_tm_to_udp_client(&mut self, socket: &UdpSocket, recv_addr: &SocketAddr);
}
/*
pub struct StaticUdpTmHandler {
pub tm_rx: mpsc::Receiver<PacketInPool>,
pub tm_store: SharedStaticMemoryPool,
@@ -45,22 +45,17 @@ impl UdpTmHandler for StaticUdpTmHandler {
}
}
}
*/
pub struct DynamicUdpTmHandler {
pub tm_rx: mpsc::Receiver<PacketAsVec>,
pub struct UdpTmHandlerWithChannel {
pub tm_rx: mpsc::Receiver<CcsdsTmPacketOwned>,
}
impl UdpTmHandler for DynamicUdpTmHandler {
impl UdpTmHandlerProvider for UdpTmHandlerWithChannel {
fn send_tm_to_udp_client(&mut self, socket: &UdpSocket, recv_addr: &SocketAddr) {
while let Ok(tm) = self.tm_rx.try_recv() {
if tm.packet.len() > 9 {
let service = tm.packet[7];
let subservice = tm.packet[8];
info!("Sending PUS TM[{service},{subservice}]")
} else {
info!("Sending PUS TM");
}
let result = socket.send_to(&tm.packet, recv_addr);
info!("Sending PUS TM with header {:?}", tm.tm_header);
let result = socket.send_to(&tm.to_vec(), recv_addr);
if let Err(e) = result {
warn!("Sending TM with UDP socket failed: {e}")
}
@@ -68,12 +63,49 @@ impl UdpTmHandler for DynamicUdpTmHandler {
}
}
pub struct UdpTmtcServer<TmHandler: UdpTmHandler> {
pub udp_tc_server: UdpTcServer<TmTcSender, GenericSendError>,
pub tm_handler: TmHandler,
#[derive(Default, Debug, Clone)]
pub struct TestTmHandler {
addrs_to_send_to: Arc<Mutex<VecDeque<SocketAddr>>>,
}
impl<TmHandler: UdpTmHandler> UdpTmtcServer<TmHandler> {
impl UdpTmHandlerProvider for TestTmHandler {
fn send_tm_to_udp_client(&mut self, _socket: &UdpSocket, recv_addr: &SocketAddr) {
self.addrs_to_send_to.lock().unwrap().push_back(*recv_addr);
}
}
pub enum UdpTmHandler {
Normal(UdpTmHandlerWithChannel),
Test(TestTmHandler),
}
impl From<UdpTmHandlerWithChannel> for UdpTmHandler {
fn from(handler: UdpTmHandlerWithChannel) -> Self {
UdpTmHandler::Normal(handler)
}
}
impl From<TestTmHandler> for UdpTmHandler {
fn from(handler: TestTmHandler) -> Self {
UdpTmHandler::Test(handler)
}
}
impl UdpTmHandlerProvider for UdpTmHandler {
fn send_tm_to_udp_client(&mut self, socket: &UdpSocket, recv_addr: &SocketAddr) {
match self {
UdpTmHandler::Normal(handler) => handler.send_tm_to_udp_client(socket, recv_addr),
UdpTmHandler::Test(handler) => handler.send_tm_to_udp_client(socket, recv_addr),
}
}
}
pub struct UdpTmtcServer {
pub udp_tc_server: UdpTcServer<TmTcSender, GenericSendError>,
pub tm_handler: UdpTmHandler,
}
impl UdpTmtcServer {
pub fn periodic_operation(&mut self) {
loop {
if self.poll_tc_server() == HandlingStatus::Empty {
@@ -107,12 +139,8 @@ impl<TmHandler: UdpTmHandler> UdpTmtcServer<TmHandler> {
#[cfg(test)]
mod tests {
use std::net::IpAddr;
use std::net::Ipv4Addr;
use std::{
collections::VecDeque,
net::IpAddr,
sync::{Arc, Mutex},
};
use arbitrary_int::traits::Integer as _;
use arbitrary_int::u14;
@@ -127,23 +155,12 @@ mod tests {
use satrs_example::config::OBSW_SERVER_ADDR;
use satrs_example::ids;
use crate::tmtc::sender::{MockSender, TmTcSender};
use crate::tmtc::sender::MockSender;
use super::*;
const UDP_SERVER_ID: ComponentId = 0x05;
#[derive(Default, Debug, Clone)]
pub struct TestTmHandler {
addrs_to_send_to: Arc<Mutex<VecDeque<SocketAddr>>>,
}
impl UdpTmHandler for TestTmHandler {
fn send_tm_to_udp_client(&mut self, _socket: &UdpSocket, recv_addr: &SocketAddr) {
self.addrs_to_send_to.lock().unwrap().push_back(*recv_addr);
}
}
#[test]
fn test_basic() {
let sock_addr = SocketAddr::new(IpAddr::V4(OBSW_SERVER_ADDR), 0);
@@ -154,7 +171,7 @@ mod tests {
let tm_handler_calls = tm_handler.addrs_to_send_to.clone();
let mut udp_dyn_server = UdpTmtcServer {
udp_tc_server,
tm_handler,
tm_handler: tm_handler.into(),
};
udp_dyn_server.periodic_operation();
let queue = udp_dyn_server
@@ -179,7 +196,7 @@ mod tests {
let tm_handler_calls = tm_handler.addrs_to_send_to.clone();
let mut udp_dyn_server = UdpTmtcServer {
udp_tc_server,
tm_handler,
tm_handler: tm_handler.into(),
};
let sph = SpHeader::new_for_unseg_tc(ids::Apid::GenericPus.raw_value(), u14::ZERO, 0);
let ping_tc = PusTcCreator::new_simple(

View File

@@ -39,6 +39,7 @@ pub fn create_verification_reporter(owner_id: ComponentId, apid: Apid) -> Verifi
VerificationReporter::new(owner_id, &verif_cfg)
}
/*
/// Simple router structure which forwards PUS telecommands to dedicated handlers.
pub struct PusTcMpscRouter {
pub test_tc_sender: mpsc::SyncSender<EcssTcAndToken>,
@@ -187,6 +188,7 @@ impl PusTcDistributor {
Ok(HandlingStatus::HandledOne)
}
}
*/
pub trait TargetedPusService {
const SERVICE_ID: u8;

View File

@@ -1,8 +1,129 @@
use satrs::spacepackets::time::cds::CdsTime;
extern crate alloc;
use satrs::spacepackets::{
ccsds_packet_len_for_user_data_len_with_checksum, time::cds::CdsTime, CcsdsPacketCreationError,
CcsdsPacketCreatorWithReservedData, CcsdsPacketIdAndPsc, SpacePacketHeader,
};
pub mod config;
pub mod ids;
#[derive(
Debug,
Copy,
Clone,
PartialEq,
Eq,
Hash,
serde::Serialize,
serde::Deserialize,
num_enum::TryFromPrimitive,
num_enum::IntoPrimitive,
)]
#[repr(u64)]
pub enum ComponentId {
Pcdu,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum MessageType {
Ping,
Mode,
Hk,
Action,
Verification,
}
/// Unserialized owned TM packet which can be cloned and sent around.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CcsdsTmPacketOwned {
pub sp_header: SpacePacketHeader,
pub tm_header: TmHeader,
pub payload: alloc::vec::Vec<u8>,
}
/// Simple type modelling packet stored in the heap. This structure is intended to
/// be used when sending a packet via a message queue, so it also contains the sender ID.
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct PacketAsVec {
pub sender_id: ComponentId,
pub packet: Vec<u8>,
}
impl PacketAsVec {
pub fn new(sender_id: ComponentId, packet: Vec<u8>) -> Self {
Self { sender_id, packet }
}
}
#[derive(Debug, thiserror::Error)]
pub enum CcsdsCreationError {
#[error("CCSDS packet creation error: {0}")]
CcsdsPacketCreation(#[from] CcsdsPacketCreationError),
#[error("postcard error: {0}")]
Postcard(#[from] postcard::Error),
#[error("timestamp generation error")]
Time,
}
impl CcsdsTmPacketOwned {
pub fn write_to_bytes(&self, buf: &mut [u8]) -> Result<usize, CcsdsCreationError> {
let response_len =
postcard::experimental::serialized_size(&self.tm_header)? + self.payload.len();
let mut ccsds_tm = CcsdsPacketCreatorWithReservedData::new_tm_with_checksum(
self.sp_header,
response_len,
buf,
)?;
let user_data = ccsds_tm.packet_data_mut();
let ser_len = postcard::to_slice(&self.tm_header, user_data)?.len();
user_data[ser_len..ser_len + self.payload.len()].copy_from_slice(&self.payload);
let ccsds_packet_len = ccsds_tm.finish();
Ok(ccsds_packet_len)
}
pub fn len_written(&self) -> usize {
ccsds_packet_len_for_user_data_len_with_checksum(
postcard::experimental::serialized_size(&self.tm_header).unwrap() as usize
+ postcard::experimental::serialized_size(&self.payload).unwrap() as usize,
)
.unwrap()
}
pub fn to_vec(&self) -> alloc::vec::Vec<u8> {
let mut buf = alloc::vec![0u8; self.len_written()];
let len = self.write_to_bytes(&mut buf).unwrap();
buf.truncate(len);
buf
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub struct TmHeader {
pub sender_id: ComponentId,
pub target_id: ComponentId,
pub message_type: MessageType,
/// Telemetry can either be sent unsolicited, or as a response to telecommands.
pub tc_id: Option<CcsdsPacketIdAndPsc>,
/// Raw CDS short timestamp.
pub timestamp: Option<[u8; 7]>,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub struct TcHeader {
pub target_id: ComponentId,
pub request_type: MessageType,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CcsdsTcPacketOwned {
pub sp_header: SpacePacketHeader,
pub tc_header: TcHeader,
pub payload: alloc::vec::Vec<u8>,
}
#[derive(Debug, PartialEq, Eq, Copy, Clone)]
pub enum DeviceMode {
Off = 0,

View File

@@ -5,12 +5,10 @@ use std::{
time::Duration,
};
use acs::mgm::{MgmHandlerLis3Mdl, SpiDummyInterface, SpiSimInterface, SpiSimInterfaceWrapper};
use eps::{
pcdu::{PcduHandler, SerialInterfaceDummy, SerialInterfaceToSim, SerialSimInterfaceWrapper},
PowerSwitchHelper,
};
use events::EventHandler;
use interface::{
sim_client_udp::create_sim_client,
tcp::{SyncTcpTmSource, TcpTask},
@@ -18,29 +16,17 @@ use interface::{
};
use log::info;
use logger::setup_logger;
use pus::{
action::create_action_service,
event::create_event_service,
hk::create_hk_service,
mode::create_mode_service,
scheduler::{create_scheduler_service, TcReleaser},
stack::PusStack,
test::create_test_service,
PusTcDistributor, PusTcMpscRouter,
};
use requests::GenericRequestRouter;
use satrs::{
hal::std::{tcp_server::ServerConfig, udp_server::UdpTcServer},
mode::{Mode, ModeAndSubmode, ModeRequest, ModeRequestHandlerMpscBounded},
mode_tree::connect_mode_nodes,
pus::{event_man::EventRequestWithToken, EcssTcCacher, HandlingStatus},
pus::{event_man::EventRequestWithToken, HandlingStatus},
request::{GenericMessage, MessageMetadata},
spacepackets::time::cds::CdsTime,
};
use satrs_example::{
config::{
components::NO_SENDER,
pool::create_sched_tc_pool,
tasks::{FREQ_MS_AOCS, FREQ_MS_PUS_STACK, FREQ_MS_UDP_TMTC, SIM_CLIENT_IDLE_DELAY_MS},
OBSW_SERVER_ADDR, PACKET_ID_VALIDATOR, SERVER_PORT,
},
@@ -54,23 +40,7 @@ use satrs_example::{
use tmtc::sender::TmTcSender;
use tmtc::{tc_source::TcSourceTask, tm_sink::TmSink};
cfg_if::cfg_if! {
if #[cfg(feature = "heap_tmtc")] {
use interface::udp::DynamicUdpTmHandler;
use satrs::pus::EcssTcVecCacher;
use tmtc::{tc_source::TcSourceTaskDynamic, tm_sink::TmSinkDynamic};
} else {
use std::sync::RwLock;
use interface::udp::StaticUdpTmHandler;
use satrs::pus::EcssTcInSharedPoolCacher;
use satrs::tmtc::{PacketSenderWithSharedPool, SharedPacketPool};
use satrs_example::config::pool::create_static_pools;
use tmtc::{
tc_source::TcSourceTaskStatic,
tm_sink::TmSinkStatic,
};
}
}
use crate::{interface::udp::UdpTmHandlerWithChannel, tmtc::tc_source::CcsdsDistributor};
mod acs;
mod eps;
@@ -78,7 +48,7 @@ mod events;
mod hk;
mod interface;
mod logger;
mod pus;
//mod pus;
mod requests;
mod spi;
mod tmtc;
@@ -87,33 +57,13 @@ fn main() {
setup_logger().expect("setting up logging with fern failed");
println!("Running OBSW example");
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let (tm_pool, tc_pool) = create_static_pools();
let shared_tm_pool = Arc::new(RwLock::new(tm_pool));
let shared_tc_pool = Arc::new(RwLock::new(tc_pool));
let shared_tm_pool_wrapper = SharedPacketPool::new(&shared_tm_pool);
let shared_tc_pool_wrapper = SharedPacketPool::new(&shared_tc_pool);
}
}
let (tc_source_tx, tc_source_rx) = mpsc::sync_channel(50);
let (tm_sink_tx, tm_sink_rx) = mpsc::sync_channel(50);
let (tm_server_tx, tm_server_rx) = mpsc::sync_channel(50);
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let tm_sender = TmTcSender::Static(
PacketSenderWithSharedPool::new(tm_sink_tx.clone(), shared_tm_pool_wrapper.clone())
);
} else if #[cfg(feature = "heap_tmtc")] {
let tm_sender = TmTcSender::Heap(tm_sink_tx.clone());
}
}
let (sim_request_tx, sim_request_rx) = mpsc::channel();
let (mgm_0_sim_reply_tx, mgm_0_sim_reply_rx) = mpsc::channel();
let (mgm_1_sim_reply_tx, mgm_1_sim_reply_rx) = mpsc::channel();
//let (mgm_0_sim_reply_tx, mgm_0_sim_reply_rx) = mpsc::channel();
//let (mgm_1_sim_reply_tx, mgm_1_sim_reply_rx) = mpsc::channel();
let (pcdu_sim_reply_tx, pcdu_sim_reply_rx) = mpsc::channel();
let mut opt_sim_client = create_sim_client(sim_request_rx);
@@ -136,134 +86,44 @@ fn main() {
.composite_router_map
.insert(PCDU.id(), pcdu_handler_composite_tx);
// This helper structure is used by all telecommand providers which need to send telecommands
// to the TC source.
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let tc_sender_with_shared_pool =
PacketSenderWithSharedPool::new(tc_source_tx, shared_tc_pool_wrapper.clone());
let tc_in_mem_converter =
EcssTcCacher::Static(EcssTcInSharedPoolCacher::new(shared_tc_pool, 4096));
} else if #[cfg(feature = "heap_tmtc")] {
let tc_in_mem_converter = EcssTcCacher::Heap(EcssTcVecCacher::default());
}
}
// Create event handling components
// These sender handles are used to send event requests, for example to enable or disable
// certain events.
let (event_tx, event_rx) = mpsc::sync_channel(100);
//let (event_tx, event_rx) = mpsc::sync_channel(100);
let (event_request_tx, event_request_rx) = mpsc::channel::<EventRequestWithToken>();
// The event task is the core handler to perform the event routing and TM handling as specified
// in the sat-rs documentation.
let mut event_handler = EventHandler::new(tm_sink_tx.clone(), event_rx, event_request_rx);
//let mut event_handler = EventHandler::new(tm_sink_tx.clone(), event_rx, event_request_rx);
let (pus_test_tx, pus_test_rx) = mpsc::sync_channel(20);
let (pus_event_tx, pus_event_rx) = mpsc::sync_channel(10);
let (pus_sched_tx, pus_sched_rx) = mpsc::sync_channel(50);
let (pus_hk_tx, pus_hk_rx) = mpsc::sync_channel(50);
let (pus_action_tx, pus_action_rx) = mpsc::sync_channel(50);
let (pus_mode_tx, pus_mode_rx) = mpsc::sync_channel(50);
//let (pus_test_tx, pus_test_rx) = mpsc::sync_channel(20);
//let (pus_event_tx, pus_event_rx) = mpsc::sync_channel(10);
//let (pus_sched_tx, pus_sched_rx) = mpsc::sync_channel(50);
//let (pus_hk_tx, pus_hk_rx) = mpsc::sync_channel(50);
//let (pus_action_tx, pus_action_rx) = mpsc::sync_channel(50);
//let (pus_mode_tx, pus_mode_rx) = mpsc::sync_channel(50);
let (_pus_action_reply_tx, pus_action_reply_rx) = mpsc::channel();
//let (_pus_action_reply_tx, pus_action_reply_rx) = mpsc::channel();
let (pus_hk_reply_tx, pus_hk_reply_rx) = mpsc::sync_channel(50);
let (pus_mode_reply_tx, pus_mode_reply_rx) = mpsc::sync_channel(30);
//let (pus_mode_reply_tx, pus_mode_reply_rx) = mpsc::sync_channel(30);
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let tc_releaser = TcReleaser::Static(tc_sender_with_shared_pool.clone());
} else if #[cfg(feature = "heap_tmtc")] {
let tc_releaser = TcReleaser::Heap(tc_source_tx.clone());
}
}
let pus_router = PusTcMpscRouter {
test_tc_sender: pus_test_tx,
event_tc_sender: pus_event_tx,
sched_tc_sender: pus_sched_tx,
hk_tc_sender: pus_hk_tx,
action_tc_sender: pus_action_tx,
mode_tc_sender: pus_mode_tx,
let mut ccsds_distributor = CcsdsDistributor::default();
let mut tmtc_task = TcSourceTask::new(
tc_source_rx,
ccsds_distributor,
//PusTcDistributor::new(tm_sender.clone(), pus_router),
);
let tc_sender = TmTcSender::Normal(tc_source_tx.clone());
let udp_tm_handler = UdpTmHandlerWithChannel {
tm_rx: tm_server_rx,
};
let pus_test_service = create_test_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
event_tx.clone(),
pus_test_rx,
);
let pus_scheduler_service = create_scheduler_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
tc_releaser,
pus_sched_rx,
create_sched_tc_pool(),
);
let pus_event_service = create_event_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
pus_event_rx,
event_request_tx,
);
let pus_action_service = create_action_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
pus_action_rx,
request_map.clone(),
pus_action_reply_rx,
);
let pus_hk_service = create_hk_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
pus_hk_rx,
request_map.clone(),
pus_hk_reply_rx,
);
let pus_mode_service = create_mode_service(
tm_sender.clone(),
tc_in_mem_converter.clone(),
pus_mode_rx,
request_map,
pus_mode_reply_rx,
);
let mut pus_stack = PusStack::new(
pus_test_service,
pus_hk_service,
pus_event_service,
pus_action_service,
pus_scheduler_service,
pus_mode_service,
);
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let mut tmtc_task = TcSourceTask::Static(TcSourceTaskStatic::new(
shared_tc_pool_wrapper.clone(),
tc_source_rx,
PusTcDistributor::new(tm_sender.clone(), pus_router),
));
let tc_sender = TmTcSender::Static(tc_sender_with_shared_pool);
let udp_tm_handler = StaticUdpTmHandler {
tm_rx: tm_server_rx,
tm_store: shared_tm_pool.clone(),
};
} else if #[cfg(feature = "heap_tmtc")] {
let mut tmtc_task = TcSourceTask::Heap(TcSourceTaskDynamic::new(
tc_source_rx,
PusTcDistributor::new(tm_sender.clone(), pus_router),
));
let tc_sender = TmTcSender::Heap(tc_source_tx.clone());
let udp_tm_handler = DynamicUdpTmHandler {
tm_rx: tm_server_rx,
};
}
}
let sock_addr = SocketAddr::new(IpAddr::V4(OBSW_SERVER_ADDR), SERVER_PORT);
let udp_tc_server = UdpTcServer::new(UDP_SERVER.id(), sock_addr, 2048, tc_sender.clone())
.expect("creating UDP TMTC server failed");
let mut udp_tmtc_server = UdpTmtcServer {
udp_tc_server,
tm_handler: udp_tm_handler,
tm_handler: udp_tm_handler.into(),
};
let tcp_server_cfg = ServerConfig::new(
@@ -282,88 +142,75 @@ fn main() {
)
.expect("tcp server creation failed");
cfg_if::cfg_if! {
if #[cfg(not(feature = "heap_tmtc"))] {
let mut tm_sink = TmSink::Static(TmSinkStatic::new(
shared_tm_pool_wrapper,
sync_tm_tcp_source,
tm_sink_rx,
tm_server_tx,
));
} else if #[cfg(feature = "heap_tmtc")] {
let mut tm_sink = TmSink::Heap(TmSinkDynamic::new(
sync_tm_tcp_source,
tm_sink_rx,
tm_server_tx,
));
}
}
let mut tm_sink = TmSink::new(sync_tm_tcp_source, tm_sink_rx, tm_server_tx);
let shared_switch_set = Arc::new(Mutex::default());
let (switch_request_tx, switch_request_rx) = mpsc::sync_channel(20);
let switch_helper = PowerSwitchHelper::new(switch_request_tx, shared_switch_set.clone());
let shared_mgm_0_set = Arc::default();
let shared_mgm_1_set = Arc::default();
//let shared_mgm_0_set = Arc::default();
//let shared_mgm_1_set = Arc::default();
let mgm_0_mode_node = ModeRequestHandlerMpscBounded::new(MGM0.into(), mgm_0_handler_mode_rx);
let mgm_1_mode_node = ModeRequestHandlerMpscBounded::new(MGM1.into(), mgm_1_handler_mode_rx);
let (mgm_0_spi_interface, mgm_1_spi_interface) =
if let Some(sim_client) = opt_sim_client.as_mut() {
sim_client
.add_reply_recipient(satrs_minisim::SimComponent::Mgm0Lis3Mdl, mgm_0_sim_reply_tx);
sim_client
.add_reply_recipient(satrs_minisim::SimComponent::Mgm1Lis3Mdl, mgm_1_sim_reply_tx);
(
SpiSimInterfaceWrapper::Sim(SpiSimInterface {
sim_request_tx: sim_request_tx.clone(),
sim_reply_rx: mgm_0_sim_reply_rx,
}),
SpiSimInterfaceWrapper::Sim(SpiSimInterface {
sim_request_tx: sim_request_tx.clone(),
sim_reply_rx: mgm_1_sim_reply_rx,
}),
)
} else {
(
SpiSimInterfaceWrapper::Dummy(SpiDummyInterface::default()),
SpiSimInterfaceWrapper::Dummy(SpiDummyInterface::default()),
)
};
let mut mgm_0_handler = MgmHandlerLis3Mdl::new(
MGM0,
"MGM_0",
mgm_0_mode_node,
mgm_0_handler_composite_rx,
pus_hk_reply_tx.clone(),
switch_helper.clone(),
tm_sender.clone(),
mgm_0_spi_interface,
shared_mgm_0_set,
);
let mut mgm_1_handler = MgmHandlerLis3Mdl::new(
MGM1,
"MGM_1",
mgm_1_mode_node,
mgm_1_handler_composite_rx,
pus_hk_reply_tx.clone(),
switch_helper.clone(),
tm_sender.clone(),
mgm_1_spi_interface,
shared_mgm_1_set,
);
// Connect PUS service to device handlers.
connect_mode_nodes(
&mut pus_stack.mode_srv,
mgm_0_handler_mode_tx,
&mut mgm_0_handler,
pus_mode_reply_tx.clone(),
);
connect_mode_nodes(
&mut pus_stack.mode_srv,
mgm_1_handler_mode_tx,
&mut mgm_1_handler,
pus_mode_reply_tx.clone(),
);
/*
let (mgm_0_spi_interface, mgm_1_spi_interface) =
if let Some(sim_client) = opt_sim_client.as_mut() {
sim_client
.add_reply_recipient(satrs_minisim::SimComponent::Mgm0Lis3Mdl, mgm_0_sim_reply_tx);
sim_client
.add_reply_recipient(satrs_minisim::SimComponent::Mgm1Lis3Mdl, mgm_1_sim_reply_tx);
(
SpiSimInterfaceWrapper::Sim(SpiSimInterface {
sim_request_tx: sim_request_tx.clone(),
sim_reply_rx: mgm_0_sim_reply_rx,
}),
SpiSimInterfaceWrapper::Sim(SpiSimInterface {
sim_request_tx: sim_request_tx.clone(),
sim_reply_rx: mgm_1_sim_reply_rx,
}),
)
} else {
(
SpiSimInterfaceWrapper::Dummy(SpiDummyInterface::default()),
SpiSimInterfaceWrapper::Dummy(SpiDummyInterface::default()),
)
};
let mut mgm_0_handler = MgmHandlerLis3Mdl::new(
MGM0,
"MGM_0",
mgm_0_mode_node,
mgm_0_handler_composite_rx,
pus_hk_reply_tx.clone(),
switch_helper.clone(),
tm_sender.clone(),
mgm_0_spi_interface,
shared_mgm_0_set,
);
let mut mgm_1_handler = MgmHandlerLis3Mdl::new(
MGM1,
"MGM_1",
mgm_1_mode_node,
mgm_1_handler_composite_rx,
pus_hk_reply_tx.clone(),
switch_helper.clone(),
tm_sender.clone(),
mgm_1_spi_interface,
shared_mgm_1_set,
);
// Connect PUS service to device handlers.
connect_mode_nodes(
&mut pus_stack.mode_srv,
mgm_0_handler_mode_tx,
&mut mgm_0_handler,
pus_mode_reply_tx.clone(),
);
connect_mode_nodes(
&mut pus_stack.mode_srv,
mgm_1_handler_mode_tx,
&mut mgm_1_handler,
pus_mode_reply_tx.clone(),
);
*/
let pcdu_serial_interface = if let Some(sim_client) = opt_sim_client.as_mut() {
sim_client.add_reply_recipient(satrs_minisim::SimComponent::Pcdu, pcdu_sim_reply_tx);
@@ -382,16 +229,18 @@ fn main() {
pcdu_handler_composite_rx,
pus_hk_reply_tx,
switch_request_rx,
tm_sender.clone(),
tm_sink_tx.clone(),
pcdu_serial_interface,
shared_switch_set,
);
/*
connect_mode_nodes(
&mut pus_stack.mode_srv,
pcdu_handler_mode_tx.clone(),
&mut pcdu_handler,
pus_mode_reply_tx,
);
*/
// The PCDU is a critical component which should be in normal mode immediately.
pcdu_handler_mode_tx
@@ -455,8 +304,8 @@ fn main() {
let jh_aocs = thread::Builder::new()
.name("sat-rs aocs".to_string())
.spawn(move || loop {
mgm_0_handler.periodic_operation();
mgm_1_handler.periodic_operation();
//mgm_0_handler.periodic_operation();
//mgm_1_handler.periodic_operation();
thread::sleep(Duration::from_millis(FREQ_MS_AOCS));
})
.unwrap();
@@ -482,8 +331,8 @@ fn main() {
let jh_pus_handler = thread::Builder::new()
.name("sat-rs pus".to_string())
.spawn(move || loop {
event_handler.periodic_operation();
pus_stack.periodic_operation();
//event_handler.periodic_operation();
//pus_stack.periodic_operation();
thread::sleep(Duration::from_millis(FREQ_MS_PUS_STACK));
})
.unwrap();

View File

@@ -1,10 +1,8 @@
use std::{cell::RefCell, collections::VecDeque, sync::mpsc};
use satrs::{
pus::EcssTmSender,
queue::GenericSendError,
spacepackets::ecss::WritablePusPacket,
tmtc::{PacketAsVec, PacketHandler, PacketSenderWithSharedPool},
tmtc::{PacketAsVec, PacketHandler},
ComponentId,
};
@@ -14,8 +12,7 @@ pub struct MockSender(pub RefCell<VecDeque<PacketAsVec>>);
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub enum TmTcSender {
Static(PacketSenderWithSharedPool),
Heap(mpsc::SyncSender<PacketAsVec>),
Normal(mpsc::SyncSender<PacketAsVec>),
Mock(MockSender),
}
@@ -29,6 +26,7 @@ impl TmTcSender {
}
}
/*
impl EcssTmSender for TmTcSender {
fn send_tm(
&self,
@@ -36,7 +34,7 @@ impl EcssTmSender for TmTcSender {
tm: satrs::pus::PusTmVariant,
) -> Result<(), satrs::pus::EcssTmtcError> {
match self {
TmTcSender::Static(sync_sender) => sync_sender.send_tm(sender_id, tm),
//TmTcSender::Static(sync_sender) => sync_sender.send_tm(sender_id, tm),
TmTcSender::Heap(sync_sender) => match tm {
satrs::pus::PusTmVariant::InStore(_) => panic!("can not send TM in store"),
satrs::pus::PusTmVariant::Direct(pus_tm_creator) => sync_sender
@@ -47,19 +45,15 @@ impl EcssTmSender for TmTcSender {
}
}
}
*/
impl PacketHandler for TmTcSender {
type Error = GenericSendError;
fn handle_packet(&self, sender_id: ComponentId, packet: &[u8]) -> Result<(), Self::Error> {
match self {
TmTcSender::Static(packet_sender_with_shared_pool) => {
if let Err(e) = packet_sender_with_shared_pool.handle_packet(sender_id, packet) {
log::error!("Error sending packet via Static TM/TC sender: {:?}", e);
}
}
TmTcSender::Heap(sync_sender) => {
if let Err(e) = sync_sender.handle_packet(sender_id, packet) {
TmTcSender::Normal(sync_sender) => {
if let Err(e) = sync_sender.send(PacketAsVec::new(sender_id, packet.to_vec())) {
log::error!("Error sending packet via Heap TM/TC sender: {:?}", e);
}
}

View File

@@ -1,13 +1,16 @@
use satrs::{
pool::PoolProvider,
pus::HandlingStatus,
tmtc::{PacketAsVec, PacketInPool, SharedPacketPool},
spacepackets::{CcsdsPacketReader, ChecksumType},
tmtc::PacketAsVec,
};
use satrs_example::{CcsdsTcPacketOwned, ComponentId, TcHeader};
use std::{
collections::HashMap,
sync::mpsc::{self, TryRecvError},
};
use std::sync::mpsc::{self, TryRecvError};
use crate::pus::PusTcDistributor;
// TC source components where static pools are the backing memory of the received telecommands.
/*
pub struct TcSourceTaskStatic {
shared_tc_pool: SharedPacketPool,
tc_receiver: mpsc::Receiver<PacketInPool>,
@@ -64,19 +67,25 @@ impl TcSourceTaskStatic {
}
}
}
*/
pub type CcsdsDistributor = HashMap<ComponentId, std::sync::mpsc::SyncSender<CcsdsTcPacketOwned>>;
// TC source components where the heap is the backing memory of the received telecommands.
pub struct TcSourceTaskDynamic {
pub struct TcSourceTask {
pub tc_receiver: mpsc::Receiver<PacketAsVec>,
pus_distributor: PusTcDistributor,
ccsds_distributor: CcsdsDistributor,
}
#[allow(dead_code)]
impl TcSourceTaskDynamic {
pub fn new(tc_receiver: mpsc::Receiver<PacketAsVec>, pus_receiver: PusTcDistributor) -> Self {
//#[allow(dead_code)]
impl TcSourceTask {
pub fn new(
tc_receiver: mpsc::Receiver<PacketAsVec>,
ccsds_distributor: CcsdsDistributor,
) -> Self {
Self {
tc_receiver,
pus_distributor: pus_receiver,
ccsds_distributor,
}
}
@@ -88,10 +97,41 @@ impl TcSourceTaskDynamic {
// Right now, we only expect ECSS PUS packets.
// If packets like CFDP are expected, we might have to check the APID first.
match self.tc_receiver.try_recv() {
Ok(packet_as_vec) => {
self.pus_distributor
.handle_tc_packet_vec(packet_as_vec)
.ok();
Ok(packet) => {
let ccsds_tc_reader_result =
CcsdsPacketReader::new(&packet.packet, Some(ChecksumType::WithCrc16));
if ccsds_tc_reader_result.is_err() {
log::warn!(
"received invalid CCSDS TC packet: {:?}",
ccsds_tc_reader_result.err()
);
// TODO: Send a dedicated TM packet.
return HandlingStatus::HandledOne;
}
let ccsds_tc_reader = ccsds_tc_reader_result.unwrap();
let tc_header_result =
postcard::take_from_bytes::<TcHeader>(ccsds_tc_reader.user_data());
if tc_header_result.is_err() {
log::warn!(
"received CCSDS TC packet with invalid TC header: {:?}",
tc_header_result.err()
);
// TODO: Send a dedicated TM packet.
return HandlingStatus::HandledOne;
}
let (tc_header, payload) = tc_header_result.unwrap();
if let Some(sender) = self.ccsds_distributor.get(&tc_header.target_id) {
sender
.send(CcsdsTcPacketOwned {
sp_header: *ccsds_tc_reader.sp_header(),
tc_header,
payload: payload.to_vec(),
})
.ok();
} else {
log::warn!("no TC handler for target ID {:?}", tc_header.target_id);
// TODO: Send a dedicated TM packet.
}
HandlingStatus::HandledOne
}
Err(e) => match e {
@@ -104,18 +144,3 @@ impl TcSourceTaskDynamic {
}
}
}
#[allow(dead_code)]
pub enum TcSourceTask {
Static(TcSourceTaskStatic),
Heap(TcSourceTaskDynamic),
}
impl TcSourceTask {
pub fn periodic_operation(&mut self) {
match self {
TcSourceTask::Static(task) => task.periodic_operation(),
TcSourceTask::Heap(task) => task.periodic_operation(),
}
}
}

View File

@@ -6,16 +6,14 @@ use std::{
use arbitrary_int::{u11, u14};
use log::info;
use satrs::{
pool::PoolProvider,
spacepackets::{
ecss::{tm::PusTmZeroCopyWriter, PusPacket},
seq_count::SequenceCounter,
seq_count::SequenceCounterCcsdsSimple,
time::cds::MIN_CDS_FIELD_LEN,
seq_count::{SequenceCounter, SequenceCounterCcsdsSimple},
CcsdsPacket,
},
tmtc::{PacketAsVec, PacketInPool, SharedPacketPool},
tmtc::PacketAsVec,
};
use satrs_example::CcsdsTmPacketOwned;
use crate::interface::tcp::SyncTcpTmSource;
@@ -83,6 +81,7 @@ impl TmFunnelCommon {
}
}
/*
pub struct TmSinkStatic {
common: TmFunnelCommon,
shared_tm_store: SharedPacketPool,
@@ -130,19 +129,20 @@ impl TmSinkStatic {
}
}
}
*/
pub struct TmSinkDynamic {
pub struct TmSink {
common: TmFunnelCommon,
tm_funnel_rx: mpsc::Receiver<PacketAsVec>,
tm_server_tx: mpsc::SyncSender<PacketAsVec>,
tm_funnel_rx: mpsc::Receiver<CcsdsTmPacketOwned>,
tm_server_tx: mpsc::SyncSender<CcsdsTmPacketOwned>,
}
#[allow(dead_code)]
impl TmSinkDynamic {
//#[allow(dead_code)]
impl TmSink {
pub fn new(
sync_tm_tcp_source: SyncTcpTmSource,
tm_funnel_rx: mpsc::Receiver<PacketAsVec>,
tm_server_tx: mpsc::SyncSender<PacketAsVec>,
tm_funnel_rx: mpsc::Receiver<CcsdsTmPacketOwned>,
tm_server_tx: mpsc::SyncSender<CcsdsTmPacketOwned>,
) -> Self {
Self {
common: TmFunnelCommon::new(sync_tm_tcp_source),
@@ -153,13 +153,21 @@ impl TmSinkDynamic {
pub fn operation(&mut self) {
if let Ok(mut tm) = self.tm_funnel_rx.recv() {
//let tm_raw = tm.to_vec();
tm.sp_header.set_seq_count(
self.common
.seq_counter_map
.get_and_increment(tm.sp_header.apid()),
);
// Read the TM, set sequence counter and message counter, and finally update
// the CRC.
/*
let zero_copy_writer =
PusTmZeroCopyWriter::new(&mut tm.packet, MIN_CDS_FIELD_LEN, true)
PusTmZeroCopyWriter::new(&mut tm_raw, MIN_CDS_FIELD_LEN, true)
.expect("Creating TM zero copy writer failed");
self.common.apply_packet_processing(zero_copy_writer);
self.common.sync_tm_tcp_source.add_tm(&tm.packet);
*/
self.common.sync_tm_tcp_source.add_tm(&tm.to_vec());
self.tm_server_tx
.send(tm)
.expect("Sending TM to server failed");
@@ -167,12 +175,13 @@ impl TmSinkDynamic {
}
}
#[allow(dead_code)]
pub enum TmSink {
Static(TmSinkStatic),
Heap(TmSinkDynamic),
}
//#[allow(dead_code)]
//pub enum TmSink {
//Static(TmSinkStatic),
//Heap(TmSinkDynamic),
//}
/*
impl TmSink {
pub fn operation(&mut self) {
match self {
@@ -181,3 +190,4 @@ impl TmSink {
}
}
}
*/