use std::cell::Cell;
use std::time::Duration;
use hidapi::{HidApi, HidDevice};
use crate::Error;
use crate::analog::{AdcReading, VoltageReference};
use crate::commands::{FlashDataSubCode, McpCommand, UsbReport};
use crate::constants::{COMMAND_SUCCESS, MCP2221_PID, MICROCHIP_VID};
use crate::gpio::{GpioChanges, GpioValues, Pins};
use crate::i2c::{I2cCancelTransferResponse, I2cSpeed, ReadType, WriteType};
use crate::settings::{
ChipSettings, DeviceString, GpSettings, InterruptSettingsChanges, SramSettingsChanges,
};
use crate::status::Status;
mod i2c_eh;
#[derive(Debug)]
pub struct MCP2221 {
inner: HidDevice,
pins_taken: Cell<bool>,
}
impl MCP2221 {
pub fn connect() -> Result<Self, Error> {
MCP2221::connect_with_vid_and_pid(MICROCHIP_VID, MCP2221_PID)
}
pub fn connect_with_vid_and_pid(vendor_id: u16, product_id: u16) -> Result<Self, Error> {
let hidapi = HidApi::new()?;
let device = hidapi.open(vendor_id, product_id)?;
Ok(Self {
inner: device,
pins_taken: Cell::new(false),
})
}
fn transfer(&self, command: &UsbReport) -> Result<Option<[u8; 64]>, Error> {
let out_command_byte = command.write_buffer[0];
let written = self.inner.write(&command.report_bytes())?;
if command.has_no_response() {
return Ok(None);
}
let mut read_buffer = [0u8; 64];
let read = self.inner.read(&mut read_buffer)?;
let read_command_byte = read_buffer[0];
assert_eq!(written, 65, "Didn't write full report.");
assert_eq!(read, 64, "Didn't read full report.");
if read_command_byte != out_command_byte {
return Err(Error::MismatchedCommandCodeEcho {
sent: out_command_byte,
received: read_command_byte,
});
}
let status_code = read_buffer[1];
if status_code == COMMAND_SUCCESS {
Ok(Some(read_buffer))
} else {
command
.check_error_code(status_code)
.and(Err(Error::CommandFailed(status_code)))
}
}
pub fn reset(self) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::ResetChip);
command.set_data_byte(2, 0xCD);
command.set_data_byte(3, 0xEF);
self.transfer(&command)?;
Ok(())
}
pub fn status(&self) -> Result<Status, Error> {
let buf = self
.transfer(&UsbReport::new(McpCommand::StatusSetParameters))?
.expect("Always has response buffer.");
Ok(Status::from_buffer(&buf))
}
pub fn i2c_set_bus_speed(&self, speed: I2cSpeed) -> Result<(), Error> {
let mut uc = UsbReport::new(McpCommand::StatusSetParameters);
uc.set_data_byte(3, 0x20);
uc.set_data_byte(4, speed.to_clock_divider());
let read_buffer = self.transfer(&uc)?.expect("Always has response buffer.");
match read_buffer[3] {
0x20 => Ok(()),
0x21 => Err(Error::I2cCouldNotChangeSpeed),
_ => unreachable!("Invalid response from MCP2221 for I2C speed set command."),
}
}
pub fn i2c_cancel_transfer(&self) -> Result<I2cCancelTransferResponse, Error> {
if self.status()?.i2c.communication_state.is_idle() {
return Ok(I2cCancelTransferResponse::NoTransfer);
}
let mut uc = UsbReport::new(McpCommand::StatusSetParameters);
uc.set_data_byte(2, 0x10);
let read_buffer = self.transfer(&uc)?.expect("Always has response buffer.");
match read_buffer[2] {
0x10 => Ok(I2cCancelTransferResponse::MarkedForCancellation),
0x11 => Ok(I2cCancelTransferResponse::NoTransfer),
0x00 => Ok(I2cCancelTransferResponse::Done),
code => unreachable!("Unknown code received from I2C cancel attempt {code}"),
}
}
pub fn i2c_read(&self, seven_bit_address: u8, read_buffer: &mut [u8]) -> Result<(), Error> {
self._i2c_read(seven_bit_address, read_buffer, ReadType::Normal)
}
pub fn i2c_read_repeated_start(
&self,
seven_bit_address: u8,
read_buffer: &mut [u8],
) -> Result<(), Error> {
self._i2c_read(seven_bit_address, read_buffer, ReadType::RepeatedStart)
}
fn i2c_bail_for_nack(&self) -> Result<(), Error> {
match self.status()?.i2c.target_acknowledged_address {
true => Ok(()),
false => {
self.i2c_cancel_transfer()?;
Err(Error::I2cAddressNack)
}
}
}
fn _i2c_read(
&self,
seven_bit_address: u8,
read_buffer: &mut [u8],
read_type: ReadType,
) -> Result<(), Error> {
if read_buffer.is_empty() {
return Err(Error::I2cTransferEmpty);
}
let Ok(tx_len): Result<u16, _> = read_buffer.len().try_into() else {
return Err(Error::I2cTransferTooLong);
};
use crate::i2c::I2cAddressing;
let mut read_command = UsbReport::new(read_type.into());
let [tx_len_low, tx_len_high] = tx_len.to_le_bytes();
read_command.set_data_byte(1, tx_len_low);
read_command.set_data_byte(2, tx_len_high);
read_command.set_data_byte(3, seven_bit_address.into_read_address());
self.transfer(&read_command)?;
self.i2c_bail_for_nack()?;
self.i2c_read_get_data(read_buffer)
}
fn i2c_read_get_data(&self, read_buffer: &mut [u8]) -> Result<(), Error> {
const MAX_RETRIES: u8 = 20;
const RETRY_DELAY: Duration = Duration::from_millis(2);
if read_buffer.is_empty() {
return Err(Error::I2cTransferEmpty);
}
let transfer_length = read_buffer.len();
let mut read_so_far: usize = 0;
let get_command = UsbReport::new(McpCommand::I2cGetData);
let mut retries = MAX_RETRIES;
while read_so_far < transfer_length {
match self.transfer(&get_command) {
Ok(Some(buffer)) => {
retries = MAX_RETRIES;
if buffer[3] == 127 {
continue;
}
let data_length = buffer[3] as usize;
read_buffer[read_so_far..read_so_far + data_length]
.copy_from_slice(&buffer[4..4 + data_length]);
read_so_far += data_length;
}
Ok(None) => unreachable!("Get Data always returns a buffer."),
Err(Error::I2cEngineReadError) if retries > 0 => {
retries -= 1;
std::thread::sleep(RETRY_DELAY);
continue;
}
e @ Err(_) => {
e?;
}
}
}
Ok(())
}
pub fn i2c_write(&self, seven_bit_address: u8, write_buffer: &[u8]) -> Result<(), Error> {
self._i2c_write(seven_bit_address, write_buffer, WriteType::Normal)
}
pub fn i2c_write_repeated_start(
&self,
seven_bit_address: u8,
write_buffer: &[u8],
) -> Result<(), Error> {
self._i2c_write(seven_bit_address, write_buffer, WriteType::RepeatedStart)
}
pub fn i2c_write_no_stop(
&self,
seven_bit_address: u8,
write_buffer: &[u8],
) -> Result<(), Error> {
self._i2c_write(seven_bit_address, write_buffer, WriteType::NoStop)
}
fn _i2c_write(
&self,
seven_bit_address: u8,
write_buffer: &[u8],
write_type: WriteType,
) -> Result<(), Error> {
let Ok([tx_len_low, tx_len_high]) = u16::try_from(write_buffer.len()).map(u16::to_le_bytes)
else {
return Err(Error::I2cTransferTooLong);
};
if write_buffer.is_empty() {
return Err(Error::I2cTransferEmpty);
}
use crate::i2c::I2cAddressing;
let mut command = UsbReport::new(write_type.into());
command.set_data_byte(1, tx_len_low);
command.set_data_byte(2, tx_len_high);
command.set_data_byte(3, seven_bit_address.into_write_address());
const MAX_RETRIES: u8 = 20;
const RETRY_DELAY: Duration = Duration::from_millis(2);
for (idx, chunk) in write_buffer.chunks(60).enumerate() {
let mut retries = MAX_RETRIES;
loop {
command.write_buffer[4..4 + chunk.len()].copy_from_slice(chunk);
match self.transfer(&command) {
Ok(_) => {
if idx == 0 {
self.i2c_bail_for_nack()?;
}
break;
}
Err(Error::I2cEngineBusy) if retries > 0 => {
retries -= 1;
std::thread::sleep(RETRY_DELAY);
continue;
}
e @ Err(_) => e?,
};
}
}
Ok(())
}
pub fn i2c_write_read(
&self,
seven_bit_address: u8,
write_buffer: &[u8],
read_buffer: &mut [u8],
) -> Result<(), Error> {
self.i2c_write_no_stop(seven_bit_address, write_buffer)?;
self.i2c_read_repeated_start(seven_bit_address, read_buffer)
}
pub fn i2c_check_address(&self, seven_bit_address: u8) -> Result<bool, Error> {
use crate::i2c::I2cAddressing;
let mut command = UsbReport::new(McpCommand::I2cWriteData);
command.set_data_byte(3, seven_bit_address.into_write_address());
const MAX_RETRIES: u8 = 20;
const RETRY_DELAY: Duration = Duration::from_millis(2);
for _ in 0..MAX_RETRIES {
match self.transfer(&command) {
Ok(_) => {
let address_ack = self.status()?.i2c.target_acknowledged_address;
self.i2c_cancel_transfer()?;
return Ok(address_ack);
}
Err(Error::I2cEngineBusy) => {
std::thread::sleep(RETRY_DELAY);
continue;
}
e @ Err(_) => {
e?;
}
};
}
Err(Error::I2cOperationFailed)
}
pub fn gpio_take_pins(&self) -> Option<Pins> {
if self.pins_taken.get() {
None
} else {
self.pins_taken.set(true);
Some(Pins::new(self))
}
}
pub fn gpio_read(&self) -> Result<GpioValues, Error> {
let buf = self
.transfer(&UsbReport::new(McpCommand::GetGpioValues))?
.expect("Always has response buffer.");
Ok(GpioValues::from_buffer(&buf))
}
pub fn gpio_write(&self, changes: &GpioChanges) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::SetGpioOutputValues);
changes.apply_to_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn analog_read(&self) -> Result<AdcReading, Error> {
let raw = self.status()?.adc_values;
let (chip_settings, gp) = self.sram_read_settings()?;
let reading = AdcReading {
vref: chip_settings.adc_reference,
gp1: gp.gp1_mode.is_analog_input().then_some(raw.ch1),
gp2: gp.gp2_mode.is_analog_input().then_some(raw.ch2),
gp3: gp.gp3_mode.is_analog_input().then_some(raw.ch3),
};
Ok(reading)
}
pub fn analog_write(&self, value: u8) -> Result<(), Error> {
self.sram_write_settings(SramSettingsChanges::new().with_dac_value(value))?;
Ok(())
}
pub fn analog_set_input_reference(&self, source: VoltageReference) -> Result<(), Error> {
self.sram_write_settings(SramSettingsChanges::new().with_adc_reference(source))?;
Ok(())
}
pub fn analog_set_output_reference(&self, source: VoltageReference) -> Result<(), Error> {
self.sram_write_settings(SramSettingsChanges::new().with_dac_reference(source))?;
Ok(())
}
pub fn interrupt_detected(&self) -> Result<bool, Error> {
self.status().map(|s| s.interrupt_detected)
}
pub fn interrupt_clear(&self) -> Result<(), Error> {
self.sram_write_settings(
SramSettingsChanges::new()
.with_interrupt_settings(InterruptSettingsChanges::clear_flag(true)),
)
}
pub fn sram_read_settings(&self) -> Result<(ChipSettings, GpSettings), Error> {
let command = UsbReport::new(McpCommand::GetSRAMSettings);
let buf = self
.transfer(&command)?
.expect("Always has response buffer.");
Ok((
ChipSettings::from_buffer(&buf),
GpSettings::try_from_sram_buffer(&buf)?,
))
}
pub fn sram_write_settings(&self, settings: &SramSettingsChanges) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::SetSRAMSettings);
settings.apply_to_sram_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn sram_write_gp_settings(&self, gp_settings: GpSettings) -> Result<(), Error> {
let (chip_settings, _) = self.sram_read_settings()?;
self.sram_write_settings(SramSettingsChanges::new().with_gp_modes(
gp_settings,
Some(chip_settings.dac_reference),
Some(chip_settings.adc_reference),
))
}
pub fn flash_read_chip_settings(&self) -> Result<ChipSettings, Error> {
let command = McpCommand::ReadFlashData(FlashDataSubCode::ChipSettings);
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
Ok(ChipSettings::from_buffer(&buf))
}
pub fn flash_write_chip_settings(&self, cs: ChipSettings) -> Result<(), Error> {
let mut command =
UsbReport::new(McpCommand::WriteFlashData(FlashDataSubCode::ChipSettings));
cs.apply_to_flash_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn flash_read_gp_settings(&self) -> Result<GpSettings, Error> {
let command = McpCommand::ReadFlashData(FlashDataSubCode::GPSettings);
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
GpSettings::try_from_flash_buffer(&buf)
}
pub fn flash_write_gp_settings(&self, gp: GpSettings) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::WriteFlashData(FlashDataSubCode::GPSettings));
gp.apply_to_flash_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn usb_manufacturer(&self) -> Result<DeviceString, Error> {
let command = McpCommand::ReadFlashData(FlashDataSubCode::UsbManufacturerDescriptor);
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
DeviceString::try_from_buffer(&buf)
}
pub fn usb_change_manufacturer(&self, s: &DeviceString) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::WriteFlashData(
FlashDataSubCode::UsbManufacturerDescriptor,
));
s.apply_to_flash_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn usb_product(&self) -> Result<DeviceString, Error> {
let command = McpCommand::ReadFlashData(FlashDataSubCode::UsbProductDescriptor);
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
DeviceString::try_from_buffer(&buf)
}
pub fn usb_change_product(&self, s: &DeviceString) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::WriteFlashData(
FlashDataSubCode::UsbProductDescriptor,
));
s.apply_to_flash_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn usb_serial_number(&self) -> Result<DeviceString, Error> {
let command = McpCommand::ReadFlashData(FlashDataSubCode::UsbSerialNumberDescriptor);
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
DeviceString::try_from_buffer(&buf)
}
pub fn usb_change_serial_number(&self, s: &DeviceString) -> Result<(), Error> {
let mut command = UsbReport::new(McpCommand::WriteFlashData(
FlashDataSubCode::UsbSerialNumberDescriptor,
));
s.apply_to_flash_buffer(&mut command.write_buffer);
self.transfer(&command)?;
Ok(())
}
pub fn factory_serial_number(&self) -> Result<String, Error> {
let command = McpCommand::ReadChipFactorySerialNumber;
let buf = self
.transfer(&UsbReport::new(command))?
.expect("Always has response buffer.");
let length = buf[2] as usize;
let serial_number_portion = &buf[4..(4 + length)];
Ok(String::from_utf8_lossy(serial_number_portion).into())
}
pub fn usb_device_info(&self) -> Result<hidapi::DeviceInfo, Error> {
self.inner.get_device_info().map_err(Error::from)
}
}