opentitanlib/test_utils/bitbanging/
spi.rs

1// Copyright lowRISC contributors (OpenTitan project).
2// Licensed under the Apache License, Version 2.0, see LICENSE for details.
3// SPDX-License-Identifier: Apache-2.0
4
5use super::Bit;
6use serde::{Deserialize, Serialize};
7use std::iter::Peekable;
8use thiserror::Error;
9
10#[derive(Clone, Copy, Debug, PartialEq)]
11pub enum SpiEndpoint {
12    Host,   // i.e. "Controller"
13    Device, // i.e. "Peripheral"
14}
15
16/// The SPI data mode, indicating how many data lines to use for transmission.
17#[derive(Clone, Copy, Debug, PartialEq)]
18pub enum SpiDataMode {
19    Single,
20    Dual,
21    Quad,
22}
23
24/// Configuration for SPI bitbanging
25#[derive(Clone, Debug)]
26pub struct SpiBitbangConfig {
27    pub cpol: bool, // Clock polarity
28    pub cpha: bool, // Clock Phase
29    pub data_mode: SpiDataMode,
30    pub bits_per_word: u32,
31    // TODO: add DDR (Dual Data Rate) support
32}
33
34/// Additional delays required to synchronise with a SPI device, all
35/// measured in SPI clock cycles (2 bitbanging samples per cycle).
36#[derive(Clone, Debug)]
37pub struct SpiEncodingDelays {
38    pub inter_word_delay: u32,
39    pub cs_hold_delay: u32,
40    pub cs_release_delay: u32,
41}
42
43#[derive(Error, Debug, PartialEq, Serialize, Deserialize)]
44pub enum SpiTransferEncodeError {
45    #[error("CS must be asserted before bitbanging a SPI transaction.")]
46    CsNotAsserted,
47}
48
49/// An encoder for SPI transmissions, parameterized over bits in the output
50/// bitfields to use for transmission. Does not support encoding as a SPI
51/// device (only works as a SPI host).
52pub struct SpiBitbangEncoder<
53    const D0: u8,
54    const D1: u8,
55    const D2: u8,
56    const D3: u8,
57    const CLK: u8,
58    const CS: u8,
59> {
60    pub config: SpiBitbangConfig,
61    pub delays: SpiEncodingDelays,
62    first_word: bool,
63    cs_asserted: bool,
64}
65
66// Bits for unused pins should not be used (high-impedance), so default to low.
67const UNUSED: Bit = Bit::Low;
68const CS_LOW: Bit = Bit::Low;
69const CS_HIGH: Bit = Bit::High;
70
71impl<const D0: u8, const D1: u8, const D2: u8, const D3: u8, const CLK: u8, const CS: u8>
72    SpiBitbangEncoder<D0, D1, D2, D3, CLK, CS>
73{
74    pub fn new(config: SpiBitbangConfig, delays: SpiEncodingDelays) -> Self {
75        Self {
76            config,
77            delays,
78            first_word: true,
79            cs_asserted: false,
80        }
81    }
82
83    // Reset the state of the SPI bitbanging encoder.
84    pub fn reset(&mut self) {
85        self.first_word = true;
86        self.cs_asserted = false;
87    }
88
89    // A helper to change the data mode that is used for encoding.
90    pub fn set_data_mode(&mut self, mode: SpiDataMode) {
91        self.config.data_mode = mode;
92    }
93
94    /// Construct a sample bitmap for a set of values on SPI pins
95    fn sample(&self, d0: Bit, d1: Bit, d2: Bit, d3: Bit, clk: Bit, cs: Bit) -> u8 {
96        ((d0 as u8) << D0)
97            | ((d1 as u8) << D1)
98            | ((d2 as u8) << D2)
99            | ((d3 as u8) << D3)
100            | ((clk as u8) << CLK)
101            | ((cs as u8) << CS)
102    }
103
104    /// Encode up to 4 data bits into 2 bitbanging samples corresponding to 1 SPI clock
105    fn encode_data(&self, d0: Bit, d1: Bit, d2: Bit, d3: Bit, samples: &mut Vec<u8>) {
106        let clk_idle = Bit::from(self.config.cpol);
107        let clk_active = Bit::from(!self.config.cpol);
108        samples.extend(if self.config.cpha {
109            // CPHA=1, so output bits on (idle->active) edges
110            [
111                self.sample(d0, d1, d2, d3, clk_active, CS_LOW),
112                self.sample(d0, d1, d2, d3, clk_idle, CS_LOW),
113            ]
114        } else {
115            // CPHA=0, so output bits on (active->idle) edges
116            [
117                self.sample(d0, d1, d2, d3, clk_idle, CS_LOW),
118                self.sample(d0, d1, d2, d3, clk_active, CS_LOW),
119            ]
120        })
121    }
122
123    /// Encode 1 SPI word into bitbanging samples. This does not handle
124    /// additional logic for wait times and CPHA that must be considered
125    /// with the first word.
126    fn encode_word(
127        &mut self,
128        words: &[u8],
129        samples: &mut Vec<u8>,
130    ) -> Result<(), SpiTransferEncodeError> {
131        if !self.cs_asserted {
132            return Err(SpiTransferEncodeError::CsNotAsserted);
133        }
134        if self.first_word {
135            self.first_word = false;
136        }
137        let mut byte = 0x00u8;
138        let mut encoded_bits = 0u32; // Count of bits in the current word
139        let mut words = words.iter();
140        while encoded_bits < self.config.bits_per_word {
141            let bits = encoded_bits % 8;
142            if bits == 0 {
143                if let Some(&next_byte) = words.next() {
144                    byte = next_byte;
145                } else {
146                    break;
147                }
148            }
149            match self.config.data_mode {
150                SpiDataMode::Single => {
151                    let d0 = Bit::from((byte >> (7 - bits)) & 0x01);
152                    self.encode_data(d0, UNUSED, UNUSED, UNUSED, samples);
153                    encoded_bits += 1;
154                }
155                SpiDataMode::Dual => {
156                    let d1 = Bit::from((byte >> (7 - bits)) & 0x01);
157                    let d0 = Bit::from((byte >> (7 - (bits + 1))) & 0x01);
158                    self.encode_data(d0, d1, UNUSED, UNUSED, samples);
159                    encoded_bits += 2;
160                }
161                SpiDataMode::Quad => {
162                    let d3 = Bit::from((byte >> (7 - bits)) & 0x01);
163                    let d2 = Bit::from((byte >> (7 - (bits + 1))) & 0x01);
164                    let d1 = Bit::from((byte >> (7 - (bits + 2))) & 0x01);
165                    let d0 = Bit::from((byte >> (7 - (bits + 3))) & 0x01);
166                    self.encode_data(d0, d1, d2, d3, samples);
167                    encoded_bits += 4;
168                }
169            }
170        }
171
172        // If not enough data is given, pad with 0s until it fits.
173        while encoded_bits != 0 && encoded_bits < self.config.bits_per_word {
174            match self.config.data_mode {
175                SpiDataMode::Single => {
176                    self.encode_data(Bit::Low, UNUSED, UNUSED, UNUSED, samples);
177                    encoded_bits += 1;
178                }
179                SpiDataMode::Dual => {
180                    self.encode_data(Bit::Low, Bit::Low, UNUSED, UNUSED, samples);
181                    encoded_bits += 2;
182                }
183                SpiDataMode::Quad => {
184                    self.encode_data(Bit::Low, Bit::Low, Bit::Low, Bit::Low, samples);
185                    encoded_bits += 4;
186                }
187            }
188        }
189        Ok(())
190    }
191
192    /// Encode a sequence of SPI words into GPIO bitbanging samples.
193    fn encode_words(
194        &mut self,
195        words: &[u8],
196        samples: &mut Vec<u8>,
197    ) -> Result<(), SpiTransferEncodeError> {
198        // Keep encoding words in the data while more exist
199        let clk_idle = Bit::from(self.config.cpol);
200        let bytes_per_word = self.config.bits_per_word.div_ceil(8) as usize;
201        for word in words.chunks(bytes_per_word) {
202            if !self.first_word {
203                // Optional delays between words
204                for _ in 0..(self.delays.inter_word_delay * 2) {
205                    samples.push(self.sample(UNUSED, UNUSED, UNUSED, UNUSED, clk_idle, CS_LOW))
206                }
207            }
208            self.encode_word(word, samples)?;
209        }
210        Ok(())
211    }
212
213    /// Output GPIO bitbanging samples for pulling CS low/high (active/inactive).
214    pub fn assert_cs(&mut self, assert: bool, samples: &mut Vec<u8>) {
215        if (assert && self.cs_asserted) || (!assert && !self.cs_asserted) {
216            return;
217        }
218        self.cs_asserted = assert;
219        let clk_idle = Bit::from(self.config.cpol);
220        let cs_low = self.sample(UNUSED, UNUSED, UNUSED, UNUSED, clk_idle, CS_LOW);
221        if assert {
222            // If CPHA=0, we omit the initial CS low sample here as that is
223            // combined with the first data write.
224            let wait_samples = match self.config.cpha {
225                true => self.delays.cs_hold_delay * 2 + 1,
226                false => self.delays.cs_hold_delay * 2,
227            };
228            for _ in 0..wait_samples {
229                samples.push(cs_low);
230            }
231            self.first_word = true;
232        } else {
233            // If CPHA=0, we must do a final transition of CLK back to idle before
234            // we de-assert the CS (not accounting for release delays).
235            let wait_samples = match self.config.cpha {
236                true => self.delays.cs_release_delay * 2,
237                false => self.delays.cs_release_delay * 2 + 1,
238            };
239            for _ in 0..wait_samples {
240                samples.push(cs_low);
241            }
242            samples.push(self.sample(UNUSED, UNUSED, UNUSED, UNUSED, clk_idle, CS_HIGH));
243        }
244    }
245
246    /// Encode a read transmission of several SPI words into GPIO bitbanging samples.
247    /// CS should already be asserted via `assert_cs` before calling.
248    pub fn encode_read(
249        &mut self,
250        words: usize,
251        samples: &mut Vec<u8>,
252    ) -> Result<(), SpiTransferEncodeError> {
253        self.encode_words(&vec![0; words], samples)
254    }
255
256    /// Encode a write transmission of several SPI words into GPIO bitbanging samples.
257    /// CS should already be asserted via `assert_cs` before calling.
258    pub fn encode_write(
259        &mut self,
260        data: &[u8],
261        samples: &mut Vec<u8>,
262    ) -> Result<(), SpiTransferEncodeError> {
263        self.encode_words(data, samples)
264    }
265
266    /// A helper function to encode a full SPI transmission, including logic for
267    /// asserting and later de-asserting the CS.
268    pub fn encode_transaction(
269        &mut self,
270        data: &[u8],
271        samples: &mut Vec<u8>,
272    ) -> Result<(), SpiTransferEncodeError> {
273        self.assert_cs(true, samples);
274        self.encode_words(data, samples)?;
275        self.assert_cs(false, samples);
276        Ok(())
277    }
278}
279
280/// A sample of SPI pins at a given instant. The const generics should all be
281/// different bit indexes to refer to different pins.
282#[derive(Clone, Debug)]
283struct Sample<const D0: u8, const D1: u8, const D2: u8, const D3: u8, const CLK: u8, const CS: u8> {
284    raw: u8,
285}
286
287impl<const D0: u8, const D1: u8, const D2: u8, const D3: u8, const CLK: u8, const CS: u8>
288    Sample<D0, D1, D2, D3, CLK, CS>
289{
290    fn d0(&self) -> Bit {
291        ((self.raw >> D0) & 0x01).into()
292    }
293    fn d1(&self) -> Bit {
294        ((self.raw >> D1) & 0x01).into()
295    }
296    fn d2(&self) -> Bit {
297        ((self.raw >> D2) & 0x01).into()
298    }
299    fn d3(&self) -> Bit {
300        ((self.raw >> D3) & 0x01).into()
301    }
302    fn clk(&self) -> Bit {
303        ((self.raw >> CLK) & 0x01).into()
304    }
305    fn cs(&self) -> Bit {
306        ((self.raw >> CS) & 0x01).into()
307    }
308}
309
310#[derive(Error, Debug, PartialEq, Serialize, Deserialize)]
311pub enum SpiTransferDecodeError {
312    #[error("Settings mismatch: Clock level when idle is {0:?}, but cpol expects {1:?}")]
313    ClockPolarityMismatch(Bit, Bit),
314    #[error("Chip select was de-asserted while a SPI transaction was in progress")]
315    ChipSelectDeassertedEarly,
316    #[error("Not enough samples were given to complete the SPI transaction")]
317    UnfinishedTransaction,
318}
319
320/// A decoder for SPI transmissions, parameterized over bits in input sample
321/// bitfields of SPI transmissions.
322pub struct SpiBitbangDecoder<
323    const D0: u8,
324    const D1: u8,
325    const D2: u8,
326    const D3: u8,
327    const CLK: u8,
328    const CS: u8,
329> {
330    pub config: SpiBitbangConfig,
331    pub endpoint: SpiEndpoint,
332}
333
334impl<const D0: u8, const D1: u8, const D2: u8, const D3: u8, const CLK: u8, const CS: u8>
335    SpiBitbangDecoder<D0, D1, D2, D3, CLK, CS>
336{
337    pub fn new(config: SpiBitbangConfig, endpoint: SpiEndpoint) -> Self {
338        Self { config, endpoint }
339    }
340
341    pub fn set_data_mode(&mut self, mode: SpiDataMode) {
342        self.config.data_mode = mode;
343    }
344
345    /// Iterate through samples until a low (active) CS level is found. Then, check
346    /// the clock level based on CPOL config. Returns `true` if a low CS was found.
347    fn wait_cs<I>(&self, samples: &mut Peekable<I>) -> Result<bool, SpiTransferDecodeError>
348    where
349        I: Iterator<Item = Sample<D0, D1, D2, D3, CLK, CS>>,
350    {
351        let samples = samples.by_ref();
352        let clk_idle_level = Bit::from(self.config.cpol);
353        while let Some(sample) = samples.peek() {
354            if sample.cs() != Bit::Low {
355                samples.next();
356                continue;
357            }
358            return if sample.clk() == clk_idle_level {
359                Ok(true)
360            } else {
361                Err(SpiTransferDecodeError::ClockPolarityMismatch(
362                    sample.clk(),
363                    clk_idle_level,
364                ))
365            };
366        }
367        Ok(false)
368    }
369
370    /// Get the sample corresponding to the next data bit, directly after an edge
371    /// that depends on CPOL/CPHA configuration. Set `first_edge=true` to indicate
372    /// that this is the first edge sampled of this SPI word transmission.
373    fn sample_on_edge<I>(
374        &self,
375        samples: &mut I,
376        first_edge: bool,
377    ) -> Result<Option<Sample<D0, D1, D2, D3, CLK, CS>>, SpiTransferDecodeError>
378    where
379        I: Iterator<Item = Sample<D0, D1, D2, D3, CLK, CS>>,
380    {
381        let (wait_level, sample_level) = if self.config.cpol == self.config.cpha {
382            (Bit::Low, Bit::High)
383        } else {
384            (Bit::High, Bit::Low)
385        };
386        let mut last_sample = None;
387        for level in [wait_level, sample_level] {
388            let Some(sample) = samples
389                .by_ref()
390                .find(|sample| sample.clk() == level || sample.cs() == Bit::High)
391            else {
392                if !first_edge {
393                    return Err(SpiTransferDecodeError::UnfinishedTransaction);
394                }
395                return Ok(None);
396            };
397            if sample.cs() == Bit::High {
398                if !first_edge {
399                    return Err(SpiTransferDecodeError::ChipSelectDeassertedEarly);
400                }
401                return Ok(None);
402            }
403            last_sample = Some(sample);
404        }
405        Ok(last_sample)
406    }
407
408    /// Decode a SPI word from some input GPIO samples. Returns an error if CS
409    /// is deasserted early or the samples are unfinished.
410    fn decode_word<I>(&self, samples: &mut I) -> Result<Vec<u8>, SpiTransferDecodeError>
411    where
412        I: Iterator<Item = Sample<D0, D1, D2, D3, CLK, CS>>,
413    {
414        let bytes_per_word = self.config.bits_per_word.div_ceil(8) as usize;
415        let mut byte: u8 = 0x00;
416        let mut decoded_bits = 0u32;
417        let mut word: Vec<u8> = Vec::with_capacity(bytes_per_word);
418        while decoded_bits < self.config.bits_per_word {
419            let Some(sample) = self.sample_on_edge(samples, decoded_bits == 0)? else {
420                break;
421            };
422            match self.config.data_mode {
423                SpiDataMode::Single => {
424                    byte <<= 1;
425                    // In single reads, devices read from COPI whilst hosts read from CIPO.
426                    byte |= match self.endpoint {
427                        SpiEndpoint::Device => sample.d0() as u8,
428                        SpiEndpoint::Host => sample.d1() as u8,
429                    };
430                    decoded_bits += 1;
431                }
432                SpiDataMode::Dual => {
433                    byte <<= 1;
434                    byte |= sample.d1() as u8;
435                    byte <<= 1;
436                    byte |= sample.d0() as u8;
437                    decoded_bits += 2;
438                }
439                SpiDataMode::Quad => {
440                    byte <<= 1;
441                    byte |= sample.d3() as u8;
442                    byte <<= 1;
443                    byte |= sample.d2() as u8;
444                    byte <<= 1;
445                    byte |= sample.d1() as u8;
446                    byte <<= 1;
447                    byte |= sample.d0() as u8;
448                    decoded_bits += 4;
449                }
450            }
451            if decoded_bits.is_multiple_of(8) {
452                word.push(byte);
453                byte = 0x00;
454            }
455        }
456        if !decoded_bits.is_multiple_of(8) {
457            // For < 8 bits per word, we shift partial data back into the MSBs
458            byte <<= 8 - (decoded_bits % 8);
459            word.push(byte);
460        }
461        Ok(word)
462    }
463
464    /// Decode a full SPI transmission from input GPIO samples, which may contain many
465    /// SPI words. Expects CS to be deasserted by the end of all transactions.
466    /// Returns the SPI words as a vector of bytes, using the LSBs for partial bytes
467    /// (e.g. for 12-bit words, the mask is [0xFF, 0X0F, ...]).
468    pub fn run(&self, samples: Vec<u8>) -> Result<Vec<u8>, SpiTransferDecodeError> {
469        let mut samples = samples
470            .into_iter()
471            .map(|raw| Sample::<D0, D1, D2, D3, CLK, CS> { raw })
472            .peekable();
473        let mut bytes = Vec::new();
474        if !self.wait_cs(&mut samples)? {
475            return Ok(bytes);
476        }
477        loop {
478            let word = self.decode_word(&mut samples)?;
479            if word.is_empty() && !self.wait_cs(&mut samples)? {
480                break;
481            }
482            bytes.extend(word);
483        }
484        Ok(bytes)
485    }
486}
487
488#[cfg(test)]
489mod test {
490    use super::*;
491    use anyhow::Result;
492    use std::cmp::Ordering;
493
494    fn spi_encode_decode(
495        config: SpiBitbangConfig,
496        delays: SpiEncodingDelays,
497        data: Option<&[u8]>,
498    ) -> Result<()> {
499        let bytes_per_word = config.bits_per_word.div_ceil(8) as usize;
500        let mut encoder = SpiBitbangEncoder::<0, 1, 2, 3, 4, 5>::new(config.clone(), delays);
501        let decoder =
502            SpiBitbangDecoder::<0, 1, 2, 3, 4, 5>::new(config.clone(), SpiEndpoint::Device);
503        let mut data = Vec::from(data.unwrap_or(b"Hello, this is a simple SPI test message."));
504        while data.len() % bytes_per_word != 0 {
505            data.push(0);
506        }
507        let mut samples = Vec::new();
508        encoder.encode_transaction(&data, &mut samples)?;
509        assert!(!samples.is_empty());
510        let decoded = decoder
511            .run(samples)
512            .expect("Should have decoded the bitbanged message");
513        assert_eq!(decoded, data);
514        Ok(())
515    }
516
517    #[test]
518    fn smoke() -> Result<()> {
519        // Encode and decode some test strings
520        let config = SpiBitbangConfig {
521            cpol: false,
522            cpha: false,
523            data_mode: SpiDataMode::Single,
524            bits_per_word: 8,
525        };
526        let delays = SpiEncodingDelays {
527            inter_word_delay: 0,
528            cs_hold_delay: 1,
529            cs_release_delay: 1,
530        };
531        spi_encode_decode(config.clone(), delays.clone(), None)?;
532        spi_encode_decode(config.clone(), delays.clone(), Some(b"abc def GHI JKL"))?;
533        spi_encode_decode(config.clone(), delays.clone(), Some(b"12345678"))?;
534
535        // Check bitbang encoding against a known sample.
536        let mut encoder = SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(config, delays);
537        let bytes = [0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF];
538        let mut samples = Vec::new();
539        encoder.encode_transaction(&bytes, &mut samples)?;
540        let expected = [
541            0, 0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 0, 1, 0, 1, 0,
542            1, 4, 5, 4, 5, 0, 1, 4, 5, 0, 1, 0, 1, 0, 1, 4, 5, 0, 1, 4, 5, 0, 1, 4, 5, 4, 5, 0, 1,
543            0, 1, 4, 5, 4, 5, 4, 5, 4, 5, 0, 1, 0, 1, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 4, 5, 0, 1, 4,
544            5, 0, 1, 4, 5, 0, 1, 4, 5, 4, 5, 4, 5, 4, 5, 0, 1, 0, 1, 4, 5, 4, 5, 0, 1, 4, 5, 4, 5,
545            4, 5, 4, 5, 0, 1, 4, 5, 4, 5, 4, 5, 4, 5, 0, 0, 0, 2,
546        ];
547        assert_eq!(samples, expected);
548        Ok(())
549    }
550
551    #[test]
552    fn communication_modes() -> Result<()> {
553        // Test encoding/decoding all possible modes both with and without
554        // additional CS delays on either end, and between words.
555        for config_bitmap in 0..32 {
556            let config = SpiBitbangConfig {
557                cpol: config_bitmap & 0x01 > 0,
558                cpha: (config_bitmap >> 1) & 0x01 > 0,
559                data_mode: SpiDataMode::Single,
560                bits_per_word: 8,
561            };
562            let delays = SpiEncodingDelays {
563                inter_word_delay: (config_bitmap >> 2) & 0x01,
564                cs_hold_delay: (config_bitmap >> 3) & 0x01,
565                cs_release_delay: (config_bitmap >> 4) & 0x01,
566            };
567            spi_encode_decode(config.clone(), delays.clone(), None)?;
568            spi_encode_decode(config, delays, Some(b"communication mode test message"))?;
569        }
570
571        // Use small transfers to test that the output matches what we expect.
572        let tests = [
573            // CPOL=0,CPHA=0 so bit #1 is output with CS low on idle (low) clock in 1st sample
574            ((false, false), vec![4, 5, 0, 1]),
575            // CPOL=0,CPHA=1 so bit #1 is output with active (high) clock in 2nd sample
576            ((false, true), vec![0, 5, 4, 1]),
577            // CPOL=1,CPHA=0 so bit #1 is output with CS low on idle (high) clock in 1st sample
578            ((true, false), vec![5, 4, 1, 0]),
579            // CPOL=1,CPHA=1 so bit #1 is output with active (low) clock in 2nd sample
580            ((true, true), vec![1, 4, 5, 0]),
581        ];
582        let delays = SpiEncodingDelays {
583            inter_word_delay: 0,
584            cs_hold_delay: 0,
585            cs_release_delay: 0,
586        };
587        for ((cpol, cpha), expected) in tests {
588            let config = SpiBitbangConfig {
589                cpol,
590                cpha,
591                data_mode: SpiDataMode::Single,
592                bits_per_word: 8,
593            };
594            let mut samples = Vec::new();
595            SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(config, delays.clone())
596                .encode_transaction(&[0xA5], &mut samples)?;
597            assert_eq!(samples[..4], expected);
598        }
599        Ok(())
600    }
601
602    #[test]
603    fn data_modes() -> Result<()> {
604        // Test encoding/decoding for each data channel mode.
605        let delays = SpiEncodingDelays {
606            inter_word_delay: 0,
607            cs_hold_delay: 1,
608            cs_release_delay: 1,
609        };
610        for data_mode in [SpiDataMode::Single, SpiDataMode::Dual, SpiDataMode::Quad] {
611            let config = SpiBitbangConfig {
612                cpol: false,
613                cpha: false,
614                data_mode,
615                bits_per_word: 8,
616            };
617            spi_encode_decode(config.clone(), delays.clone(), None)?;
618            spi_encode_decode(
619                config,
620                delays.clone(),
621                Some(b"A slightly longer message that can be used to test SPI data modes"),
622            )?;
623        }
624
625        // Check that the number of transfers is being appropriately decreased.
626        let delays = SpiEncodingDelays {
627            inter_word_delay: 0,
628            cs_hold_delay: 0,
629            cs_release_delay: 0,
630        };
631        let tests = [
632            // (64 bits * 2 half clks) + 1 half clk to return to idle + 1 sample for CS high
633            (SpiDataMode::Single, 64 * 2 + 2),
634            (SpiDataMode::Dual, (64 / 2) * 2 + 2),
635            (SpiDataMode::Quad, (64 / 4) * 2 + 2),
636        ];
637        for (data_mode, expected_half_clocks) in tests {
638            let config = SpiBitbangConfig {
639                cpol: false,
640                cpha: false,
641                data_mode,
642                bits_per_word: 8,
643            };
644            let mut samples = Vec::new();
645            SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(config, delays.clone()).encode_transaction(
646                &[0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF],
647                &mut samples,
648            )?;
649            assert_eq!(samples.len(), expected_half_clocks);
650        }
651        Ok(())
652    }
653
654    #[test]
655    fn bits_per_word() -> Result<()> {
656        // Test encoding/decoding for a variety of bits per SPI word.
657        let delays = SpiEncodingDelays {
658            inter_word_delay: 0,
659            cs_hold_delay: 1,
660            cs_release_delay: 1,
661        };
662        for bits_per_word in 1..=64 {
663            let config = SpiBitbangConfig {
664                cpol: false,
665                cpha: false,
666                data_mode: SpiDataMode::Single,
667                bits_per_word,
668            };
669            // Use a message with [0, 1, 2, ..., 63] bitwise reversed into the MSBs, but for
670            // each test we first mask the data so that the unused ending LSBs are zeroed for
671            // non-byte-divisible word sizes, to allow comparison of decoded results.
672            let mut bytes = (0..64).map(|b: u8| b.reverse_bits()).collect::<Vec<u8>>();
673            let mut bits_in_current_word = 0;
674            for byte in bytes.iter_mut() {
675                let mask_size = bits_per_word - bits_in_current_word;
676                match mask_size.cmp(&8) {
677                    Ordering::Greater => {
678                        bits_in_current_word += 8;
679                    }
680                    Ordering::Equal => {
681                        bits_in_current_word = 0;
682                    }
683                    Ordering::Less => {
684                        let zero_mask = (0x01 << (8 - mask_size)) - 1;
685                        *byte &= !zero_mask;
686                        bits_in_current_word = 0;
687                    }
688                }
689            }
690            spi_encode_decode(config, delays.clone(), Some(&bytes))?;
691        }
692        Ok(())
693    }
694
695    #[test]
696    fn encoding_delays() -> Result<()> {
697        // Test a variety of SPI encoding delays, with delays after setting CS low,
698        // before releasing CS, and between each SPI word transfer.
699        let tests = [
700            (
701                (0, 0, 0),
702                vec![4, 5, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 0, 1, 4, 5, 0, 2],
703            ),
704            (
705                (3, 0, 0),
706                vec![
707                    4, 5, 0, 1, 4, 5, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 4, 5, 0, 1, 4, 5, 0, 2,
708                ],
709            ),
710            (
711                (0, 3, 0),
712                vec![
713                    0, 0, 0, 0, 0, 0, 4, 5, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 0, 1, 4, 5, 0, 2,
714                ],
715            ),
716            (
717                (0, 0, 3),
718                vec![
719                    4, 5, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 0, 1, 4, 5, 0, 0, 0, 0, 0, 0, 0, 2,
720                ],
721            ),
722            (
723                (1, 2, 3),
724                vec![
725                    0, 0, 0, 0, 4, 5, 0, 1, 4, 5, 0, 1, 0, 0, 0, 1, 4, 5, 0, 1, 4, 5, 0, 0, 0, 0,
726                    0, 0, 0, 2,
727                ],
728            ),
729        ];
730        for ((inter_word_delay, cs_hold_delay, cs_release_delay), expected) in tests {
731            let delays = SpiEncodingDelays {
732                inter_word_delay,
733                cs_hold_delay,
734                cs_release_delay,
735            };
736            // Run an encode/decode test with these delays for each CPHA.
737            for cpha in [false, true] {
738                let config = SpiBitbangConfig {
739                    cpol: false,
740                    cpha,
741                    data_mode: SpiDataMode::Single,
742                    bits_per_word: 8,
743                };
744                spi_encode_decode(config, delays.clone(), None)?;
745            }
746            let config = SpiBitbangConfig {
747                cpol: false,
748                cpha: false,
749                data_mode: SpiDataMode::Single,
750                bits_per_word: 4,
751            };
752            let mut samples = Vec::new();
753            SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(config, delays)
754                .encode_transaction(&[0xA << 4, 0x5 << 4], &mut samples)?;
755            assert_eq!(samples, expected);
756        }
757        Ok(())
758    }
759
760    #[test]
761    fn decoding_endpoints() -> Result<()> {
762        // Test decoding as both a SPI Host and Device
763        let config = SpiBitbangConfig {
764            cpol: false,
765            cpha: false,
766            data_mode: SpiDataMode::Single,
767            bits_per_word: 8,
768        };
769        let delays = SpiEncodingDelays {
770            inter_word_delay: 0,
771            cs_hold_delay: 1,
772            cs_release_delay: 1,
773        };
774        let bytes = [0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF];
775        let mut samples = Vec::new();
776        // Encode on COPI = D0 = bit 0.
777        SpiBitbangEncoder::<0, 1, 2, 3, 4, 5>::new(config.clone(), delays.clone())
778            .encode_transaction(&bytes, &mut samples)?;
779        // Decode as a SPI Device on COPI = Bit 0
780        let decoder =
781            SpiBitbangDecoder::<0, 1, 2, 3, 4, 5>::new(config.clone(), SpiEndpoint::Device);
782        let mut decoded = decoder
783            .run(samples.clone())
784            .expect("Should have decoded the bitbanged message");
785        assert_eq!(decoded, &bytes);
786        // Decode as a SPI Host on CIPO = Bit 0
787        let decoder = SpiBitbangDecoder::<1, 0, 2, 3, 4, 5>::new(config.clone(), SpiEndpoint::Host);
788        decoded = decoder
789            .run(samples)
790            .expect("Should have decoded the bitbanged message");
791        assert_eq!(decoded, &bytes);
792        Ok(())
793    }
794
795    #[test]
796    fn encode_host_reads() -> Result<()> {
797        // Encode a SPI host read
798        let mut encoder = SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(
799            SpiBitbangConfig {
800                cpol: false,
801                cpha: false,
802                data_mode: SpiDataMode::Single,
803                bits_per_word: 8,
804            },
805            SpiEncodingDelays {
806                inter_word_delay: 0,
807                cs_hold_delay: 0,
808                cs_release_delay: 0,
809            },
810        );
811        let mut samples = Vec::new();
812        encoder.assert_cs(true, &mut samples);
813        encoder.encode_read(5, &mut samples)?;
814        encoder.assert_cs(false, &mut samples);
815        // Only expect to see clock changes with CS low, no data.
816        let mut expected = vec![0];
817        expected.append(&mut [1, 0].repeat(8 * 5)); // clock cycles
818        expected.push(2); // CS high (inactive)
819        assert_eq!(samples, expected);
820        Ok(())
821    }
822
823    #[test]
824    fn encode_cs_assertions() -> Result<()> {
825        // Test errors surrounding CS assertions
826        let mut encoder = SpiBitbangEncoder::<2, 3, 4, 5, 0, 1>::new(
827            SpiBitbangConfig {
828                cpol: false,
829                cpha: true,
830                data_mode: SpiDataMode::Single,
831                bits_per_word: 8,
832            },
833            SpiEncodingDelays {
834                inter_word_delay: 0,
835                cs_hold_delay: 0,
836                cs_release_delay: 0,
837            },
838        );
839        let mut samples = vec![];
840        // No additional samples if already de-asserted
841        encoder.assert_cs(false, &mut samples);
842        assert_eq!(samples.len(), 0);
843        // Do not allow reads/writes if de-asserted
844        assert_eq!(
845            encoder.encode_write(&[0], &mut samples),
846            Err(SpiTransferEncodeError::CsNotAsserted)
847        );
848        assert_eq!(
849            encoder.encode_read(1, &mut samples),
850            Err(SpiTransferEncodeError::CsNotAsserted)
851        );
852        assert_eq!(samples.len(), 0);
853        // Samples are encoded on a CS assertion
854        encoder.assert_cs(true, &mut samples);
855        let mut sample_len = samples.len();
856        assert!(sample_len > 0);
857        // No additional samples if already asserted
858        encoder.assert_cs(true, &mut samples);
859        assert_eq!(samples.len(), sample_len);
860        // Samples are encoded on a CS de-assertion
861        encoder.assert_cs(false, &mut samples);
862        assert!(samples.len() > sample_len);
863        sample_len = samples.len();
864        // No additional samples again if already de-asserted
865        encoder.assert_cs(false, &mut samples);
866        assert_eq!(samples.len(), sample_len);
867        Ok(())
868    }
869
870    #[test]
871    fn spi_decoding_errors() -> Result<()> {
872        // Test clock polarity mismatch errors
873        assert_eq!(
874            SpiBitbangDecoder::<2, 3, 4, 5, 0, 1>::new(
875                SpiBitbangConfig {
876                    cpol: false,
877                    cpha: false,
878                    data_mode: SpiDataMode::Single,
879                    bits_per_word: 2,
880                },
881                SpiEndpoint::Device
882            )
883            .run(vec![3, 1, 0, 1, 0, 1, 0, 2]),
884            Err(SpiTransferDecodeError::ClockPolarityMismatch(
885                Bit::High,
886                Bit::Low
887            ))
888        );
889        assert_eq!(
890            SpiBitbangDecoder::<2, 3, 4, 5, 0, 1>::new(
891                SpiBitbangConfig {
892                    cpol: true,
893                    cpha: false,
894                    data_mode: SpiDataMode::Single,
895                    bits_per_word: 2,
896                },
897                SpiEndpoint::Device
898            )
899            .run(vec![2, 0, 1, 0, 1, 0, 1, 3]),
900            Err(SpiTransferDecodeError::ClockPolarityMismatch(
901                Bit::Low,
902                Bit::High
903            ))
904        );
905        // Test for errors when de-asserting the CS early at any point
906        // before a word transfer is finished.
907        let valid_samples = vec![4, 5, 0, 1, 4, 5, 0, 1, 0, 1, 4, 5, 0, 1, 4, 5, 0, 2];
908        let decoder = SpiBitbangDecoder::<2, 3, 4, 5, 0, 1>::new(
909            SpiBitbangConfig {
910                cpol: false,
911                cpha: false,
912                data_mode: SpiDataMode::Single,
913                bits_per_word: 8,
914            },
915            SpiEndpoint::Device,
916        );
917        assert!(decoder.run(valid_samples.clone()).is_ok());
918        for index in 1..(valid_samples.len() - 2) {
919            let mut invalid_samples = valid_samples.clone();
920            invalid_samples[index] |= 0x01 << 1;
921            assert_eq!(
922                decoder.run(invalid_samples),
923                Err(SpiTransferDecodeError::ChipSelectDeassertedEarly)
924            );
925        }
926        // Test errors with unfinished transactions.
927        for index in 2..(valid_samples.len() - 2) {
928            let mut invalid_samples = valid_samples.clone();
929            invalid_samples.truncate(index);
930            assert_eq!(
931                decoder.run(invalid_samples),
932                Err(SpiTransferDecodeError::UnfinishedTransaction)
933            )
934        }
935        Ok(())
936    }
937}
opentitanlib/test_utils/bitbanging/spi.rs

opentitanlib/test_utils/bitbanging/
spi.rs
