rust/va416xx-rs
7
0
Fork 0
Code Issues Pull Requests Projects Releases 14 Wiki Activity

Flashloader and UART improvements
Some checks are pending
Rust/va416xx-rs/pipeline/pr-main Build started...
Details

Browse Source
This commit is contained in:
Robin Müller
2024-09-14 11:29:51 +02:00
parent 3e67749452
commit aae870c767
6 changed files with 608 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
flashloader/README.md
View File

@ -6,12 +6,18 @@ a simple PUS (CCSDS) interface to update the software. It also provides a Python
called the `image-loader.py` which can be used to upload compiled images to the flashloader
application to write them to the NVM.
Please note that the both the application and the image loader are tailored towards usage
with the [bootloader provided by this repository](https://egit.irs.uni-stuttgart.de/rust/va416xx-rs/src/branch/main/bootloader).
The software can quickly be adapted to interface with a real primary on-board software instead of
the Python script provided here to upload images because it uses a low-level CCSDS based packet
interface.
## Using the Python image loader
The Python image loader communicates with the Rust flashload application using a dedicated serial
port with a baudrate of 115200.
It is recommended to run the script in a dedicated virtual environment. For example, on UNIX
systems you can use `python3 -m venv venv` and then `source venv/bin/activate` to create
and activate a virtual environment.

498
flashloader/image-loader.py
View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3
from typing import List, Tuple
from spacepackets.ecss.defs import PusService
from spacepackets.ecss.tm import PusTm
from tmtccmd.com import ComInterface
import toml
import struct
import logging
@ -21,20 +23,27 @@ from elftools.elf.elffile import ELFFile
BAUD_RATE = 115200
BOOTLOADER_START_ADDR = 0x0
BOOTLOADER_END_ADDR = 0x4000
BOOTLOADER_CRC_ADDR = 0x3FFC
BOOTLOADER_CRC_ADDR = BOOTLOADER_END_ADDR - 4
BOOTLOADER_MAX_SIZE = BOOTLOADER_END_ADDR - BOOTLOADER_START_ADDR - 4
APP_A_START_ADDR = 0x4000
APP_A_END_ADDR = 0x22000
# The actual size of the image which is relevant for CRC calculation.
APP_A_SIZE_ADDR = 0x21FF8
APP_A_CRC_ADDR = 0x21FFC
APP_A_SIZE_ADDR = APP_A_END_ADDR - 8
APP_A_CRC_ADDR = APP_A_END_ADDR - 4
APP_A_MAX_SIZE = APP_A_END_ADDR - APP_A_START_ADDR - 8
APP_B_START_ADDR = 0x22000
APP_B_END_ADDR = 0x40000
# The actual size of the image which is relevant for CRC calculation.
APP_B_SIZE_ADDR = 0x3FFF8
APP_B_CRC_ADDR = 0x3FFFC
APP_IMG_SZ = 0x1E000
APP_B_SIZE_ADDR = APP_B_END_ADDR - 8
APP_B_CRC_ADDR = APP_B_END_ADDR - 4
APP_B_MAX_SIZE = APP_A_END_ADDR - APP_A_START_ADDR - 8
APP_IMG_SZ = (APP_B_END_ADDR - APP_A_START_ADDR) // 2
CHUNK_SIZE = 896
@ -52,6 +61,7 @@ class ActionId(enum.IntEnum):
_LOGGER = logging.getLogger(__name__)
SEQ_PROVIDER = SeqCountProvider(bit_width=14)
@dataclasses.dataclass
@ -62,7 +72,174 @@ class LoadableSegment:
data: bytes
SEQ_PROVIDER = SeqCountProvider(bit_width=14)
class Target(enum.Enum):
BOOTLOADER = 0
APP_A = 1
APP_B = 2
class ImageLoader:
def __init__(self, com_if: ComInterface, verificator: PusVerificator) -> None:
self.com_if = com_if
self.verificator = verificator
def handle_ping_cmd(self):
_LOGGER.info("Sending ping command")
ping_tc = PusTc(
apid=0x00,
service=PusService.S17_TEST,
subservice=1,
seq_count=SEQ_PROVIDER.get_and_increment(),
app_data=bytes(PING_PAYLOAD_SIZE),
)
self.verificator.add_tc(ping_tc)
self.com_if.send(bytes(ping_tc.pack()))
data_available = self.com_if.data_available(0.4)
if not data_available:
_LOGGER.warning("no ping reply received")
for reply in self.com_if.receive():
result = self.verificator.add_tm(
Service1Tm.from_tm(PusTm.unpack(reply, 0), UnpackParams(0))
)
if result is not None and result.completed:
_LOGGER.info("received ping completion reply")
def handle_corruption_cmd(self, target: Target):
if target == Target.BOOTLOADER:
_LOGGER.error("can not corrupt bootloader")
if target == Target.APP_A:
self.send_tc(
PusTc(
apid=0,
service=ACTION_SERVICE,
subservice=ActionId.CORRUPT_APP_A,
),
)
if target == Target.APP_B:
self.send_tc(
PusTc(
apid=0,
service=ACTION_SERVICE,
subservice=ActionId.CORRUPT_APP_B,
),
)
def handle_flash_cmd(self, target: Target, file_path: Path) -> int:
loadable_segments = []
_LOGGER.info("Parsing ELF file for loadable sections")
total_size = 0
loadable_segments, total_size = create_loadable_segments(target, file_path)
segments_info_str(target, loadable_segments, total_size, file_path)
result = self._perform_flashing_algorithm(loadable_segments)
if result != 0:
return result
self._crc_and_app_size_postprocessing(target, total_size, loadable_segments)
return 0
def _perform_flashing_algorithm(
self,
loadable_segments: List[LoadableSegment],
) -> int:
# Perform the flashing algorithm.
for segment in loadable_segments:
segment_end = segment.offset + segment.size
current_addr = segment.offset
pos_in_segment = 0
while pos_in_segment < segment.size:
next_chunk_size = min(segment_end - current_addr, CHUNK_SIZE)
data = segment.data[pos_in_segment : pos_in_segment + next_chunk_size]
next_packet = pack_memory_write_command(current_addr, data)
_LOGGER.info(
f"Sending memory write command for address {current_addr:#08x} and data with "
f"length {len(data)}"
)
self.verificator.add_tc(next_packet)
self.com_if.send(bytes(next_packet.pack()))
current_addr += next_chunk_size
pos_in_segment += next_chunk_size
start_time = time.time()
while True:
if time.time() - start_time > 1.0:
_LOGGER.error("Timeout while waiting for reply")
return -1
data_available = self.com_if.data_available(0.1)
done = False
if not data_available:
continue
replies = self.com_if.receive()
for reply in replies:
tm = PusTm.unpack(reply, 0)
if tm.service != 1:
continue
service_1_tm = Service1Tm.from_tm(tm, UnpackParams(0))
check_result = self.verificator.add_tm(service_1_tm)
# We could send after we have received the step reply, but that can
# somehow lead to overrun errors. I think it's okay to do it like
# this as long as the flash loader only uses polling..
if (
check_result is not None
and check_result.status.completed == StatusField.SUCCESS
):
done = True
# This is an optimized variant, but I think the small delay is not an issue.
"""
if (
check_result is not None
and check_result.status.step == StatusField.SUCCESS
and len(check_result.status.step_list) == 1
):
done = True
"""
self.verificator.remove_completed_entries()
if done:
break
return 0
def _crc_and_app_size_postprocessing(
self,
target: Target,
total_size: int,
loadable_segments: List[LoadableSegment],
):
if target == Target.BOOTLOADER:
_LOGGER.info("Blanking the bootloader checksum")
# Blank the checksum. For the bootloader, the bootloader will calculate the
# checksum itself on the initial run.
checksum_write_packet = pack_memory_write_command(
BOOTLOADER_CRC_ADDR, bytes([0x00, 0x00, 0x00, 0x00])
)
self.send_tc(checksum_write_packet)
else:
crc_addr = None
size_addr = None
if target == Target.APP_A:
crc_addr = APP_A_CRC_ADDR
size_addr = APP_A_SIZE_ADDR
elif target == Target.APP_B:
crc_addr = APP_B_CRC_ADDR
size_addr = APP_B_SIZE_ADDR
assert crc_addr is not None
assert size_addr is not None
_LOGGER.info(f"Writing app size {total_size} at address {size_addr:#08x}")
size_write_packet = pack_memory_write_command(
size_addr, struct.pack("!I", total_size)
)
self.com_if.send(bytes(size_write_packet.pack()))
time.sleep(0.2)
crc_calc = PredefinedCrc("crc-32")
for segment in loadable_segments:
crc_calc.update(segment.data)
checksum = crc_calc.digest()
_LOGGER.info(
f"Writing checksum 0x[{checksum.hex(sep=',')}] at address {crc_addr:#08x}"
)
self.send_tc(pack_memory_write_command(crc_addr, checksum))
def send_tc(self, tc: PusTc):
self.com_if.send(bytes(tc.pack()))
def main() -> int:
@ -102,213 +279,134 @@ def main() -> int:
verificator = PusVerificator()
com_if = SerialCobsComIF(serial_cfg)
com_if.open()
target = None
if args.target == "bl":
target = Target.BOOTLOADER
elif args.target == "a":
target = Target.APP_A
elif args.target == "b":
target = Target.APP_B
image_loader = ImageLoader(com_if, verificator)
file_path = None
result = -1
if args.ping:
_LOGGER.info("Sending ping command")
ping_tc = PusTc(
apid=0x00,
service=PusService.S17_TEST,
subservice=1,
seq_count=SEQ_PROVIDER.get_and_increment(),
app_data=bytes(PING_PAYLOAD_SIZE),
)
verificator.add_tc(ping_tc)
com_if.send(ping_tc.pack())
data_available = com_if.data_available(0.4)
if not data_available:
_LOGGER.warning("no ping reply received")
for reply in com_if.receive():
result = verificator.add_tm(
Service1Tm.from_tm(PusTm.unpack(reply, 0), UnpackParams(0))
)
if result is not None and result.completed:
_LOGGER.info("received ping completion reply")
if not args.target:
return 0
if args.target:
image_loader.handle_ping_cmd()
com_if.close()
return 0
if target:
if not args.corrupt:
if not args.path:
_LOGGER.error("App Path needs to be specified for the flash process")
return -1
file_path = Path(args.path)
if not file_path.exists():
_LOGGER.error("File does not exist")
return -1
if args.corrupt:
if not args.target:
if not target:
_LOGGER.error("target for corruption command required")
com_if.close()
return -1
if args.target == "bl":
_LOGGER.error("can not corrupt bootloader")
if args.target == "a":
packet = PusTc(
apid=0,
service=ACTION_SERVICE,
subservice=ActionId.CORRUPT_APP_A,
)
com_if.send(packet.pack())
if args.target == "b":
packet = PusTc(
apid=0,
service=ACTION_SERVICE,
subservice=ActionId.CORRUPT_APP_B,
)
com_if.send(packet.pack())
image_loader.handle_corruption_cmd(target)
else:
assert file_path is not None
loadable_segments = []
_LOGGER.info("Parsing ELF file for loadable sections")
total_size = 0
with open(file_path, "rb") as app_file:
elf_file = ELFFile(app_file)
assert target is not None
result = image_loader.handle_flash_cmd(target, file_path)
for (idx, segment) in enumerate(elf_file.iter_segments("PT_LOAD")):
if segment.header.p_filesz == 0:
continue
# Basic validity checks of the base addresses.
if idx == 0:
if (
args.target == "bl"
and segment.header.p_paddr != BOOTLOADER_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"bootloader, expected {BOOTLOADER_START_ADDR}"
)
if (
args.target == "a"
and segment.header.p_paddr != APP_A_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"App A, expected {APP_A_START_ADDR}"
)
if (
args.target == "b"
and segment.header.p_paddr != APP_B_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"App B, expected {APP_B_START_ADDR}"
)
name = None
for section in elf_file.iter_sections():
if (
section.header.sh_offset == segment.header.p_offset
and section.header.sh_size > 0
):
name = section.name
if name is None:
_LOGGER.warning("no fitting section found for segment")
continue
# print(f"Segment Addr: {segment.header.p_paddr}")
# print(f"Segment Offset: {segment.header.p_offset}")
# print(f"Segment Filesize: {segment.header.p_filesz}")
loadable_segments.append(
LoadableSegment(
name=name,
offset=segment.header.p_paddr,
size=segment.header.p_filesz,
data=segment.data(),
)
)
total_size += segment.header.p_filesz
context_str = None
if args.target == "bl":
context_str = "Bootloader"
elif args.target == "a":
context_str = "App Slot A"
elif args.target == "b":
context_str = "App Slot B"
_LOGGER.info(
f"Flashing {context_str} with image {file_path} (size {total_size})"
)
for idx, segment in enumerate(loadable_segments):
_LOGGER.info(
f"Loadable section {idx} {segment.name} with offset {segment.offset:#08x} and size {segment.size}"
)
for segment in loadable_segments:
segment_end = segment.offset + segment.size
current_addr = segment.offset
pos_in_segment = 0
while pos_in_segment < segment.size:
next_chunk_size = min(segment_end - current_addr, CHUNK_SIZE)
data = segment.data[
pos_in_segment : pos_in_segment + next_chunk_size
]
next_packet = pack_memory_write_command(current_addr, data)
_LOGGER.info(
f"Sending memory write command for address {current_addr:#08x} and data with "
f"length {len(data)}"
)
verificator.add_tc(next_packet)
com_if.send(next_packet.pack())
current_addr += next_chunk_size
pos_in_segment += next_chunk_size
while True:
data_available = com_if.data_available(0.1)
done = False
if not data_available:
continue
replies = com_if.receive()
for reply in replies:
tm = PusTm.unpack(reply, 0)
if tm.service != 1:
continue
service_1_tm = Service1Tm.from_tm(tm, UnpackParams(0))
check_result = verificator.add_tm(service_1_tm)
# We could send after we have received the step reply, but that can
# somehow lead to overrun errors. I think it's okay to do it like
# this as long as the flash loader only uses polling..
if (
check_result is not None
and check_result.status.completed == StatusField.SUCCESS
):
done = True
# Still keep a small delay
# time.sleep(0.05)
verificator.remove_completed_entries()
if done:
break
if args.target == "bl":
_LOGGER.info("Blanking the bootloader checksum")
# Blank the checksum. For the bootloader, the bootloader will calculate the
# checksum itself on the initial run.
checksum_write_packet = pack_memory_write_command(
BOOTLOADER_CRC_ADDR, bytes([0x00, 0x00, 0x00, 0x00])
)
com_if.send(checksum_write_packet.pack())
else:
crc_addr = None
size_addr = None
if args.target == "a":
crc_addr = APP_A_CRC_ADDR
size_addr = APP_A_SIZE_ADDR
elif args.target == "b":
crc_addr = APP_B_CRC_ADDR
size_addr = APP_B_SIZE_ADDR
assert crc_addr is not None
assert size_addr is not None
_LOGGER.info(
f"Writing app size {total_size} at address {size_addr:#08x}"
)
size_write_packet = pack_memory_write_command(
size_addr, struct.pack("!I", total_size)
)
com_if.send(size_write_packet.pack())
time.sleep(0.2)
crc_calc = PredefinedCrc("crc-32")
for segment in loadable_segments:
crc_calc.update(segment.data)
checksum = crc_calc.digest()
_LOGGER.info(
f"Writing checksum 0x[{checksum.hex(sep=',')}] at address {crc_addr:#08x}"
)
checksum_write_packet = pack_memory_write_command(crc_addr, checksum)
com_if.send(checksum_write_packet.pack())
com_if.close()
return 0
return result
def create_loadable_segments(
target: Target, file_path: Path
) -> Tuple[List[LoadableSegment], int]:
loadable_segments = []
total_size = 0
with open(file_path, "rb") as app_file:
elf_file = ELFFile(app_file)
for idx, segment in enumerate(elf_file.iter_segments("PT_LOAD")):
if segment.header.p_filesz == 0:
continue
# Basic validity checks of the base addresses.
if idx == 0:
if (
target == Target.BOOTLOADER
and segment.header.p_paddr != BOOTLOADER_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"bootloader, expected {BOOTLOADER_START_ADDR}"
)
if (
target == Target.APP_A
and segment.header.p_paddr != APP_A_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"App A, expected {APP_A_START_ADDR}"
)
if (
target == Target.APP_B
and segment.header.p_paddr != APP_B_START_ADDR
):
raise ValueError(
f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
f"App B, expected {APP_B_START_ADDR}"
)
name = None
for section in elf_file.iter_sections():
if (
section.header.sh_offset == segment.header.p_offset
and section.header.sh_size > 0
):
name = section.name
if name is None:
_LOGGER.warning("no fitting section found for segment")
continue
# print(f"Segment Addr: {segment.header.p_paddr}")
# print(f"Segment Offset: {segment.header.p_offset}")
# print(f"Segment Filesize: {segment.header.p_filesz}")
loadable_segments.append(
LoadableSegment(
name=name,
offset=segment.header.p_paddr,
size=segment.header.p_filesz,
data=segment.data(),
)
)
total_size += segment.header.p_filesz
return loadable_segments, total_size
def segments_info_str(
target: Target,
loadable_segments: List[LoadableSegment],
total_size: int,
file_path: Path,
):
# Set context string and perform basic sanity checks.
if target == Target.BOOTLOADER:
if total_size > BOOTLOADER_MAX_SIZE:
_LOGGER.error(
f"provided bootloader app larger than allowed {total_size} bytes"
)
return -1
context_str = "Bootloader"
elif target == Target.APP_A:
if total_size > APP_A_MAX_SIZE:
_LOGGER.error(f"provided App A larger than allowed {total_size} bytes")
return -1
context_str = "App Slot A"
elif target == Target.APP_B:
if total_size > APP_B_MAX_SIZE:
_LOGGER.error(f"provided App B larger than allowed {total_size} bytes")
return -1
context_str = "App Slot B"
_LOGGER.info(f"Flashing {context_str} with image {file_path} (size {total_size})")
for idx, segment in enumerate(loadable_segments):
_LOGGER.info(
f"Loadable section {idx} {segment.name} with offset {segment.offset:#08x} and "
f"size {segment.size}"
)
def pack_memory_write_command(addr: int, data: bytes) -> PusTc:
@ -324,7 +422,7 @@ def pack_memory_write_command(addr: int, data: bytes) -> PusTc:
service=MEMORY_SERVICE,
subservice=RAW_MEMORY_WRITE_SUBSERVICE,
seq_count=SEQ_PROVIDER.get_and_increment(),
app_data=app_data,
app_data=bytes(app_data),
)

32
flashloader/src/main.rs
View File

@ -109,6 +109,7 @@ mod app {
tc::PusTcReader, tm::PusTmCreator, EcssEnumU8, PusPacket, WritablePusPacket,
};
use va416xx_hal::irq_router::enable_and_init_irq_router;
use va416xx_hal::uart::IrqContextTimeoutOrMaxSize;
use va416xx_hal::{
clock::ClkgenExt,
edac,
@ -132,6 +133,7 @@ mod app {
struct Local {
uart_rx: uart::RxWithIrq<pac::Uart0>,
uart_tx: uart::Tx<pac::Uart0>,
rx_context: IrqContextTimeoutOrMaxSize,
rom_spi: Option<pac::Spi3>,
// We handle all TM in one task.
tm_cons: DataConsumer<BUF_RB_SIZE_TM, SIZES_RB_SIZE_TM>,
@ -178,7 +180,7 @@ mod app {
&mut cx.device.sysconfig,
&clocks,
);
let (tx, mut rx, _) = uart0.split_with_irq();
let (tx, rx) = uart0.split();
let verif_reporter = VerificationReportCreator::new(0).unwrap();
@ -191,7 +193,9 @@ mod app {
Mono::start(cx.core.SYST, clocks.sysclk().raw());
CLOCKS.set(clocks).unwrap();
rx.read_fixed_len_using_irq(MAX_TC_FRAME_SIZE, true)
let mut rx = rx.to_rx_with_irq();
let mut rx_context = IrqContextTimeoutOrMaxSize::new(MAX_TC_FRAME_SIZE);
rx.read_fixed_len_or_timeout_based_using_irq(&mut rx_context)
.expect("initiating UART RX failed");
pus_tc_handler::spawn().unwrap();
pus_tm_tx_handler::spawn().unwrap();
@ -205,6 +209,7 @@ mod app {
Local {
uart_rx: rx,
uart_tx: tx,
rx_context,
rom_spi: Some(cx.device.spi3),
tm_cons: DataConsumer {
buf_cons: buf_cons_tm,
@ -231,20 +236,26 @@ mod app {
}
}
// This is the interrupt handler to read all bytes received on the UART0.
#[task(
binds = UART0_RX,
local = [
cnt: u32 = 0,
rx_buf: [u8; MAX_TC_FRAME_SIZE] = [0; MAX_TC_FRAME_SIZE],
rx_context,
uart_rx,
tc_prod
],
)]
fn uart_rx_irq(cx: uart_rx_irq::Context) {
match cx.local.uart_rx.irq_handler(cx.local.rx_buf) {
match cx
.local
.uart_rx
.irq_handler_max_size_or_timeout_based(cx.local.rx_context, cx.local.rx_buf)
{
Ok(result) => {
if RX_DEBUGGING {
log::debug!("RX Info: {:?}", cx.local.uart_rx.irq_info());
log::debug!("RX Info: {:?}", cx.local.rx_context);
log::debug!("RX Result: {:?}", result);
}
if result.complete() {
@ -279,7 +290,7 @@ mod app {
// Initiate next transfer.
cx.local
.uart_rx
.read_fixed_len_using_irq(MAX_TC_FRAME_SIZE, true)
.read_fixed_len_or_timeout_based_using_irq(cx.local.rx_context)
.expect("read operation failed");
}
if result.error() {
@ -438,7 +449,12 @@ mod app {
return;
}
let data = &app_data[10..10 + data_len as usize];
log::info!("writing {} bytes at offset {} to NVM", data_len, offset);
log::info!(
target: "TC Handler",
"writing {} bytes at offset {} to NVM",
data_len,
offset
);
// Safety: We only use this for NVM handling and we only do NVM
// handling here.
let mut sys_cfg = unsafe { pac::Sysconfig::steal() };
@ -455,7 +471,9 @@ mod app {
.completion_success(cx.local.src_data_buf, started_token, 0, 0, &[])
.expect("completion success failed");
write_and_send(&tm);
log::info!("NVM operation done");
log::info!(
target: "TC Handler",
"NVM operation done");
}
}
}