Updating after review

This commit is contained in:
Ryan Summers 2021-07-27 13:12:57 +02:00
parent 7a4f73d558
commit de63be09e4
5 changed files with 54 additions and 54 deletions

View File

@ -22,7 +22,7 @@ HEADER_FORMAT = '<HBBI'
# All supported formats by this reception script. # All supported formats by this reception script.
# #
# The items in this dict are functions that will be provided the sampel batch size and will return # The items in this dict are functions that will be provided the sample batch size and will return
# the struct deserialization code to unpack a single batch. # the struct deserialization code to unpack a single batch.
FORMAT = { FORMAT = {
1: lambda batch_size: f'<{batch_size}H{batch_size}H{batch_size}H{batch_size}H' 1: lambda batch_size: f'<{batch_size}H{batch_size}H{batch_size}H{batch_size}H'
@ -32,7 +32,7 @@ def parse_packet(buf):
""" Attempt to parse packets from the received buffer. """ """ Attempt to parse packets from the received buffer. """
# Attempt to parse a block from the buffer. # Attempt to parse a block from the buffer.
if len(buf) < struct.calcsize(HEADER_FORMAT): if len(buf) < struct.calcsize(HEADER_FORMAT):
return [] return
# Parse out the packet header # Parse out the packet header
magic, format_id, batch_size, sequence_number = struct.unpack_from(HEADER_FORMAT, buf) magic, format_id, batch_size, sequence_number = struct.unpack_from(HEADER_FORMAT, buf)
@ -40,10 +40,7 @@ def parse_packet(buf):
if magic != MAGIC_HEADER: if magic != MAGIC_HEADER:
logging.warning('Encountered bad magic header: %s', hex(magic)) logging.warning('Encountered bad magic header: %s', hex(magic))
return [] return
if format_id not in FORMAT:
raise Exception(f'Unknown format specifier: {format_id}')
frame_format = FORMAT[format_id](batch_size) frame_format = FORMAT[format_id](batch_size)
@ -53,9 +50,8 @@ def parse_packet(buf):
for offset in range(batch_count): for offset in range(batch_count):
data = struct.unpack_from(frame_format, buf) data = struct.unpack_from(frame_format, buf)
buf = buf[struct.calcsize(frame_format):] buf = buf[struct.calcsize(frame_format):]
packets.append(Packet(sequence_number + offset, data)) yield Packet(sequence_number + offset, data)
return packets
class Timer: class Timer:
@ -129,7 +125,7 @@ def main():
while True: while True:
# Receive any data over UDP and parse it. # Receive any data over UDP and parse it.
data = connection.recv(1024) data = connection.recv(4096)
if data and not timer.is_started(): if data and not timer.is_started():
timer.start() timer.start()

View File

@ -194,7 +194,10 @@ const APP: () = {
stabilizer.net.mac_address, stabilizer.net.mac_address,
); );
let generator = network.enable_streaming(StreamFormat::AdcDacData); let generator = network.configure_streaming(
StreamFormat::AdcDacData,
SAMPLE_BUFFER_SIZE as u8,
);
// Spawn a settings update for default settings. // Spawn a settings update for default settings.
c.spawn.settings_update().unwrap(); c.spawn.settings_update().unwrap();

View File

@ -231,7 +231,10 @@ const APP: () = {
stabilizer.net.mac_address, stabilizer.net.mac_address,
); );
let generator = network.enable_streaming(StreamFormat::AdcDacData); let generator = network.configure_streaming(
StreamFormat::AdcDacData,
SAMPLE_BUFFER_SIZE as u8,
);
let settings = Settings::default(); let settings = Settings::default();

View File

@ -9,41 +9,34 @@
//! header is constant for all streaming capabilities, but the serialization format after the header //! header is constant for all streaming capabilities, but the serialization format after the header
//! is application-defined. //! is application-defined.
//! //!
//! ## Header Format //! ## Frame Header
//! The header of each stream frame consists of 7 bytes. All data is stored in little-endian format. //! The header consists of the following, all in little-endian.
//! //!
//! Elements appear sequentially as follows: //! * **Magic word 0x057B** <u16>: a constant to identify Stabilizer streaming data.
//! * Magic word 0x057B <u16> //! * **Format Code** <u8>: a unique ID that indicates the serialization format of each batch of data
//! * Format Code <u8> //! in the frame. Refer to [StreamFormat] for further information.
//! * Batch Size <u8> //! * **Batch Size** <u8>: the number of samples in each batch of data.
//! * Sequence Number <u32> //! * **Sequence Number** <u32>: an the sequence number of the first batch in the frame.
//! //! This can be used to determine if and how many stream batches are lost.
//! The "Magic word" is a constant field for all packets. The value is alway 0x057B.
//!
//! The "Format Code" is a unique specifier that indicates the serialization format of each batch of
//! data in the frame. Refer to [StreamFormat] for further information.
//!
//! The "Batch size" is the value of [SAMPLE_BUFFER_SIZE].
//!
//! The "Sequence Number" is an identifier that increments for ever execution of the DSP process.
//! This can be used to determine if a stream frame was lost.
//! //!
//! # Example //! # Example
//! A sample Python script is available in `scripts/stream_throughput.py` to demonstrate reception //! A sample Python script is available in `scripts/stream_throughput.py` to demonstrate reception
//! of livestreamed data. //! of livestreamed data.
use heapless::spsc::{Consumer, Producer, Queue}; use heapless::spsc::{Consumer, Producer, Queue};
use miniconf::MiniconfAtomic; use miniconf::MiniconfAtomic;
use num_enum::IntoPrimitive;
use serde::Deserialize; use serde::Deserialize;
use smoltcp_nal::embedded_nal::{IpAddr, Ipv4Addr, SocketAddr, UdpClientStack}; use smoltcp_nal::embedded_nal::{IpAddr, Ipv4Addr, SocketAddr, UdpClientStack};
use crate::hardware::design_parameters::SAMPLE_BUFFER_SIZE;
use heapless::pool::{Box, Init, Pool, Uninit}; use heapless::pool::{Box, Init, Pool, Uninit};
use super::NetworkReference; use super::NetworkReference;
const MAGIC_WORD: u16 = 0x057B; const MAGIC_WORD: u16 = 0x057B;
// The size of the header, calculated in bytes.
// The header has a 16-bit magic word, an 8-bit format, 8-bit batch-size, and 32-bit sequence
// number, which corresponds to 8 bytes total.
const HEADER_SIZE: usize = 8; const HEADER_SIZE: usize = 8;
// The number of frames that can be buffered. // The number of frames that can be buffered.
@ -52,6 +45,10 @@ const FRAME_COUNT: usize = 4;
// The size of each livestream frame in bytes. // The size of each livestream frame in bytes.
const FRAME_SIZE: usize = 1024 + HEADER_SIZE; const FRAME_SIZE: usize = 1024 + HEADER_SIZE;
// The size of the frame queue must be at least as large as the number of frame buffers. Every
// allocated frame buffer should fit in the queue.
const FRAME_QUEUE_SIZE: usize = FRAME_COUNT * 2;
// Static storage used for a heapless::Pool of frame buffers. // Static storage used for a heapless::Pool of frame buffers.
static mut FRAME_DATA: [u8; FRAME_SIZE * FRAME_COUNT] = static mut FRAME_DATA: [u8; FRAME_SIZE * FRAME_COUNT] =
[0; FRAME_SIZE * FRAME_COUNT]; [0; FRAME_SIZE * FRAME_COUNT];
@ -74,7 +71,7 @@ pub struct StreamTarget {
/// Specifies the format of streamed data /// Specifies the format of streamed data
#[repr(u8)] #[repr(u8)]
#[derive(Debug, Copy, Clone, PartialEq)] #[derive(Debug, Copy, Clone, PartialEq, IntoPrimitive)]
pub enum StreamFormat { pub enum StreamFormat {
/// Reserved, unused format specifier. /// Reserved, unused format specifier.
Unknown = 0, Unknown = 0,
@ -89,12 +86,6 @@ pub enum StreamFormat {
AdcDacData = 1, AdcDacData = 1,
} }
impl From<StreamFormat> for u8 {
fn from(format: StreamFormat) -> u8 {
format as u8
}
}
impl From<StreamTarget> for SocketAddr { impl From<StreamTarget> for SocketAddr {
fn from(target: StreamTarget) -> SocketAddr { fn from(target: StreamTarget) -> SocketAddr {
SocketAddr::new( SocketAddr::new(
@ -120,8 +111,10 @@ impl From<StreamTarget> for SocketAddr {
pub fn setup_streaming( pub fn setup_streaming(
stack: NetworkReference, stack: NetworkReference,
) -> (FrameGenerator, DataStream) { ) -> (FrameGenerator, DataStream) {
// The queue needs to be at least as large as the frame count to ensure that every allocated
// frame can potentially be enqueued for transmission.
let queue = let queue =
cortex_m::singleton!(: Queue<StreamFrame, FRAME_COUNT> = Queue::new()) cortex_m::singleton!(: Queue<StreamFrame, FRAME_QUEUE_SIZE> = Queue::new())
.unwrap(); .unwrap();
let (producer, consumer) = queue.split(); let (producer, consumer) = queue.split();
@ -167,17 +160,13 @@ impl StreamFrame {
where where
F: FnMut(&mut [u8]), F: FnMut(&mut [u8]),
{ {
assert!(!self.is_full::<T>(), "Batch cannot be added to full frame"); f(&mut self.buffer[self.offset..self.offset + T]);
let result = f(&mut self.buffer[self.offset..self.offset + T]);
self.offset += T; self.offset += T;
result
} }
pub fn is_full<const T: usize>(&self) -> bool { pub fn is_full<const T: usize>(&self) -> bool {
self.offset + T >= self.buffer.len() self.offset + T > self.buffer.len()
} }
pub fn finish(&mut self) -> &[u8] { pub fn finish(&mut self) -> &[u8] {
@ -187,37 +176,42 @@ impl StreamFrame {
/// The data generator for a stream. /// The data generator for a stream.
pub struct FrameGenerator { pub struct FrameGenerator {
queue: Producer<'static, StreamFrame, FRAME_COUNT>, queue: Producer<'static, StreamFrame, FRAME_QUEUE_SIZE>,
pool: &'static Pool<[u8; FRAME_SIZE]>, pool: &'static Pool<[u8; FRAME_SIZE]>,
current_frame: Option<StreamFrame>, current_frame: Option<StreamFrame>,
sequence_number: u32, sequence_number: u32,
format: u8, format: u8,
batch_size: u8,
} }
impl FrameGenerator { impl FrameGenerator {
fn new( fn new(
queue: Producer<'static, StreamFrame, FRAME_COUNT>, queue: Producer<'static, StreamFrame, FRAME_QUEUE_SIZE>,
pool: &'static Pool<[u8; FRAME_SIZE]>, pool: &'static Pool<[u8; FRAME_SIZE]>,
) -> Self { ) -> Self {
Self { Self {
queue, queue,
pool, pool,
batch_size: 0,
format: StreamFormat::Unknown.into(), format: StreamFormat::Unknown.into(),
current_frame: None, current_frame: None,
sequence_number: 0, sequence_number: 0,
} }
} }
/// Specify the format of the stream. /// Configure the format of the stream.
/// ///
/// # Note: /// # Note:
/// This function may only be called once upon initializing streaming /// This function shall only be called once upon initializing streaming
/// ///
/// # Args /// # Args
/// * `format` - The desired format of the stream. /// * `format` - The desired format of the stream.
/// * `batch_size` - The number of samples in each data batch. See
/// [crate::hardware::design_parameters::SAMPLE_BUFFER_SIZE]
#[doc(hidden)] #[doc(hidden)]
pub(crate) fn set_format(&mut self, format: impl Into<u8>) { pub(crate) fn configure(&mut self, format: impl Into<u8>, batch_size: u8) {
self.format = format.into(); self.format = format.into();
self.batch_size = batch_size;
} }
/// Add a batch to the current stream frame. /// Add a batch to the current stream frame.
@ -237,7 +231,7 @@ impl FrameGenerator {
self.current_frame.replace(StreamFrame::new( self.current_frame.replace(StreamFrame::new(
buffer, buffer,
self.format as u8, self.format as u8,
SAMPLE_BUFFER_SIZE as u8, self.batch_size,
sequence_number, sequence_number,
)); ));
} else { } else {
@ -245,11 +239,14 @@ impl FrameGenerator {
} }
} }
// Note(unwrap): We ensure the frame is present above.
let current_frame = self.current_frame.as_mut().unwrap(); let current_frame = self.current_frame.as_mut().unwrap();
current_frame.add_batch::<_, T>(f); current_frame.add_batch::<_, T>(f);
if current_frame.is_full::<T>() { if current_frame.is_full::<T>() {
// Note(unwrap): The queue is designed to be at least as large as the frame buffer
// count, so this enqueue should always succeed.
self.queue self.queue
.enqueue(self.current_frame.take().unwrap()) .enqueue(self.current_frame.take().unwrap())
.unwrap(); .unwrap();
@ -264,7 +261,7 @@ impl FrameGenerator {
pub struct DataStream { pub struct DataStream {
stack: NetworkReference, stack: NetworkReference,
socket: Option<<NetworkReference as UdpClientStack>::UdpSocket>, socket: Option<<NetworkReference as UdpClientStack>::UdpSocket>,
queue: Consumer<'static, StreamFrame, FRAME_COUNT>, queue: Consumer<'static, StreamFrame, FRAME_QUEUE_SIZE>,
frame_pool: &'static Pool<[u8; FRAME_SIZE]>, frame_pool: &'static Pool<[u8; FRAME_SIZE]>,
remote: SocketAddr, remote: SocketAddr,
} }
@ -278,7 +275,7 @@ impl DataStream {
/// * `frame_pool` - The Pool to return stream frame objects into. /// * `frame_pool` - The Pool to return stream frame objects into.
fn new( fn new(
stack: NetworkReference, stack: NetworkReference,
consumer: Consumer<'static, StreamFrame, FRAME_COUNT>, consumer: Consumer<'static, StreamFrame, FRAME_QUEUE_SIZE>,
frame_pool: &'static Pool<[u8; FRAME_SIZE]>, frame_pool: &'static Pool<[u8; FRAME_SIZE]>,
) -> Self { ) -> Self {
Self { Self {

View File

@ -116,12 +116,13 @@ where
/// ///
/// # Args /// # Args
/// * `format` - A unique u8 code indicating the format of the data. /// * `format` - A unique u8 code indicating the format of the data.
pub fn enable_streaming( pub fn configure_streaming(
&mut self, &mut self,
format: impl Into<u8>, format: impl Into<u8>,
batch_size: u8,
) -> FrameGenerator { ) -> FrameGenerator {
let mut generator = self.generator.take().unwrap(); let mut generator = self.generator.take().unwrap();
generator.set_format(format); generator.configure(format, batch_size);
generator generator
} }