rust/sat-rs
rust/sat-rs
5
1
Fork 0
Code Issues 9 Pull Requests Projects Releases 22 Wiki Activity

start adding docs
Some checks failed
Rust/sat-rs/pipeline/pr-main There was a failure building this commit
Details

Browse Source
This commit is contained in:
Robin Müller 2024-03-12 15:49:07 +01:00
parent 1fbb400809
commit 41db3b86da
3 changed files with 229 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
satrs/src/event_man.rs
View File

@ -101,40 +101,40 @@ pub trait ListenerMapProvider {
}
pub trait SenderMapProvider<
SP: EventSendProvider<EV, AUX>,
EV: GenericEvent = EventU32,
AUX = Params,
EventSender: EventSendProvider<Ev, Data>,
Ev: GenericEvent = EventU32,
Data = Params,
>
{
fn contains_send_event_provider(&self, id: &ChannelId) -> bool;
fn get_send_event_provider(&self, id: &ChannelId) -> Option<&SP>;
fn add_send_event_provider(&mut self, send_provider: SP) -> bool;
fn get_send_event_provider(&self, id: &ChannelId) -> Option<&EventSender>;
fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool;
}
/// Generic event manager implementation.
///
/// # Generics
///
/// * `ERP`: [EventReceiveProvider] used to receive all events.
/// * `SMP`: [SenderMapProvider] which maps channel IDs to send providers.
/// * `LTR`: [ListenerMapProvider] which maps listener keys to channel IDs.
/// * `SP`: [EventSendProvider] contained within the sender map which sends the events.
/// * `EV`: The event type. This type must implement the [GenericEvent]. Currently only [EventU32]
/// * `EventReceiver`: [EventReceiveProvider] used to receive all events.
/// * `SenderMap`: [SenderMapProvider] which maps channel IDs to send providers.
/// * `ListenerMap`: [ListenerMapProvider] which maps listener keys to channel IDs.
/// * `EventSender`: [EventSendProvider] contained within the sender map which sends the events.
/// * `Ev`: The event type. This type must implement the [GenericEvent]. Currently only [EventU32]
/// and [EventU16] are supported.
/// * `AUX`: Auxiliary data which is sent with the event to provide optional context information
/// * `Data`: Auxiliary data which is sent with the event to provide optional context information
pub struct EventManager<
ERP: EventReceiveProvider<EV, AUX>,
SMP: SenderMapProvider<SP, EV, AUX>,
LTR: ListenerMapProvider,
SP: EventSendProvider<EV, AUX>,
EV: GenericEvent = EventU32,
AUX = Params,
EventReceiver: EventReceiveProvider<Ev, Data>,
SenderMap: SenderMapProvider<EventSender, Ev, Data>,
ListenerMap: ListenerMapProvider,
EventSender: EventSendProvider<Ev, Data>,
Ev: GenericEvent = EventU32,
Data = Params,
> {
event_receiver: ERP,
sender_map: SMP,
listener_map: LTR,
phantom: core::marker::PhantomData<(SP, EV, AUX)>,
event_receiver: EventReceiver,
sender_map: SenderMap,
listener_map: ListenerMap,
phantom: core::marker::PhantomData<(EventSender, Ev, Data)>,
}
#[derive(Debug)]
@ -157,26 +157,26 @@ pub enum EventRoutingError {
}
#[derive(Debug)]
pub struct EventRoutingErrorsWithResult<EV: GenericEvent, AUX> {
pub result: EventRoutingResult<EV, AUX>,
pub struct EventRoutingErrorsWithResult<Ev: GenericEvent, Data> {
pub result: EventRoutingResult<Ev, Data>,
pub errors: [Option<EventRoutingError>; 3],
}
impl<
ER: EventReceiveProvider<EV, AUX>,
S: SenderMapProvider<SP, EV, AUX>,
L: ListenerMapProvider,
SP: EventSendProvider<EV, AUX>,
EV: GenericEvent + Copy,
AUX: Clone,
> EventManager<ER, S, L, SP, EV, AUX>
EventReceiver: EventReceiveProvider<Ev, Data>,
SenderMap: SenderMapProvider<EventSender, Ev, Data>,
ListenerMap: ListenerMapProvider,
EventSender: EventSendProvider<Ev, Data>,
Ev: GenericEvent + Copy,
Data: Clone,
> EventManager<EventReceiver, SenderMap, ListenerMap, EventSender, Ev, Data>
{
pub fn remove_duplicates(&mut self, key: &ListenerKey) {
self.listener_map.remove_duplicates(key)
}
/// Subscribe for a unique event.
pub fn subscribe_single(&mut self, event: &EV, sender_id: ChannelId) {
pub fn subscribe_single(&mut self, event: &Ev, sender_id: ChannelId) {
self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id);
}

271
satrs/src/pus/action.rs
View File

@ -129,15 +129,13 @@ pub mod std_mod {
verification::{
self, FailParams, FailParamsWithStep, TcStateStarted, VerificationReportingProvider,
},
ActiveRequestMapProvider, ActiveRequestProvider, DefaultActiveRequestMap,
EcssTcInMemConverter, EcssTcReceiverCore, EcssTmSenderCore, EcssTmtcError,
GenericRoutingError, PusPacketHandlerResult, PusPacketHandlingError,
PusRoutingErrorHandler, PusServiceHelper,
ActiveRequestMapProvider, DefaultActiveRequestMap, EcssTcInMemConverter,
EcssTcReceiverCore, EcssTmSenderCore, EcssTmtcError, GenericRoutingError,
PusPacketHandlerResult, PusPacketHandlingError, PusRoutingErrorHandler,
PusServiceHelper, PusServiceReplyHandler, ReplyHandlerHook,
},
request::RequestId,
};
use core::{marker::PhantomData, time::Duration};
use delegate::delegate;
use core::time::Duration;
use spacepackets::time::UnixTimestamp;
use std::time::SystemTimeError;
@ -252,160 +250,17 @@ pub mod std_mod {
pub type DefaultActiveActionRequestMap = DefaultActiveRequestMap<ActiveActionRequest>;
pub trait ReplyHandlerHook<ActiveRequestType, Reply> {
fn handle_unexpected_reply(&mut self, reply: &Reply);
fn timeout_callback(&self, active_request: &ActiveRequestType);
fn timeout_error_code(&self) -> ResultU16;
}
pub struct PusServiceReplyHandler<
VerificationReporter: VerificationReportingProvider,
ActiveRequestMap: ActiveRequestMapProvider<ActiveRequestType>,
UserHook: ReplyHandlerHook<ActiveRequestType, Reply>,
ActiveRequestType: ActiveRequestProvider,
Reply,
> {
active_request_map: ActiveRequestMap,
verification_reporter: VerificationReporter,
fail_data_buf: alloc::vec::Vec<u8>,
current_time: UnixTimestamp,
pub user_hook: UserHook,
phantom: PhantomData<(ActiveRequestType, Reply)>,
}
impl<
VerificationReporter: VerificationReportingProvider,
ActiveRequestMap: ActiveRequestMapProvider<ActiveRequestType>,
UserHook: ReplyHandlerHook<ActiveRequestType, ReplyType>,
ActiveRequestType: ActiveRequestProvider,
ReplyType,
>
PusServiceReplyHandler<
VerificationReporter,
ActiveRequestMap,
UserHook,
ActiveRequestType,
ReplyType,
>
{
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn new_from_now(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
) -> Result<Self, SystemTimeError> {
let current_time = UnixTimestamp::from_now()?;
Ok(Self::new(
verification_reporter,
active_request_map,
fail_data_buf_size,
user_hook,
current_time,
))
}
pub fn new(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
init_time: UnixTimestamp,
) -> Self {
Self {
active_request_map,
verification_reporter,
fail_data_buf: alloc::vec![0; fail_data_buf_size],
current_time: init_time,
user_hook,
phantom: PhantomData,
}
}
pub fn add_routed_request(
&mut self,
request_id: verification::RequestId,
active_request_type: ActiveRequestType,
) {
self.active_request_map
.insert(&request_id.into(), active_request_type);
}
pub fn request_active(&self, request_id: RequestId) -> bool {
self.active_request_map.get(request_id).is_some()
}
/// Check for timeouts across all active requests.
///
/// It will call [Self::handle_timeout] for all active requests which have timed out.
pub fn check_for_timeouts(&mut self, time_stamp: &[u8]) -> Result<(), EcssTmtcError> {
let mut timed_out_commands = alloc::vec::Vec::new();
self.active_request_map.for_each(|request_id, active_req| {
let diff = self.current_time - active_req.start_time();
if diff.duration_absolute > active_req.timeout() {
self.handle_timeout(active_req, time_stamp);
}
timed_out_commands.push(*request_id);
});
for timed_out_command in timed_out_commands {
self.active_request_map.remove(timed_out_command);
}
Ok(())
}
/// Handle the timeout for a given active request.
///
/// This implementation will report a verification completion failure with a user-provided
/// error code. It supplies the configured request timeout in milliseconds as a [u64]
/// serialized in big-endian format as the failure data.
pub fn handle_timeout(&self, active_request: &ActiveRequestType, time_stamp: &[u8]) {
let timeout = active_request.timeout().as_millis() as u64;
let timeout_raw = timeout.to_be_bytes();
self.verification_reporter
.completion_failure(
active_request.token(),
FailParams::new(
time_stamp,
&self.user_hook.timeout_error_code(),
&timeout_raw,
),
)
.unwrap();
self.user_hook.timeout_callback(active_request);
}
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn update_time_from_now(&mut self) -> Result<(), SystemTimeError> {
self.current_time = UnixTimestamp::from_now()?;
Ok(())
}
}
/// Type definition for a PUS 8 action service reply handler which constrains the
/// [PusServiceReplyHandler] active request and reply generics to the [ActiveActionRequest] and
/// [ActionReplyPusWithIds] type.
pub type PusService8ReplyHandler<VerificationReporter, ActiveRequestMap, UserHook> =
PusServiceReplyHandler<
VerificationReporter,
DefaultActiveActionRequestMap,
ActiveRequestMap,
UserHook,
ActiveActionRequest,
ActionReplyPusWithIds,
>;
/*
pub struct PusService8ReplyHandler<
VerificationReporter: VerificationReportingProvider,
ActiveRequestMap: ActiveRequestMapProvider<ActiveActionRequest>,
UserHook: ReplyHandlerHook<ActiveActionRequest, ActionReplyPusWithIds>,
> {
pub inner: PusServiceReplyHandler<
VerificationReporter,
ActiveRequestMap,
UserHook,
ActiveActionRequest,
ActionReplyPusWithIds,
>,
}
*/
impl<
VerificationReporter: VerificationReportingProvider,
@ -413,43 +268,8 @@ pub mod std_mod {
UserHook: ReplyHandlerHook<ActiveActionRequest, ActionReplyPusWithIds>,
> PusService8ReplyHandler<VerificationReporter, ActiveRequestMap, UserHook>
{
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn new_from_now(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
) -> Result<Self, SystemTimeError> {
let current_time = UnixTimestamp::from_now()?;
Ok(Self::new(
verification_reporter,
active_request_map,
fail_data_buf_size,
user_hook,
current_time,
))
}
pub fn new(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
init_time: UnixTimestamp,
) -> Self {
Self {
inner: PusServiceReplyHandler::new(
verification_reporter,
active_request_map,
fail_data_buf_size,
user_hook,
init_time,
),
}
}
pub fn add_routed_request(
/// Helper method to register a recently routed action request.
pub fn add_routed_action_request(
&mut self,
request_id: verification::RequestId,
target_id: TargetId,
@ -457,68 +277,45 @@ pub mod std_mod {
token: VerificationToken<TcStateStarted>,
timeout: Duration,
) {
self.inner.active_request_map.insert(
self.active_request_map.insert(
&request_id.into(),
ActiveActionRequest {
action_id,
common: ActiveRequest {
target_id,
token,
start_time: self.inner.current_time,
start_time: self.current_time,
timeout,
},
},
);
}
delegate! {
to self.inner {
pub fn request_active(&self, request_id: RequestId) -> bool;
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn update_time_from_now(&mut self) -> Result<(), SystemTimeError>;
/// Check for timeouts across all active requests.
///
/// It will call [Self::handle_timeout] for all active requests which have timed out.
pub fn check_for_timeouts(&mut self, time_stamp: &[u8]) -> Result<(), EcssTmtcError>;
/// Handle the timeout for a given active request.
///
/// This implementation will report a verification completion failure with a user-provided
/// error code. It supplies the configured request timeout in milliseconds as a [u64]
/// serialized in big-endian format as the failure data.
pub fn handle_timeout(&self, active_request: &ActiveActionRequest, time_stamp: &[u8]);
}
}
/// Main handler function to handle all received action replies.
pub fn handle_action_reply(
&mut self,
action_reply_with_ids: ActionReplyPusWithIds,
time_stamp: &[u8],
) -> Result<(), EcssTmtcError> {
let active_req = self
.inner
.active_request_map
.get(action_reply_with_ids.request_id);
if active_req.is_none() {
self.inner
.user_hook
self.user_hook
.handle_unexpected_reply(&action_reply_with_ids);
return Ok(());
}
let active_req = active_req.unwrap().clone();
let remove_entry = match action_reply_with_ids.reply {
ActionReplyPus::CompletionFailed { error_code, params } => {
let fail_data_len = params.write_to_be_bytes(&mut self.inner.fail_data_buf)?;
self.inner
.verification_reporter
let fail_data_len = params.write_to_be_bytes(&mut self.fail_data_buf)?;
self.verification_reporter
.completion_failure(
active_req.common.token,
FailParams::new(
time_stamp,
&error_code,
&self.inner.fail_data_buf[..fail_data_len],
&self.fail_data_buf[..fail_data_len],
),
)
.map_err(|e| e.0)?;
@ -529,30 +326,28 @@ pub mod std_mod {
step,
params,
} => {
let fail_data_len = params.write_to_be_bytes(&mut self.inner.fail_data_buf)?;
self.inner
.verification_reporter
let fail_data_len = params.write_to_be_bytes(&mut self.fail_data_buf)?;
self.verification_reporter
.step_failure(
active_req.common.token,
FailParamsWithStep::new(
time_stamp,
&EcssEnumU16::new(step),
&error_code,
&self.inner.fail_data_buf[..fail_data_len],
&self.fail_data_buf[..fail_data_len],
),
)
.map_err(|e| e.0)?;
true
}
ActionReplyPus::Completed => {
self.inner
.verification_reporter
self.verification_reporter
.completion_success(active_req.common.token, time_stamp)
.map_err(|e| e.0)?;
true
}
ActionReplyPus::StepSuccess { step } => {
self.inner.verification_reporter.step_success(
self.verification_reporter.step_success(
&active_req.common.token,
time_stamp,
EcssEnumU16::new(step),
@ -561,8 +356,7 @@ pub mod std_mod {
}
};
if remove_entry {
self.inner
.active_request_map
self.active_request_map
.remove(action_reply_with_ids.request_id);
}
Ok(())
@ -574,6 +368,9 @@ pub mod std_mod {
UserHook: ReplyHandlerHook<ActiveActionRequest, ActionReplyPusWithIds>,
> PusService8ReplyHandler<VerificationReporter, DefaultActiveActionRequestMap, UserHook>
{
/// Create a new PUS Service 8 reply handler with the [ActiveRequestMap] generic
/// constrained to the [DefaultActiveActionRequestMap] object and with the current time
/// set to the OS time.
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn new_from_now_with_default_map(
@ -590,6 +387,8 @@ pub mod std_mod {
))
}
/// Create a new PUS Service 8 reply handler with the [ActiveRequestMap] generic
/// constrained to the [DefaultActiveActionRequestMap] object.
pub fn new_with_default_map(
verification_reporter: VerificationReporter,
fail_data_buf_size: usize,
@ -638,7 +437,8 @@ mod tests {
FailParams, TcStateNone, TcStateStarted, VerificationReportingProvider,
},
EcssTcInVecConverter, EcssTmtcError, GenericRoutingError, MpscTcReceiver,
PusPacketHandlerResult, PusPacketHandlingError, TmAsVecSenderWithMpsc,
PusPacketHandlerResult, PusPacketHandlingError, ReplyHandlerHook,
TmAsVecSenderWithMpsc,
},
};
@ -859,12 +659,7 @@ mod tests {
}
pub fn next_unrequested_reply(&self) -> Option<ActionReplyPusWithIds> {
self.handler
.inner
.user_hook
.unexpected_replies
.front()
.cloned()
self.handler.user_hook.unexpected_replies.front().cloned()
}
pub fn assert_request_completion_success(&self, step: Option<u16>, request_id: RequestId) {
@ -931,7 +726,7 @@ mod tests {
panic!("request already present");
}
self.handler
.add_routed_request(request_id, target_id, action_id, token, timeout);
.add_routed_action_request(request_id, target_id, action_id, token, timeout);
if !self.handler.request_active(request_id.into()) {
panic!("request should be active now");
}

166
satrs/src/pus/mod.rs
View File

@ -8,11 +8,13 @@ use crate::queue::{GenericReceiveError, GenericSendError};
use crate::request::RequestId;
use crate::{ChannelId, TargetId};
use core::fmt::{Display, Formatter};
use core::marker::PhantomData;
use core::time::Duration;
#[cfg(feature = "alloc")]
use downcast_rs::{impl_downcast, Downcast};
#[cfg(feature = "alloc")]
use dyn_clone::DynClone;
use satrs_shared::res_code::ResultU16;
use spacepackets::time::UnixTimestamp;
#[cfg(feature = "std")]
use std::error::Error;
@ -42,7 +44,7 @@ pub use alloc_mod::*;
#[cfg(feature = "std")]
pub use std_mod::*;
use self::verification::TcStateStarted;
use self::verification::{FailParams, TcStateStarted, VerificationReportingProvider};
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum PusTmWrapper<'tm> {
@ -464,6 +466,168 @@ mod alloc_mod {
}
}
/// Generic user hook method.
///
/// This hook method currently serves the following tasks:
///
/// 1. Pass specific information to the reply handlers which can not be kept inside the
/// framework. This includes information like the error codes used for packet verification.
/// 2. It exposes callback methods which can be useful to perform custom user operations like
/// logging.
pub trait ReplyHandlerHook<ActiveRequestType, ReplyType> {
fn handle_unexpected_reply(&mut self, reply: &ReplyType);
fn timeout_callback(&self, active_request: &ActiveRequestType);
fn timeout_error_code(&self) -> ResultU16;
}
/// Generic reply handler structure which can be used to handle replies for a specific PUS service.
///
/// This is done by keeping track of active requests using an internal map structure. An API
/// to register new active requests is exposed as well.
/// The reply handler performs boilerplate tasks like performing the verification handling and
/// timeout handling.
///
/// This object is not useful by itself but serves as a common building block for high-level
/// PUS reply handlers. Concrete PUS handlers should constrain the [ActiveRequestProvider] and
/// the `ReplyType` generics to specific types tailored towards PUS services in addition to
/// providing an API which can process received replies and convert them into verification
/// completions or other operation like user hook calls. The framework also provides some concrete
/// PUS handlers for common PUS services like the mode, action and housekeeping service.
///
/// This object does not automatically update its internal time information used to check for
/// timeouts. The user should call the [Self::update_time] and [Self::update_time_from_now] methods
/// to do this.
pub struct PusServiceReplyHandler<
VerificationReporter: VerificationReportingProvider,
ActiveRequestMap: ActiveRequestMapProvider<ActiveRequestType>,
UserHook: ReplyHandlerHook<ActiveRequestType, ReplyType>,
ActiveRequestType: ActiveRequestProvider,
ReplyType,
> {
active_request_map: ActiveRequestMap,
verification_reporter: VerificationReporter,
fail_data_buf: alloc::vec::Vec<u8>,
current_time: UnixTimestamp,
pub user_hook: UserHook,
phantom: PhantomData<(ActiveRequestType, ReplyType)>,
}
impl<
VerificationReporter: VerificationReportingProvider,
ActiveRequestMap: ActiveRequestMapProvider<ActiveRequestType>,
UserHook: ReplyHandlerHook<ActiveRequestType, ReplyType>,
ActiveRequestType: ActiveRequestProvider,
ReplyType,
>
PusServiceReplyHandler<
VerificationReporter,
ActiveRequestMap,
UserHook,
ActiveRequestType,
ReplyType,
>
{
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn new_from_now(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
) -> Result<Self, std::time::SystemTimeError> {
let current_time = UnixTimestamp::from_now()?;
Ok(Self::new(
verification_reporter,
active_request_map,
fail_data_buf_size,
user_hook,
current_time,
))
}
pub fn new(
verification_reporter: VerificationReporter,
active_request_map: ActiveRequestMap,
fail_data_buf_size: usize,
user_hook: UserHook,
init_time: UnixTimestamp,
) -> Self {
Self {
active_request_map,
verification_reporter,
fail_data_buf: alloc::vec![0; fail_data_buf_size],
current_time: init_time,
user_hook,
phantom: PhantomData,
}
}
pub fn add_routed_request(
&mut self,
request_id: verification::RequestId,
active_request_type: ActiveRequestType,
) {
self.active_request_map
.insert(&request_id.into(), active_request_type);
}
pub fn request_active(&self, request_id: RequestId) -> bool {
self.active_request_map.get(request_id).is_some()
}
/// Check for timeouts across all active requests.
///
/// It will call [Self::handle_timeout] for all active requests which have timed out.
pub fn check_for_timeouts(&mut self, time_stamp: &[u8]) -> Result<(), EcssTmtcError> {
let mut timed_out_commands = alloc::vec::Vec::new();
self.active_request_map.for_each(|request_id, active_req| {
let diff = self.current_time - active_req.start_time();
if diff.duration_absolute > active_req.timeout() {
self.handle_timeout(active_req, time_stamp);
}
timed_out_commands.push(*request_id);
});
for timed_out_command in timed_out_commands {
self.active_request_map.remove(timed_out_command);
}
Ok(())
}
/// Handle the timeout for a given active request.
///
/// This implementation will report a verification completion failure with a user-provided
/// error code. It supplies the configured request timeout in milliseconds as a [u64]
/// serialized in big-endian format as the failure data.
pub fn handle_timeout(&self, active_request: &ActiveRequestType, time_stamp: &[u8]) {
let timeout = active_request.timeout().as_millis() as u64;
let timeout_raw = timeout.to_be_bytes();
self.verification_reporter
.completion_failure(
active_request.token(),
FailParams::new(
time_stamp,
&self.user_hook.timeout_error_code(),
&timeout_raw,
),
)
.unwrap();
self.user_hook.timeout_callback(active_request);
}
/// Update the current time used for timeout checks based on the current OS time.
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub fn update_time_from_now(&mut self) -> Result<(), std::time::SystemTimeError> {
self.current_time = UnixTimestamp::from_now()?;
Ok(())
}
/// Update the current time used for timeout checks.
pub fn update_time(&mut self, time: UnixTimestamp) {
self.current_time = time;
}
}
#[cfg(feature = "std")]
#[cfg_attr(doc_cfg, doc(cfg(feature = "std")))]
pub mod std_mod {