rust/sat-rs
11
1
Fork 0
Code Issues 7 Pull Requests 2 Projects Releases 29 Wiki Activity

simpler event manager

Browse Source
This commit is contained in:
Robin Müller
2026-01-21 15:11:20 +01:00
parent 3c7fefbb8f
commit 2719d81a57
22 changed files with 100 additions and 1234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
satrs-example/client/src/main.rs
+1
View File
@@ -122,6 +122,7 @@ fn handle_raw_tm_packet(data: &[u8]) -> anyhow::Result<()> {
models::ComponentId::UdpServer => todo!(),
models::ComponentId::TcpServer => todo!(),
models::ComponentId::Ground => todo!(),
models::ComponentId::EventManager => todo!(),
}
}
Err(_) => todo!(),
satrs-example/models/Cargo.toml
+1
View File
@@ -6,6 +6,7 @@ edition = "2024"
[dependencies]
serde = { version = "1", features = ["derive"] }
spacepackets = { version = "0.17", git = "https://egit.irs.uni-stuttgart.de/rust/spacepackets.git", default-features = false }
satrs = { path = "../../satrs" }
num_enum = { version = "0.7" }
strum = { version = "0.27", features = ["derive"] }
postcard = { version = "1" }
satrs-example/models/src/control.rs
+14
View File
@@ -1,7 +1,21 @@
use crate::Message;
#[derive(serde::Serialize, serde::Deserialize, Clone, Copy, Debug)]
pub enum Event {
TestEvent,
}
impl Message for Event {
fn message_type(&self) -> crate::MessageType {
crate::MessageType::Event
}
}
pub mod request {
#[derive(serde::Serialize, serde::Deserialize, Clone, Copy, Debug)]
pub enum Request {
Ping,
TestEvent,
}
}
satrs-example/models/src/lib.rs
+13
View File
@@ -36,6 +36,7 @@ pub enum ComponentId {
UdpServer,
TcpServer,
EventManager,
Ground,
}
@@ -50,6 +51,17 @@ pub enum Apid {
Eps = 6,
}
#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize)]
pub enum Event {
ControllerEvent(control::Event),
}
impl Message for Event {
fn message_type(&self) -> MessageType {
MessageType::Event
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub struct TmHeader {
@@ -133,6 +145,7 @@ pub enum MessageType {
Mode,
Hk,
Action,
Event,
Verification,
}
satrs-example/src/config.rs
+1 -6
View File
@@ -10,10 +10,7 @@ use std::{collections::HashSet, net::Ipv4Addr};
use strum::IntoEnumIterator as _;
use num_enum::{IntoPrimitive, TryFromPrimitive};
use satrs::{
events_legacy::{EventU32TypedSev, SeverityInfo},
pool::{StaticMemoryPool, StaticPoolConfig},
};
use satrs::pool::{StaticMemoryPool, StaticPoolConfig};
#[derive(Copy, Clone, PartialEq, Eq, Debug, TryFromPrimitive, IntoPrimitive)]
#[repr(u8)]
@@ -39,8 +36,6 @@ pub enum GroupId {
pub const OBSW_SERVER_ADDR: Ipv4Addr = Ipv4Addr::UNSPECIFIED;
pub const SERVER_PORT: u16 = 7301;
pub const TEST_EVENT: EventU32TypedSev<SeverityInfo> = EventU32TypedSev::<SeverityInfo>::new(0, 0);
lazy_static! {
pub static ref PACKET_ID_VALIDATOR: HashSet<PacketId> = {
let mut set = HashSet::new();
satrs-example/src/control.rs
+10 -1
View File
@@ -9,14 +9,20 @@ use crate::ccsds::pack_ccsds_tm_packet_for_now;
pub struct Controller {
pub tc_rx: std::sync::mpsc::Receiver<CcsdsTcPacketOwned>,
pub tm_tx: std::sync::mpsc::SyncSender<CcsdsTmPacketOwned>,
pub event_ctrl_tx: std::sync::mpsc::SyncSender<control::Event>,
}
impl Controller {
pub fn new(
tc_rx: std::sync::mpsc::Receiver<CcsdsTcPacketOwned>,
tm_tx: std::sync::mpsc::SyncSender<CcsdsTmPacketOwned>,
event_ctrl_tx: std::sync::mpsc::SyncSender<control::Event>,
) -> Self {
Self { tc_rx, tm_tx }
Self {
tc_rx,
tm_tx,
event_ctrl_tx,
}
}
pub fn periodic_operation(&mut self) {
@@ -38,6 +44,9 @@ impl Controller {
match request {
control::request::Request::Ping => self
.send_telemetry(Some(tc_id), control::response::Response::Ok),
control::request::Request::TestEvent => {
self.event_ctrl_tx.send(control::Event::TestEvent).unwrap()
}
}
}
Err(e) => {
satrs-example/src/event_manager.rs
+33
View File
@@ -0,0 +1,33 @@
use models::{ccsds::CcsdsTmPacketOwned, control, ComponentId, Event, Message};
use crate::ccsds::pack_ccsds_tm_packet_for_now;
pub struct EventManager {
pub ctrl_rx: std::sync::mpsc::Receiver<control::Event>,
pub tm_tx: std::sync::mpsc::SyncSender<CcsdsTmPacketOwned>,
}
impl EventManager {
pub fn periodic_operation(&mut self) {
if let Ok(event) = self.ctrl_rx.try_recv() {
self.event_to_tm(ComponentId::Controller, &Event::ControllerEvent(event));
}
}
pub fn event_to_tm(
&mut self,
sender_id: ComponentId,
event: &(impl serde::Serialize + Message),
) {
match pack_ccsds_tm_packet_for_now(sender_id, None, event) {
Ok(packet) => {
if let Err(e) = self.tm_tx.send(packet) {
log::warn!("error sending event TM packet: {:?}", e);
}
}
Err(e) => {
log::warn!("error packing event TM packet: {:?}", e);
}
}
}
}
satrs-example/src/main.rs
+11 -7
View File
@@ -20,7 +20,7 @@ use models::ComponentId;
use satrs::{
hal::std::{tcp_server::ServerConfig, udp_server::UdpTcServer},
mode::{Mode, ModeAndSubmode, ModeRequest, ModeRequestHandlerMpscBounded},
pus::{event_man::EventRequestWithToken, HandlingStatus},
pus::HandlingStatus,
request::{GenericMessage, MessageMetadata},
spacepackets::time::cds::CdsTime,
};
@@ -39,6 +39,7 @@ use crate::{
acs::mgm::{MgmHandlerLis3Mdl, SpiDummyInterface, SpiSimInterface, SpiSimInterfaceWrapper},
control::Controller,
eps::pcdu::SwitchSet,
event_manager::EventManager,
interface::udp::UdpTmHandlerWithChannel,
tmtc::tc_source::CcsdsDistributor,
};
@@ -47,6 +48,7 @@ mod acs;
mod ccsds;
mod control;
mod eps;
mod event_manager;
mod interface;
mod logger;
mod spi;
@@ -54,7 +56,7 @@ mod tmtc;
fn main() {
setup_logger().expect("setting up logging with fern failed");
println!("Running OBSW example");
println!("Runng OBSW example");
let (tc_source_tx, tc_source_rx) = mpsc::sync_channel(50);
let (tm_sink_tx, tm_sink_rx) = mpsc::sync_channel(50);
@@ -75,12 +77,13 @@ fn main() {
let (_mgm_1_handler_mode_tx, mgm_1_handler_mode_rx) = mpsc::sync_channel(5);
let (pcdu_handler_mode_tx, pcdu_handler_mode_rx) = mpsc::sync_channel(5);
// Create event handling components
// These sender handles are used to send event requests, for example to enable or disable
// certain events.
let (_event_request_tx, _event_request_rx) = mpsc::channel::<EventRequestWithToken>();
let (event_ctrl_tx, event_ctrl_rx) = mpsc::sync_channel(10);
let mut event_manager = EventManager {
ctrl_rx: event_ctrl_rx,
tm_tx: tm_sink_tx.clone(),
};
let mut controller = Controller::new(controller_tc_rx, tm_sink_tx.clone());
let mut controller = Controller::new(controller_tc_rx, tm_sink_tx.clone(), event_ctrl_tx);
let ccsds_distributor = CcsdsDistributor::default();
let mut tc_source = TcSourceTask::new(tc_source_rx, ccsds_distributor);
@@ -312,6 +315,7 @@ fn main() {
.name("CTRL".to_string())
.spawn(move || loop {
controller.periodic_operation();
event_manager.periodic_operation();
thread::sleep(Duration::from_millis(FREQ_MS_CONTROLLER));
})
.unwrap();
satrs/Cargo.toml
-5
View File
@@ -78,8 +78,3 @@ test_util = []
[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--generate-link-to-definition"]
[[test]]
name = "event_test"
path = "tests/pus_events.rs"
required-features = ["test_util"]
satrs/src/event_man.rs → satrs/src/legacy/event_man.rs
+5 -4
View File
@@ -50,7 +50,7 @@
//! show how the event management modules can be integrated into a more complex software.
use crate::{
ComponentId,
events::{Event, EventId, GroupId},
legacy::events::{Event, EventId, GroupId},
queue::GenericSendError,
};
use core::marker::PhantomData;
@@ -96,6 +96,7 @@ impl<EventInstance: Event> EventMessage<EventInstance> {
pub trait EventSender<EventInstance: Event> {
type Error;
/// Destination component ID.
fn target_id(&self) -> ComponentId;
fn send(&self, message: EventMessage<EventInstance>) -> Result<(), Self::Error>;
@@ -317,7 +318,7 @@ pub mod alloc_mod {
use alloc::vec::Vec;
use hashbrown::HashMap;
use crate::events::EventErasedAlloc;
use crate::legacy::events::EventErasedAlloc;
use super::*;
@@ -460,7 +461,7 @@ pub mod alloc_mod {
#[cfg(feature = "std")]
pub mod std_mod {
use crate::{
events::{EventErasedAlloc, EventErasedHeapless},
legacy::events::{EventErasedAlloc, EventErasedHeapless},
queue::GenericReceiveError,
};
@@ -577,7 +578,7 @@ mod tests {
use arbitrary_int::u14;
use super::*;
use crate::events::{EventErasedAlloc, Severity};
use crate::legacy::events::{EventErasedAlloc, Severity};
use crate::pus::test_util::{TEST_COMPONENT_ID_0, TEST_COMPONENT_ID_1};
use std::sync::mpsc;
satrs/src/event_man_legacy.rs → satrs/src/legacy/event_man_legacy.rs
+3 -3
View File
@@ -50,7 +50,7 @@
//! The [PUS event](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example/src/pus/event.rs)
//! module and the generic [events module](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-example/src/events.rs)
//! show how the event management modules can be integrated into a more complex software.
use crate::events_legacy::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
use crate::legacy::events_legacy::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
use crate::params::Params;
use crate::queue::GenericSendError;
use core::fmt::Debug;
@@ -588,8 +588,8 @@ pub mod std_mod {
#[cfg(test)]
mod tests {
use super::*;
use crate::event_man_legacy::EventManager;
use crate::events_legacy::{EventU32, GenericEvent, Severity};
use crate::legacy::event_man_legacy::EventManager;
use crate::legacy::events_legacy::{EventU32, GenericEvent, Severity};
use crate::params::{ParamsHeapless, ParamsRaw};
use crate::pus::test_util::{TEST_COMPONENT_ID_0, TEST_COMPONENT_ID_1};
use std::format;
satrs/src/events.rs → satrs/src/legacy/events.rs
View File
satrs/src/events_legacy.rs → satrs/src/legacy/events_legacy.rs
View File
satrs/src/legacy/mod.rs
+4
View File
@@ -0,0 +1,4 @@
pub mod event_man;
pub mod events;
pub mod pus;
satrs/src/pus/event.rs → satrs/src/legacy/pus/event.rs
+1 -214
View File
@@ -267,217 +267,4 @@ mod alloc_mod {
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ComponentId;
use crate::events_legacy::{EventU32, Severity};
use crate::pus::test_util::TEST_COMPONENT_ID_0;
use crate::pus::tests::CommonTmInfo;
use crate::pus::{ChannelWithId, EcssTmSender, EcssTmtcError, PusTmVariant};
use spacepackets::ByteConversionError;
use spacepackets::ecss::PusError;
use std::cell::RefCell;
use std::collections::VecDeque;
use std::vec::Vec;
const EXAMPLE_APID: u11 = u11::new(0xee);
const EXAMPLE_GROUP_ID: u16 = 2;
const EXAMPLE_EVENT_ID_0: u16 = 1;
#[allow(dead_code)]
const EXAMPLE_EVENT_ID_1: u16 = 2;
#[derive(Debug, Eq, PartialEq, Clone)]
struct TmInfo {
pub sender_id: ComponentId,
pub common: CommonTmInfo,
pub event: EventU32,
pub aux_data: Vec<u8>,
}
#[derive(Default, Clone)]
struct TestSender {
pub service_queue: RefCell<VecDeque<TmInfo>>,
}
impl ChannelWithId for TestSender {
fn id(&self) -> ComponentId {
0
}
}
impl EcssTmSender for TestSender {
fn send_tm(&self, sender_id: ComponentId, tm: PusTmVariant) -> Result<(), EcssTmtcError> {
match tm {
PusTmVariant::InStore(_) => {
panic!("TestSender: unexpected call with address");
}
PusTmVariant::Direct(tm) => {
assert!(!tm.source_data().is_empty());
let src_data = tm.source_data();
assert!(src_data.len() >= 4);
let event =
EventU32::from(u32::from_be_bytes(src_data[0..4].try_into().unwrap()));
let mut aux_data = Vec::new();
if src_data.len() > 4 {
aux_data.extend_from_slice(&src_data[4..]);
}
self.service_queue.borrow_mut().push_back(TmInfo {
sender_id,
common: CommonTmInfo::new_from_tm(&tm),
event,
aux_data,
});
Ok(())
}
}
}
}
fn severity_to_subservice(severity: Severity) -> MessageSubtypeId {
match severity {
Severity::Info => MessageSubtypeId::TmInfoReport,
Severity::Low => MessageSubtypeId::TmLowSeverityReport,
Severity::Medium => MessageSubtypeId::TmMediumSeverityReport,
Severity::High => MessageSubtypeId::TmHighSeverityReport,
}
}
fn report_basic_event(
reporter: &mut EventReporter,
sender: &mut TestSender,
time_stamp: &[u8],
event: EventU32,
severity: Severity,
aux_data: Option<&[u8]>,
) {
match severity {
Severity::Info => {
reporter
.event_info(sender, time_stamp, event, aux_data)
.expect("Error reporting info event");
}
Severity::Low => {
reporter
.event_low_severity(sender, time_stamp, event, aux_data)
.expect("Error reporting low event");
}
Severity::Medium => {
reporter
.event_medium_severity(sender, time_stamp, event, aux_data)
.expect("Error reporting medium event");
}
Severity::High => {
reporter
.event_high_severity(sender, time_stamp, event, aux_data)
.expect("Error reporting high event");
}
}
}
fn basic_event_test(
max_event_aux_data_buf: usize,
severity: Severity,
error_data: Option<&[u8]>,
) {
let mut sender = TestSender::default();
let mut reporter = EventReporter::new(
TEST_COMPONENT_ID_0.id(),
EXAMPLE_APID,
0,
max_event_aux_data_buf,
);
let time_stamp_empty: [u8; 7] = [0; 7];
let mut error_copy = Vec::new();
if let Some(err_data) = error_data {
error_copy.extend_from_slice(err_data);
}
let event = EventU32::new_checked(severity, EXAMPLE_GROUP_ID, EXAMPLE_EVENT_ID_0)
.expect("Error creating example event");
report_basic_event(
&mut reporter,
&mut sender,
&time_stamp_empty,
event,
severity,
error_data,
);
let mut service_queue = sender.service_queue.borrow_mut();
assert_eq!(service_queue.len(), 1);
let tm_info = service_queue.pop_front().unwrap();
assert_eq!(
tm_info.common.subservice,
severity_to_subservice(severity) as u8
);
assert_eq!(tm_info.common.dest_id, 0);
assert_eq!(tm_info.common.timestamp, time_stamp_empty);
assert_eq!(tm_info.common.msg_counter, 0);
assert_eq!(tm_info.common.apid, EXAMPLE_APID);
assert_eq!(tm_info.event, event);
assert_eq!(tm_info.sender_id, TEST_COMPONENT_ID_0.id());
assert_eq!(tm_info.aux_data, error_copy);
}
#[test]
fn basic_info_event_generation() {
basic_event_test(4, Severity::Info, None);
}
#[test]
fn basic_low_severity_event() {
basic_event_test(4, Severity::Low, None);
}
#[test]
fn basic_medium_severity_event() {
basic_event_test(4, Severity::Medium, None);
}
#[test]
fn basic_high_severity_event() {
basic_event_test(4, Severity::High, None);
}
#[test]
fn event_with_info_string() {
let info_string = "Test Information";
basic_event_test(32, Severity::Info, Some(info_string.as_bytes()));
}
#[test]
fn low_severity_with_raw_err_data() {
let raw_err_param: i32 = -1;
let raw_err = raw_err_param.to_be_bytes();
basic_event_test(8, Severity::Low, Some(&raw_err))
}
fn check_buf_too_small(
reporter: &mut EventReporter,
sender: &mut TestSender,
expected_found_len: usize,
) {
let time_stamp_empty: [u8; 7] = [0; 7];
let event = EventU32::new_checked(Severity::Info, EXAMPLE_GROUP_ID, EXAMPLE_EVENT_ID_0)
.expect("Error creating example event");
let err = reporter.event_info(sender, &time_stamp_empty, event, None);
assert!(err.is_err());
let err = err.unwrap_err();
if let EcssTmtcError::Pus(PusError::ByteConversion(
ByteConversionError::ToSliceTooSmall { found, expected },
)) = err
{
assert_eq!(expected, 4);
assert_eq!(found, expected_found_len);
} else {
panic!("Unexpected error {:?}", err);
}
}
#[test]
fn insufficient_buffer() {
let mut sender = TestSender::default();
for i in 0..3 {
let mut reporter = EventReporter::new(0, EXAMPLE_APID, 0, i);
check_buf_too_small(&mut reporter, &mut sender, i);
}
}
}
mod tests {}
satrs/src/legacy/pus/mod.rs
+1
View File
@@ -0,0 +1 @@
pub mod event;
satrs/src/lib.rs
+1 -6
View File
@@ -9,8 +9,6 @@
//!
//! The core modules of this crate include
//!
//! - The [event manager][event_man] module which provides a publish and
//! and subscribe to route events.
//! - The [pus] module which provides special support for projects using
//! the [ECSS PUS C standard](https://ecss.nl/standard/ecss-e-st-70-41c-space-engineering-telemetry-and-telecommand-packet-utilization-15-april-2016/).
#![no_std]
@@ -27,15 +25,12 @@ pub mod ccsds;
#[cfg(feature = "alloc")]
pub mod dev_mgmt;
pub mod encoding;
pub mod event_man;
pub mod event_man_legacy;
pub mod events;
pub mod events_legacy;
#[cfg(feature = "std")]
pub mod executable;
pub mod hal;
pub mod health;
pub mod hk;
pub mod legacy;
pub mod mode;
#[cfg(feature = "std")]
pub mod mode_tree;
satrs/src/pus/event_man.rs
-464
View File
@@ -1,464 +0,0 @@
use crate::events_legacy::{EventU32, GenericEvent, Severity};
#[cfg(feature = "alloc")]
use crate::events_legacy::{EventU32TypedSev, HasSeverity};
#[cfg(feature = "alloc")]
use core::hash::Hash;
#[cfg(feature = "alloc")]
use hashbrown::HashSet;
#[cfg(feature = "alloc")]
use crate::pus::EcssTmSender;
use crate::pus::EcssTmtcError;
#[cfg(feature = "alloc")]
pub use crate::pus::event::EventReporter;
use crate::pus::verification::TcStateToken;
#[cfg(feature = "alloc")]
pub use alloc_mod::*;
#[cfg(feature = "heapless")]
pub use heapless_mod::*;
/// This trait allows the PUS event manager implementation to stay generic over various types
/// of backend containers.
///
/// These backend containers keep track on whether a particular event is enabled or disabled for
/// reporting and also expose a simple API to enable or disable the event reporting.
///
/// For example, a straight forward implementation for host systems could use a
/// [hash set](https://docs.rs/hashbrown/latest/hashbrown/struct.HashSet.html)
/// structure to track disabled events. A more primitive and embedded friendly
/// solution could track this information in a static or pre-allocated list which contains
/// the disabled events.
pub trait PusEventReportingMapProvider<Event: GenericEvent> {
type Error;
fn event_enabled(&self, event: &Event) -> bool;
fn enable_event_reporting(&mut self, event: &Event) -> Result<bool, Self::Error>;
fn disable_event_reporting(&mut self, event: &Event) -> Result<bool, Self::Error>;
}
#[cfg(feature = "heapless")]
pub mod heapless_mod {
use super::*;
use crate::events::LargestEventRaw;
use core::marker::PhantomData;
// TODO: After a new version of heapless is released which uses hash32 version 0.3, try using
// regular Event type again.
#[derive(Default)]
pub struct HeaplessPusMgmtBackendProvider<const N: usize, Provider: GenericEvent> {
disabled: heapless::index_set::FnvIndexSet<LargestEventRaw, N>,
phantom: PhantomData<Provider>,
}
impl<const N: usize, Provider: GenericEvent> PusEventReportingMapProvider<Provider>
for HeaplessPusMgmtBackendProvider<N, Provider>
{
type Error = ();
fn event_enabled(&self, event: &Provider) -> bool {
self.disabled.contains(&event.raw_as_largest_type())
}
fn enable_event_reporting(&mut self, event: &Provider) -> Result<bool, Self::Error> {
self.disabled
.insert(event.raw_as_largest_type())
.map_err(|_| ())
}
fn disable_event_reporting(&mut self, event: &Provider) -> Result<bool, Self::Error> {
Ok(self.disabled.remove(&event.raw_as_largest_type()))
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum EventRequest<Event: GenericEvent = EventU32> {
Enable(Event),
Disable(Event),
}
#[derive(Debug)]
pub struct EventRequestWithToken<Event: GenericEvent = EventU32> {
pub request: EventRequest<Event>,
pub token: TcStateToken,
}
#[derive(Debug)]
pub enum EventManError {
EcssTmtcError(EcssTmtcError),
SeverityMissmatch(Severity, Severity),
}
impl From<EcssTmtcError> for EventManError {
fn from(v: EcssTmtcError) -> Self {
Self::EcssTmtcError(v)
}
}
#[cfg(feature = "alloc")]
pub mod alloc_mod {
use core::marker::PhantomData;
use crate::{
events_legacy::EventU16,
params::{Params, WritableToBeBytes},
pus::event::{DummyEventHook, EventTmHook},
};
use super::*;
/// Default backend provider which uses a hash set as the event reporting status container
/// like mentioned in the example of the [PusEventReportingMapProvider] documentation.
///
/// This provider is a good option for host systems or larger embedded systems where
/// the expected occasional memory allocation performed by the [HashSet] is not an issue.
pub struct DefaultPusEventReportingMap<Event: GenericEvent = EventU32> {
disabled: HashSet<Event>,
}
impl<Event: GenericEvent> Default for DefaultPusEventReportingMap<Event> {
fn default() -> Self {
Self {
disabled: HashSet::default(),
}
}
}
impl<Event: GenericEvent + PartialEq + Eq + Hash + Copy + Clone>
PusEventReportingMapProvider<Event> for DefaultPusEventReportingMap<Event>
{
type Error = ();
fn event_enabled(&self, event: &Event) -> bool {
!self.disabled.contains(event)
}
fn enable_event_reporting(&mut self, event: &Event) -> Result<bool, Self::Error> {
Ok(self.disabled.remove(event))
}
fn disable_event_reporting(&mut self, event: &Event) -> Result<bool, Self::Error> {
Ok(self.disabled.insert(*event))
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct EventGenerationResult {
pub event_was_enabled: bool,
pub params_were_propagated: bool,
}
pub struct PusEventTmCreatorWithMap<
ReportingMap: PusEventReportingMapProvider<Event>,
Event: GenericEvent,
EventTmHookInstance: EventTmHook = DummyEventHook,
> {
pub reporter: EventReporter<EventTmHookInstance>,
reporting_map: ReportingMap,
phantom: PhantomData<Event>,
}
impl<
ReportingMap: PusEventReportingMapProvider<Event>,
Event: GenericEvent,
EventTmHookInstance: EventTmHook,
> PusEventTmCreatorWithMap<ReportingMap, Event, EventTmHookInstance>
{
pub fn new(reporter: EventReporter<EventTmHookInstance>, backend: ReportingMap) -> Self {
Self {
reporter,
reporting_map: backend,
phantom: PhantomData,
}
}
pub fn enable_tm_for_event(&mut self, event: &Event) -> Result<bool, ReportingMap::Error> {
self.reporting_map.enable_event_reporting(event)
}
pub fn disable_tm_for_event(&mut self, event: &Event) -> Result<bool, ReportingMap::Error> {
self.reporting_map.disable_event_reporting(event)
}
pub fn generate_pus_event_tm_generic(
&self,
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event: Event,
params: Option<&[u8]>,
) -> Result<bool, EventManError> {
if !self.reporting_map.event_enabled(&event) {
return Ok(false);
}
match event.severity() {
Severity::Info => self
.reporter
.event_info(sender, time_stamp, event, params)
.map(|_| true)
.map_err(|e| e.into()),
Severity::Low => self
.reporter
.event_low_severity(sender, time_stamp, event, params)
.map(|_| true)
.map_err(|e| e.into()),
Severity::Medium => self
.reporter
.event_medium_severity(sender, time_stamp, event, params)
.map(|_| true)
.map_err(|e| e.into()),
Severity::High => self
.reporter
.event_high_severity(sender, time_stamp, event, params)
.map(|_| true)
.map_err(|e| e.into()),
}
}
pub fn generate_pus_event_tm_generic_with_generic_params(
&self,
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event: Event,
small_data_buf: &mut [u8],
params: Option<&Params>,
) -> Result<EventGenerationResult, EventManError> {
let mut result = EventGenerationResult {
event_was_enabled: false,
params_were_propagated: true,
};
if params.is_none() {
result.event_was_enabled =
self.generate_pus_event_tm_generic(sender, time_stamp, event, None)?;
return Ok(result);
}
let params = params.unwrap();
result.event_was_enabled = match params {
Params::Heapless(heapless_param) => {
heapless_param
.write_to_be_bytes(&mut small_data_buf[..heapless_param.written_len()])
.map_err(EcssTmtcError::ByteConversion)?;
self.generate_pus_event_tm_generic(
sender,
time_stamp,
event,
Some(small_data_buf),
)?
}
Params::Vec(vec) => {
self.generate_pus_event_tm_generic(sender, time_stamp, event, Some(vec))?
}
Params::String(string) => self.generate_pus_event_tm_generic(
sender,
time_stamp,
event,
Some(string.as_bytes()),
)?,
_ => {
result.params_were_propagated = false;
self.generate_pus_event_tm_generic(sender, time_stamp, event, None)?
}
};
Ok(result)
}
}
impl<Event: GenericEvent + Copy + PartialEq + Eq + Hash, EventTmHookInstance: EventTmHook>
PusEventTmCreatorWithMap<DefaultPusEventReportingMap<Event>, Event, EventTmHookInstance>
{
pub fn new_with_default_backend(reporter: EventReporter<EventTmHookInstance>) -> Self {
Self {
reporter,
reporting_map: DefaultPusEventReportingMap::default(),
phantom: PhantomData,
}
}
}
impl<ReportingMap: PusEventReportingMapProvider<EventU32>>
PusEventTmCreatorWithMap<ReportingMap, EventU32>
{
pub fn enable_tm_for_event_with_sev<Severity: HasSeverity>(
&mut self,
event: &EventU32TypedSev<Severity>,
) -> Result<bool, ReportingMap::Error> {
self.reporting_map.enable_event_reporting(event.as_ref())
}
pub fn disable_tm_for_event_with_sev<Severity: HasSeverity>(
&mut self,
event: &EventU32TypedSev<Severity>,
) -> Result<bool, ReportingMap::Error> {
self.reporting_map.disable_event_reporting(event.as_ref())
}
pub fn generate_pus_event_tm<Severity: HasSeverity>(
&self,
sender: &(impl EcssTmSender + ?Sized),
time_stamp: &[u8],
event: EventU32TypedSev<Severity>,
aux_data: Option<&[u8]>,
) -> Result<bool, EventManError> {
self.generate_pus_event_tm_generic(sender, time_stamp, event.into(), aux_data)
}
}
pub type DefaultPusEventU16TmCreator<EventTmHook = DummyEventHook> =
PusEventTmCreatorWithMap<DefaultPusEventReportingMap<EventU16>, EventU16, EventTmHook>;
pub type DefaultPusEventU32TmCreator<EventTmHook = DummyEventHook> =
PusEventTmCreatorWithMap<DefaultPusEventReportingMap<EventU32>, EventU32, EventTmHook>;
}
#[cfg(test)]
mod tests {
use alloc::string::{String, ToString};
use alloc::vec;
use arbitrary_int::{u11, u21};
use spacepackets::ecss::PusPacket;
use spacepackets::ecss::event::MessageSubtypeId;
use spacepackets::ecss::tm::PusTmReader;
use super::*;
use crate::request::UniqueApidTargetId;
use crate::{events_legacy::SeverityInfo, tmtc::PacketAsVec};
use std::sync::mpsc::{self, TryRecvError};
const INFO_EVENT: EventU32TypedSev<SeverityInfo> = EventU32TypedSev::<SeverityInfo>::new(1, 0);
const LOW_SEV_EVENT: EventU32 = EventU32::new(Severity::Low, 1, 5);
const EMPTY_STAMP: [u8; 7] = [0; 7];
const TEST_APID: u11 = u11::new(0x02);
const TEST_ID: UniqueApidTargetId = UniqueApidTargetId::new(TEST_APID, u21::new(0x05));
fn create_basic_man_1() -> DefaultPusEventU32TmCreator {
let reporter = EventReporter::new(TEST_ID.raw(), TEST_APID, 0, 128);
PusEventTmCreatorWithMap::new_with_default_backend(reporter)
}
fn create_basic_man_2() -> DefaultPusEventU32TmCreator {
let reporter = EventReporter::new(TEST_ID.raw(), TEST_APID, 0, 128);
let backend = DefaultPusEventReportingMap::default();
PusEventTmCreatorWithMap::new(reporter, backend)
}
#[test]
fn test_basic() {
let event_man = create_basic_man_1();
let (event_tx, event_rx) = mpsc::channel::<PacketAsVec>();
let event_sent = event_man
.generate_pus_event_tm(&event_tx, &EMPTY_STAMP, INFO_EVENT, None)
.expect("Sending info event failed");
assert!(event_sent);
// Will not check packet here, correctness of packet was tested somewhere else
event_rx.try_recv().expect("Receiving event TM failed");
}
#[test]
fn test_disable_event() {
let mut event_man = create_basic_man_2();
let (event_tx, event_rx) = mpsc::channel::<PacketAsVec>();
// let mut sender = TmAsVecSenderWithMpsc::new(0, "test", event_tx);
let res = event_man.disable_tm_for_event(&LOW_SEV_EVENT);
assert!(res.is_ok());
assert!(res.unwrap());
let mut event_sent = event_man
.generate_pus_event_tm_generic(&event_tx, &EMPTY_STAMP, LOW_SEV_EVENT, None)
.expect("Sending low severity event failed");
assert!(!event_sent);
let res = event_rx.try_recv();
assert!(res.is_err());
assert!(matches!(res.unwrap_err(), TryRecvError::Empty));
// Check that only the low severity event was disabled
event_sent = event_man
.generate_pus_event_tm(&event_tx, &EMPTY_STAMP, INFO_EVENT, None)
.expect("Sending info event failed");
assert!(event_sent);
event_rx.try_recv().expect("No info event received");
}
#[test]
fn test_reenable_event() {
let mut event_man = create_basic_man_1();
let (event_tx, event_rx) = mpsc::channel::<PacketAsVec>();
let mut res = event_man.disable_tm_for_event_with_sev(&INFO_EVENT);
assert!(res.is_ok());
assert!(res.unwrap());
res = event_man.enable_tm_for_event_with_sev(&INFO_EVENT);
assert!(res.is_ok());
assert!(res.unwrap());
let event_sent = event_man
.generate_pus_event_tm(&event_tx, &EMPTY_STAMP, INFO_EVENT, None)
.expect("Sending info event failed");
assert!(event_sent);
event_rx.try_recv().expect("No info event received");
}
#[test]
fn test_event_with_generic_string_param() {
let event_man = create_basic_man_1();
let mut small_data_buf = [0; 128];
let param_data = "hello world";
let (event_tx, event_rx) = mpsc::channel::<PacketAsVec>();
let res = event_man.generate_pus_event_tm_generic_with_generic_params(
&event_tx,
&EMPTY_STAMP,
INFO_EVENT.into(),
&mut small_data_buf,
Some(&param_data.to_string().into()),
);
assert!(res.is_ok());
let res = res.unwrap();
assert!(res.event_was_enabled);
assert!(res.params_were_propagated);
let event_tm = event_rx.try_recv().expect("no event received");
let tm = PusTmReader::new(&event_tm.packet, 7).expect("reading TM failed");
assert_eq!(tm.service_type_id(), 5);
assert_eq!(
tm.message_subtype_id(),
MessageSubtypeId::TmInfoReport as u8
);
assert_eq!(tm.user_data().len(), 4 + param_data.len());
let u32_event = u32::from_be_bytes(tm.user_data()[0..4].try_into().unwrap());
assert_eq!(u32_event, INFO_EVENT.raw());
let string_data = String::from_utf8_lossy(&tm.user_data()[4..]);
assert_eq!(string_data, param_data);
}
#[test]
fn test_event_with_generic_vec_param() {
let event_man = create_basic_man_1();
let mut small_data_buf = [0; 128];
let param_data = vec![1, 2, 3, 4];
let (event_tx, event_rx) = mpsc::channel::<PacketAsVec>();
let res = event_man.generate_pus_event_tm_generic_with_generic_params(
&event_tx,
&EMPTY_STAMP,
INFO_EVENT.into(),
&mut small_data_buf,
Some(&param_data.clone().into()),
);
assert!(res.is_ok());
let res = res.unwrap();
assert!(res.event_was_enabled);
assert!(res.params_were_propagated);
let event_tm = event_rx.try_recv().expect("no event received");
let tm = PusTmReader::new(&event_tm.packet, 7).expect("reading TM failed");
assert_eq!(tm.service_type_id(), 5);
assert_eq!(
tm.message_subtype_id(),
MessageSubtypeId::TmInfoReport as u8
);
assert_eq!(tm.user_data().len(), 4 + param_data.len());
let u32_event = u32::from_be_bytes(tm.user_data()[0..4].try_into().unwrap());
assert_eq!(u32_event, INFO_EVENT.raw());
let vec_data = tm.user_data()[4..].to_vec();
assert_eq!(vec_data, param_data);
}
#[test]
fn test_event_with_generic_store_param_not_propagated() {
// TODO: Test this.
}
#[test]
fn test_event_with_generic_heapless_param() {
// TODO: Test this.
}
}
satrs/src/pus/event_srv.rs
-355
View File
@@ -1,355 +0,0 @@
use crate::events_legacy::EventU32;
use crate::pus::event_man::{EventRequest, EventRequestWithToken};
use crate::pus::verification::TcStateToken;
use crate::pus::{DirectPusPacketHandlerResult, PartialPusHandlingError, PusPacketHandlingError};
use crate::queue::GenericSendError;
use spacepackets::ecss::PusPacket;
use spacepackets::ecss::event::MessageSubtypeId;
use std::sync::mpsc::Sender;
use super::verification::VerificationReportingProvider;
use super::{
CacheAndReadRawEcssTc, EcssTcReceiver, EcssTmSender, GenericConversionError,
GenericRoutingError, HandlingStatus, PusServiceHelper,
};
pub struct PusEventServiceHandler<
TcReceiver: EcssTcReceiver,
TmSender: EcssTmSender,
TcInMemConverter: CacheAndReadRawEcssTc,
VerificationReporter: VerificationReportingProvider,
> {
pub service_helper:
PusServiceHelper<TcReceiver, TmSender, TcInMemConverter, VerificationReporter>,
event_request_tx: Sender<EventRequestWithToken>,
}
impl<
TcReceiver: EcssTcReceiver,
TmSender: EcssTmSender,
TcInMemConverter: CacheAndReadRawEcssTc,
VerificationReporter: VerificationReportingProvider,
> PusEventServiceHandler<TcReceiver, TmSender, TcInMemConverter, VerificationReporter>
{
pub fn new(
service_helper: PusServiceHelper<
TcReceiver,
TmSender,
TcInMemConverter,
VerificationReporter,
>,
event_request_tx: Sender<EventRequestWithToken>,
) -> Self {
Self {
service_helper,
event_request_tx,
}
}
pub fn poll_and_handle_next_tc<ErrorCb: FnMut(&PartialPusHandlingError)>(
&mut self,
mut error_callback: ErrorCb,
time_stamp: &[u8],
) -> Result<DirectPusPacketHandlerResult, PusPacketHandlingError> {
let possible_packet = self.service_helper.retrieve_and_accept_next_packet()?;
if possible_packet.is_none() {
return Ok(HandlingStatus::Empty.into());
}
let ecss_tc_and_token = possible_packet.unwrap();
self.service_helper
.tc_in_mem_converter_mut()
.cache(&ecss_tc_and_token.tc_in_memory)?;
let tc = self.service_helper.tc_in_mem_converter().convert()?;
let subservice = tc.message_subtype_id();
let srv = MessageSubtypeId::try_from(subservice);
if srv.is_err() {
return Ok(DirectPusPacketHandlerResult::CustomSubservice(
tc.message_subtype_id(),
ecss_tc_and_token.token,
));
}
let mut handle_enable_disable_request =
|enable: bool| -> Result<DirectPusPacketHandlerResult, PusPacketHandlingError> {
if tc.user_data().len() < 4 {
return Err(GenericConversionError::NotEnoughAppData {
expected: 4,
found: tc.user_data().len(),
}
.into());
}
let user_data = tc.user_data();
let event_u32 =
EventU32::from(u32::from_be_bytes(user_data[0..4].try_into().unwrap()));
let mut token: TcStateToken = ecss_tc_and_token.token.into();
match self.service_helper.common.verif_reporter.start_success(
&self.service_helper.common.tm_sender,
ecss_tc_and_token.token,
time_stamp,
) {
Ok(start_token) => {
token = start_token.into();
}
Err(e) => {
error_callback(&PartialPusHandlingError::Verification(e));
}
}
let event_req_with_token = if enable {
EventRequestWithToken {
request: EventRequest::Enable(event_u32),
token,
}
} else {
EventRequestWithToken {
request: EventRequest::Disable(event_u32),
token,
}
};
self.event_request_tx
.send(event_req_with_token)
.map_err(|_| {
PusPacketHandlingError::RequestRouting(GenericRoutingError::Send(
GenericSendError::RxDisconnected,
))
})?;
Ok(HandlingStatus::HandledOne.into())
};
match srv.unwrap() {
MessageSubtypeId::TmInfoReport
| MessageSubtypeId::TmLowSeverityReport
| MessageSubtypeId::TmMediumSeverityReport
| MessageSubtypeId::TmHighSeverityReport => {
return Err(PusPacketHandlingError::RequestConversion(
GenericConversionError::WrongService(tc.message_subtype_id()),
));
}
MessageSubtypeId::TcEnableEventGeneration => {
handle_enable_disable_request(true)?;
}
MessageSubtypeId::TcDisableEventGeneration => {
handle_enable_disable_request(false)?;
}
MessageSubtypeId::TcReportDisabledList | MessageSubtypeId::TmDisabledEventsReport => {
return Ok(DirectPusPacketHandlerResult::SubserviceNotImplemented(
subservice,
ecss_tc_and_token.token,
));
}
}
Ok(HandlingStatus::HandledOne.into())
}
}
#[cfg(test)]
mod tests {
use arbitrary_int::traits::Integer as _;
use arbitrary_int::u14;
use delegate::delegate;
use spacepackets::ecss::event::MessageSubtypeId;
use spacepackets::ecss::{CreatorConfig, MessageTypeId};
use spacepackets::time::{TimeWriter, cds};
use spacepackets::util::UnsignedEnum;
use spacepackets::{
SpHeader,
ecss::{
tc::{PusTcCreator, PusTcSecondaryHeader},
tm::PusTmReader,
},
};
use std::sync::mpsc::{self, Sender};
use crate::pus::event_man::EventRequest;
use crate::pus::test_util::{PusTestHarness, SimplePusPacketHandler, TEST_APID};
use crate::pus::verification::{
RequestId, VerificationReporter, VerificationReportingProvider,
};
use crate::pus::{GenericConversionError, HandlingStatus, MpscTcReceiver};
use crate::tmtc::PacketSenderWithSharedPool;
use crate::{
events_legacy::EventU32,
pus::{
DirectPusPacketHandlerResult, EcssTcInSharedPoolCacher, PusPacketHandlingError,
event_man::EventRequestWithToken,
tests::PusServiceHandlerWithSharedStoreCommon,
verification::{TcStateAccepted, VerificationToken},
},
};
use super::PusEventServiceHandler;
const TEST_EVENT_0: EventU32 = EventU32::new(crate::events_legacy::Severity::Info, 5, 25);
struct Pus5HandlerWithStoreTester {
common: PusServiceHandlerWithSharedStoreCommon,
handler: PusEventServiceHandler<
MpscTcReceiver,
PacketSenderWithSharedPool,
EcssTcInSharedPoolCacher,
VerificationReporter,
>,
}
impl Pus5HandlerWithStoreTester {
pub fn new(event_request_tx: Sender<EventRequestWithToken>) -> Self {
let (common, srv_handler) = PusServiceHandlerWithSharedStoreCommon::new(0);
Self {
common,
handler: PusEventServiceHandler::new(srv_handler, event_request_tx),
}
}
}
impl PusTestHarness for Pus5HandlerWithStoreTester {
fn start_verification(&mut self, tc: &PusTcCreator) -> VerificationToken<TcStateAccepted> {
let init_token = self
.handler
.service_helper
.verif_reporter_mut()
.start_verification(tc);
self.handler
.service_helper
.verif_reporter()
.acceptance_success(self.handler.service_helper.tm_sender(), init_token, &[0; 7])
.expect("acceptance success failure")
}
fn send_tc(&self, token: &VerificationToken<TcStateAccepted>, tc: &PusTcCreator) {
self.common
.send_tc(self.handler.service_helper.id(), token, tc);
}
delegate! {
to self.common {
fn read_next_tm(&mut self) -> PusTmReader<'_>;
fn check_no_tm_available(&self) -> bool;
fn check_next_verification_tm(&self, subservice: u8, expected_request_id: RequestId);
}
}
}
impl SimplePusPacketHandler for Pus5HandlerWithStoreTester {
fn handle_one_tc(
&mut self,
) -> Result<DirectPusPacketHandlerResult, PusPacketHandlingError> {
let time_stamp = cds::CdsTime::new_with_u16_days(0, 0).to_vec().unwrap();
self.handler.poll_and_handle_next_tc(|_| {}, &time_stamp)
}
}
fn event_test(
test_harness: &mut (impl PusTestHarness + SimplePusPacketHandler),
subservice: MessageSubtypeId,
expected_event_req: EventRequest,
event_req_receiver: mpsc::Receiver<EventRequestWithToken>,
) {
let sp_header = SpHeader::new_for_unseg_tc(TEST_APID, u14::ZERO, 0);
let sec_header = PusTcSecondaryHeader::new_simple(MessageTypeId::new(5, subservice as u8));
let mut app_data = [0; 4];
TEST_EVENT_0
.write_to_be_bytes(&mut app_data)
.expect("writing test event failed");
let ping_tc = PusTcCreator::new(sp_header, sec_header, &app_data, CreatorConfig::default());
let token = test_harness.start_verification(&ping_tc);
test_harness.send_tc(&token, &ping_tc);
let request_id = token.request_id();
test_harness.handle_one_tc().unwrap();
test_harness.check_next_verification_tm(1, request_id);
test_harness.check_next_verification_tm(3, request_id);
// Completion TM is not generated for us.
assert!(test_harness.check_no_tm_available());
let event_request = event_req_receiver
.try_recv()
.expect("no event request received");
assert_eq!(expected_event_req, event_request.request);
}
#[test]
fn test_enabling_event_reporting() {
let (event_request_tx, event_request_rx) = mpsc::channel();
let mut test_harness = Pus5HandlerWithStoreTester::new(event_request_tx);
event_test(
&mut test_harness,
MessageSubtypeId::TcEnableEventGeneration,
EventRequest::Enable(TEST_EVENT_0),
event_request_rx,
);
}
#[test]
fn test_disabling_event_reporting() {
let (event_request_tx, event_request_rx) = mpsc::channel();
let mut test_harness = Pus5HandlerWithStoreTester::new(event_request_tx);
event_test(
&mut test_harness,
MessageSubtypeId::TcDisableEventGeneration,
EventRequest::Disable(TEST_EVENT_0),
event_request_rx,
);
}
#[test]
fn test_empty_tc_queue() {
let (event_request_tx, _) = mpsc::channel();
let mut test_harness = Pus5HandlerWithStoreTester::new(event_request_tx);
let result = test_harness.handle_one_tc();
assert!(result.is_ok());
let result = result.unwrap();
assert!(
matches!(
result,
DirectPusPacketHandlerResult::Handled(HandlingStatus::Empty)
),
"unexpected result type {result:?}"
)
}
#[test]
fn test_sending_custom_subservice() {
let (event_request_tx, _) = mpsc::channel();
let mut test_harness = Pus5HandlerWithStoreTester::new(event_request_tx);
let sp_header = SpHeader::new_for_unseg_tc(TEST_APID, u14::ZERO, 0);
let sec_header = PusTcSecondaryHeader::new_simple(MessageTypeId::new(5, 200));
let ping_tc =
PusTcCreator::new_no_app_data(sp_header, sec_header, CreatorConfig::default());
let token = test_harness.start_verification(&ping_tc);
test_harness.send_tc(&token, &ping_tc);
let result = test_harness.handle_one_tc();
assert!(result.is_ok());
let result = result.unwrap();
if let DirectPusPacketHandlerResult::CustomSubservice(subservice, _) = result {
assert_eq!(subservice, 200);
} else {
panic!("unexpected result type {result:?}")
}
}
#[test]
fn test_sending_invalid_app_data() {
let (event_request_tx, _) = mpsc::channel();
let mut test_harness = Pus5HandlerWithStoreTester::new(event_request_tx);
let sp_header = SpHeader::new_for_unseg_tc(TEST_APID, u14::ZERO, 0);
let sec_header = PusTcSecondaryHeader::new_simple(MessageTypeId::new(
5,
MessageSubtypeId::TcEnableEventGeneration as u8,
));
let ping_tc =
PusTcCreator::new(sp_header, sec_header, &[0, 1, 2], CreatorConfig::default());
let token = test_harness.start_verification(&ping_tc);
test_harness.send_tc(&token, &ping_tc);
let result = test_harness.handle_one_tc();
assert!(result.is_err());
let result = result.unwrap_err();
if let PusPacketHandlingError::RequestConversion(
GenericConversionError::NotEnoughAppData { expected, found },
) = result
{
assert_eq!(expected, 4);
assert_eq!(found, 3);
} else {
panic!("unexpected result type {result:?}")
}
}
}
satrs/src/pus/mod.rs
-4
View File
@@ -25,10 +25,6 @@ use spacepackets::ecss::tm::PusTmCreator;
use spacepackets::{ByteConversionError, SpHeader};
pub mod action;
pub mod event;
pub mod event_man;
#[cfg(feature = "std")]
pub mod event_srv;
pub mod mode;
pub mod scheduler;
#[cfg(feature = "std")]
satrs/src/pus/scheduler_srv.rs
+1 -1
View File
@@ -15,7 +15,7 @@ use std::sync::mpsc;
/// This is a helper class for [std] environments to handle generic PUS 11 (scheduling service)
/// packets. This handler is able to handle the most important PUS requests for a scheduling
/// service which provides the [PusSchedulerProvider].
/// service which provides the [PusScheduler].
///
/// Please note that this class does not do the regular periodic handling like releasing any
/// telecommands inside the scheduler. The user can retrieve the wrapped scheduler via the
satrs/tests/pus_events.rs
-164
View File
@@ -1,164 +0,0 @@
use arbitrary_int::{u11, u21};
use satrs::event_man_legacy::{
EventManagerWithMpsc, EventMessage, EventMessageU32, EventRoutingError, EventSendProvider,
EventU32SenderMpsc,
};
use satrs::events_legacy::{EventU32, EventU32TypedSev, Severity, SeverityInfo};
use satrs::params::U32Pair;
use satrs::params::{Params, ParamsHeapless, WritableToBeBytes};
use satrs::pus::event_man::{DefaultPusEventReportingMap, EventReporter, PusEventTmCreatorWithMap};
use satrs::request::UniqueApidTargetId;
use satrs::tmtc::PacketAsVec;
use spacepackets::ecss::PusError;
use spacepackets::ecss::tm::PusTmReader;
use std::sync::mpsc::{self, SendError, TryRecvError};
use std::thread;
const INFO_EVENT: EventU32TypedSev<SeverityInfo> = EventU32TypedSev::<SeverityInfo>::new(1, 0);
const LOW_SEV_EVENT: EventU32 = EventU32::new(Severity::Low, 1, 5);
const EMPTY_STAMP: [u8; 7] = [0; 7];
const TEST_APID: u11 = u11::new(0x02);
const TEST_ID: UniqueApidTargetId = UniqueApidTargetId::new(TEST_APID, u21::new(0x05));
#[derive(Debug, Clone)]
pub enum CustomTmSenderError {
SendError(SendError<Vec<u8>>),
PusError(PusError),
}
#[test]
fn test_threaded_usage() {
let (event_tx, event_rx) = mpsc::sync_channel(100);
let mut event_man = EventManagerWithMpsc::new(event_rx);
let (pus_event_man_tx, pus_event_man_rx) = mpsc::channel();
let pus_event_man_send_provider = EventU32SenderMpsc::new(1, pus_event_man_tx);
event_man.subscribe_all(pus_event_man_send_provider.target_id());
event_man.add_sender(pus_event_man_send_provider);
let (event_packet_tx, event_packet_rx) = mpsc::channel::<PacketAsVec>();
let reporter = EventReporter::new(TEST_ID.raw(), u11::new(0x02), 0, 128);
let pus_event_man =
PusEventTmCreatorWithMap::new(reporter, DefaultPusEventReportingMap::default());
let error_handler = |event_msg: &EventMessageU32, error: EventRoutingError| {
panic!("received routing error for event {event_msg:?}: {error:?}");
};
// PUS + Generic event manager thread
let jh0 = thread::spawn(move || {
let mut event_cnt = 0;
let mut params_array: [u8; 128] = [0; 128];
loop {
event_man.try_event_handling(error_handler);
match pus_event_man_rx.try_recv() {
Ok(event_msg) => {
let gen_event = |aux_data| {
pus_event_man.generate_pus_event_tm_generic(
&event_packet_tx,
&EMPTY_STAMP,
event_msg.event(),
aux_data,
)
};
let res = if let Some(aux_data) = event_msg.params() {
match aux_data {
Params::Heapless(heapless) => match heapless {
ParamsHeapless::Raw(raw) => {
raw.write_to_be_bytes(&mut params_array)
.expect("Writing raw parameter failed");
gen_event(Some(&params_array[0..raw.written_len()]))
}
ParamsHeapless::EcssEnum(e) => {
e.write_to_be_bytes(&mut params_array)
.expect("Writing ECSS enum failed");
gen_event(Some(&params_array[0..e.written_len()]))
}
},
Params::Vec(vec) => gen_event(Some(vec.as_slice())),
Params::String(str) => gen_event(Some(str.as_bytes())),
Params::Store(_) => gen_event(None),
_ => panic!("unsupported parameter type"),
}
} else {
gen_event(None)
};
event_cnt += 1;
assert!(res.is_ok());
assert!(res.unwrap());
if event_cnt == 2 {
break;
}
}
Err(e) => {
if let TryRecvError::Disconnected = e {
panic!("Event receiver disconnected!")
}
}
}
}
});
// Event sender and TM checker thread
let jh1 = thread::spawn(move || {
event_tx
.send(EventMessage::new(TEST_ID.id(), INFO_EVENT.into()))
.expect("Sending info event failed");
loop {
match event_packet_rx.try_recv() {
// Event TM received successfully
Ok(event_tm) => {
let tm = PusTmReader::new(event_tm.packet.as_slice(), 7)
.expect("Deserializing TM failed");
assert_eq!(tm.service_type_id(), 5);
assert_eq!(tm.message_subtype_id(), 1);
let src_data = tm.source_data();
assert!(!src_data.is_empty());
assert_eq!(src_data.len(), 4);
let event =
EventU32::from(u32::from_be_bytes(src_data[0..4].try_into().unwrap()));
assert_eq!(event, INFO_EVENT);
break;
}
Err(e) => {
if let TryRecvError::Disconnected = e {
panic!("Event sender disconnected!")
}
}
}
}
event_tx
.send(EventMessage::new_with_params(
TEST_ID.id(),
LOW_SEV_EVENT,
&Params::Heapless((2_u32, 3_u32).into()),
))
.expect("Sending low severity event failed");
loop {
match event_packet_rx.try_recv() {
// Event TM received successfully
Ok(event_tm) => {
let tm = PusTmReader::new(event_tm.packet.as_slice(), 7)
.expect("Deserializing TM failed");
assert_eq!(tm.service_type_id(), 5);
assert_eq!(tm.message_subtype_id(), 2);
let src_data = tm.source_data();
assert!(!src_data.is_empty());
assert_eq!(src_data.len(), 12);
let event =
EventU32::from(u32::from_be_bytes(src_data[0..4].try_into().unwrap()));
assert_eq!(event, LOW_SEV_EVENT);
let u32_pair: U32Pair =
src_data[4..].try_into().expect("Creating U32Pair failed");
assert_eq!(u32_pair.0, 2);
assert_eq!(u32_pair.1, 3);
break;
}
Err(e) => {
if let TryRecvError::Disconnected = e {
panic!("Event sender disconnected!")
}
}
}
}
});
jh0.join().expect("Joining manager thread failed");
jh1.join().expect("Joining creator thread failed");
}