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

continue event service integration

Browse Source
This commit is contained in:
Robin Müller
2022-11-13 21:07:16 +01:00
parent 8b110be07a
commit 9cc58432c8
5 changed files with 195 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
fsrc-example/src/bin/obsw/main.rs
View File

@ -2,20 +2,25 @@ mod ccsds;
mod pus;
mod tmtc;
use crate::tmtc::{core_tmtc_task, TmStore, PUS_APID};
use crate::tmtc::{core_tmtc_task, CoreTmtcArgs, TmStore, PUS_APID};
use fsrc_core::event_man::{EventManager, MpscEventReceiver, MpscEventU32SendProvider};
use fsrc_core::events::EventU32;
use fsrc_core::hal::host::udp_server::UdpTcServer;
use fsrc_core::params::Params;
use fsrc_core::pool::{LocalPool, PoolCfg, SharedPool, StoreAddr};
use fsrc_core::pus::event_man::{DefaultPusMgmtBackendProvider, EventReporter, PusEventDispatcher};
use fsrc_core::pus::event_man::{
DefaultPusMgmtBackendProvider, EventReporter, EventRequest, EventRequestWithToken,
PusEventDispatcher,
};
use fsrc_core::pus::verification::{
MpscVerifSender, VerificationReporterCfg, VerificationReporterWithSender,
};
use fsrc_core::pus::{EcssTmError, EcssTmSender};
use fsrc_core::tmtc::CcsdsError;
use fsrc_example::{OBSW_SERVER_ADDR, SERVER_PORT};
use spacepackets::time::{CdsShortTimeProvider, TimeWriter};
use spacepackets::tm::PusTm;
use std::net::{IpAddr, SocketAddr};
use std::sync::mpsc::{channel, TryRecvError};
use std::sync::mpsc::channel;
use std::sync::{mpsc, Arc, Mutex, RwLock};
use std::thread;
@ -32,6 +37,28 @@ struct UdpTmtcServer {
unsafe impl Send for UdpTmtcServer {}
struct EventTmSender {
store_helper: TmStore,
sender: mpsc::Sender<StoreAddr>,
}
impl EventTmSender {
fn new(store_helper: TmStore, sender: mpsc::Sender<StoreAddr>) -> Self {
Self {
store_helper,
sender,
}
}
}
impl EcssTmSender for EventTmSender {
type Error = mpsc::SendError<StoreAddr>;
fn send_tm(&mut self, tm: PusTm) -> Result<(), EcssTmError<Self::Error>> {
let addr = self.store_helper.add_pus_tm(&tm);
self.sender.send(addr).map_err(EcssTmError::SendError)
}
}
fn main() {
println!("Running OBSW example");
let pool_cfg = PoolCfg::new(vec![(8, 32), (4, 64), (2, 128)]);
@ -41,32 +68,38 @@ fn main() {
pool: tm_store.clone(),
};
let addr = SocketAddr::new(IpAddr::V4(OBSW_SERVER_ADDR), SERVER_PORT);
let (tm_funnel_tx, tm_funnel_rx) = mpsc::channel();
let (tm_server_tx, tm_server_rx) = mpsc::channel();
let (tm_funnel_tx, tm_funnel_rx) = channel();
let (tm_server_tx, tm_server_rx) = channel();
let sender = MpscVerifSender::new(tm_store.clone(), tm_funnel_tx.clone());
let verif_cfg = VerificationReporterCfg::new(PUS_APID, 1, 2, 8).unwrap();
let reporter_with_sender_0 = Arc::new(Mutex::new(VerificationReporterWithSender::new(
verif_cfg,
Box::new(sender),
)));
// Create event handling components
let (event_request_tx, event_request_rx) = channel::<EventRequestWithToken>();
let (event_sender, event_man_rx) = channel();
let event_recv = MpscEventReceiver::<EventU32>::new(event_man_rx);
let mut event_man = EventManager::new(Box::new(event_recv));
let event_reporter = EventReporter::new(PUS_APID, 128).unwrap();
let pus_tm_backend = DefaultPusMgmtBackendProvider::<EventU32>::default();
let pus_event_man = PusEventDispatcher::new(event_reporter, Box::new(pus_tm_backend));
let mut pus_event_dispatcher =
PusEventDispatcher::new(event_reporter, Box::new(pus_tm_backend));
let (pus_event_man_tx, pus_event_man_rx) = channel();
let pus_event_man_send_provider = MpscEventU32SendProvider::new(1, pus_event_man_tx);
let reporter1 = reporter_with_sender_0.clone();
event_man.subscribe_all(pus_event_man_send_provider);
// Create clones here to allow move for thread 0
let core_args = CoreTmtcArgs {
tm_store: tm_store_helper.clone(),
tm_sender: tm_funnel_tx.clone(),
event_sender,
event_request_tx,
};
let jh0 = thread::spawn(move || {
core_tmtc_task(
tm_funnel_tx.clone(),
tm_server_rx,
tm_store_helper.clone(),
addr,
reporter_with_sender_0.clone(),
event_sender.clone(),
);
core_tmtc_task(core_args, tm_server_rx, addr, reporter_with_sender_0);
});
let jh1 = thread::spawn(move || {
@ -84,10 +117,43 @@ fn main() {
}
});
let jh2 = thread::spawn(move || loop {
match pus_event_man_rx.try_recv() {
Ok(_) => {}
Err(_) => {}
let jh2 = thread::spawn(move || {
let mut timestamp: [u8; 7] = [0; 7];
let mut sender = EventTmSender::new(tm_store_helper, tm_funnel_tx);
let mut time_provider = CdsShortTimeProvider::new(0, 0);
let report_completion = |event_req: EventRequestWithToken, timestamp: &[u8]| {
let mut reporter = reporter1
.lock()
.expect("Locking Verification reporter failed");
reporter
.completion_success(event_req.token, timestamp)
.expect("Sending completion success failed");
};
loop {
if let Ok(event_req) = event_request_rx.try_recv() {
match event_req.request {
EventRequest::Enable(event) => {
pus_event_dispatcher
.enable_tm_for_event(&event)
.expect("Enabling TM failed");
update_time(&mut time_provider, &mut timestamp);
report_completion(event_req, &timestamp);
}
EventRequest::Disable(event) => {
pus_event_dispatcher
.disable_tm_for_event(&event)
.expect("Disabling TM failed");
update_time(&mut time_provider, &mut timestamp);
report_completion(event_req, &timestamp);
}
}
}
if let Ok((event, _param)) = pus_event_man_rx.try_recv() {
update_time(&mut time_provider, &mut timestamp);
pus_event_dispatcher
.generate_pus_event_tm_generic(&mut sender, &timestamp, event, None)
.expect("Sending TM as event failed");
}
}
});
@ -95,3 +161,12 @@ fn main() {
jh1.join().expect("Joining TM Funnel thread failed");
jh2.join().expect("Joining Event Manager thread failed");
}
pub fn update_time(time_provider: &mut CdsShortTimeProvider, timestamp: &mut [u8]) {
time_provider
.update_from_now()
.expect("Could not get current time");
time_provider
.write_to_bytes(timestamp)
.expect("Writing timestamp failed");
}

4
fsrc-example/src/bin/obsw/pus.rs
View File

@ -1,7 +1,7 @@
use crate::tmtc::TmStore;
use fsrc_core::pool::StoreAddr;
use fsrc_core::pus::verification::{
SharedStdVerifReporterWithSender, StateAccepted, VerificationToken,
SharedStdVerifReporterWithSender, TcStateAccepted, VerificationToken,
};
use fsrc_core::tmtc::tm_helper::PusTmWithCdsShortHelper;
use fsrc_core::tmtc::PusServiceProvider;
@ -71,7 +71,7 @@ impl PusServiceProvider for PusReceiver {
}
impl PusReceiver {
fn handle_test_service(&mut self, pus_tc: &PusTc, token: VerificationToken<StateAccepted>) {
fn handle_test_service(&mut self, pus_tc: &PusTc, token: VerificationToken<TcStateAccepted>) {
if pus_tc.subservice() == 1 {
println!("Received PUS ping command TC[17,1]");
println!("Sending ping reply PUS TM[17,2]");

18
fsrc-example/src/bin/obsw/tmtc.rs
View File

@ -11,12 +11,20 @@ use crate::ccsds::CcsdsReceiver;
use crate::pus::PusReceiver;
use crate::UdpTmtcServer;
use fsrc_core::pool::{SharedPool, StoreAddr};
use fsrc_core::pus::event_man::EventRequestWithToken;
use fsrc_core::pus::verification::SharedStdVerifReporterWithSender;
use fsrc_core::tmtc::{CcsdsDistributor, CcsdsError, PusDistributor};
use spacepackets::tm::PusTm;
pub const PUS_APID: u16 = 0x02;
pub struct CoreTmtcArgs {
pub tm_store: TmStore,
pub tm_sender: Sender<StoreAddr>,
pub event_sender: Sender<(EventU32, Option<Params>)>,
pub event_request_tx: Sender<EventRequestWithToken>,
}
#[derive(Clone)]
pub struct TmStore {
pub pool: SharedPool,
@ -34,17 +42,15 @@ impl TmStore {
}
pub fn core_tmtc_task(
tm_creator_tx: Sender<StoreAddr>,
args: CoreTmtcArgs,
tm_server_rx: mpsc::Receiver<StoreAddr>,
tm_store_helper: TmStore,
addr: SocketAddr,
verif_reporter: SharedStdVerifReporterWithSender,
_event_tx: Sender<(EventU32, Option<Params>)>,
) {
let pus_receiver = PusReceiver::new(
PUS_APID,
tm_creator_tx,
tm_store_helper.clone(),
args.tm_sender,
args.tm_store.clone(),
verif_reporter,
);
let pus_distributor = PusDistributor::new(Box::new(pus_receiver));
@ -58,7 +64,7 @@ pub fn core_tmtc_task(
let mut udp_tmtc_server = UdpTmtcServer {
udp_tc_server,
tm_rx: tm_server_rx,
tm_store: tm_store_helper.pool.clone(),
tm_store: args.tm_store.pool.clone(),
};
loop {
core_tmtc_loop(&mut udp_tmtc_server);