adafruit-mirror/zephyr
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 9

west: commands: runners: canopen: add download-buffer-size option

Browse source 
Adds a --download-buffer-size option to the canopen runner to allow
specifying the buffer size of the SDO download.

Signed-off-by: Fabian Blatz <fabianblatz@gmail.com>
This commit is contained in:
Fabian Blatz 2024-09-19 16:26:45 +02:00 committed by Mahesh Mahadevan
parent 46922b51ce
commit 11c83c80fc
2 changed files with 19 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
scripts/west_commands/runners/canopen_program.py
View file

@ -24,10 +24,7 @@ DEFAULT_CAN_CONTEXT = 'default'
DEFAULT_PROGRAM_NUMBER = 1
# Program download buffer size in bytes
PROGRAM_DOWNLOAD_BUFFER_SIZE = 1024
# Program download chunk size in bytes
PROGRAM_DOWNLOAD_CHUNK_SIZE = PROGRAM_DOWNLOAD_BUFFER_SIZE // 2
DEFAULT_PROGRAM_DOWNLOAD_BUFFER_SIZE = 1024
# Default timeouts and retries
DEFAULT_TIMEOUT = 10.0 # seconds
@ -58,7 +55,7 @@ class CANopenBinaryRunner(ZephyrBinaryRunner):
program_number=DEFAULT_PROGRAM_NUMBER, confirm=True,
confirm_only=True, timeout=DEFAULT_TIMEOUT,
sdo_retries=DEFAULT_SDO_RETRIES, sdo_timeout=DEFAULT_SDO_TIMEOUT,
block_transfer=False):
download_buffer_size=DEFAULT_PROGRAM_DOWNLOAD_BUFFER_SIZE, block_transfer=False):
if MISSING_REQUIREMENTS:
raise RuntimeError('one or more Python dependencies were missing; '
"see the getting started guide for details on "
@ -76,7 +73,9 @@ class CANopenBinaryRunner(ZephyrBinaryRunner):
program_number=program_number,
sdo_retries=sdo_retries,
sdo_timeout=sdo_timeout,
block_transfer=block_transfer)
download_buffer_size=download_buffer_size,
block_transfer=block_transfer,
)
@classmethod
def name(cls):
@ -113,6 +112,8 @@ class CANopenBinaryRunner(ZephyrBinaryRunner):
parser.add_argument('--sdo-timeout', type=float, default=DEFAULT_SDO_TIMEOUT,
help=f'''CANopen SDO response timeout in seconds
(default: {DEFAULT_SDO_TIMEOUT})''')
parser.add_argument('--download-buffer-size', type=int, default=DEFAULT_PROGRAM_DOWNLOAD_BUFFER_SIZE,
help=f'Program download buffer size in bytes (default: {DEFAULT_PROGRAM_DOWNLOAD_BUFFER_SIZE})')
parser.add_argument('--block-transfer', default=False, action='store_true',
help='Use SDO block transfers (experimental, default: no)')
@ -128,6 +129,7 @@ class CANopenBinaryRunner(ZephyrBinaryRunner):
timeout=args.timeout,
sdo_retries=args.sdo_retries,
sdo_timeout=args.sdo_timeout,
download_buffer_size=args.download_buffer_size,
block_transfer=args.block_transfer)
def do_run(self, command, **kwargs):
@ -188,7 +190,7 @@ class CANopenProgramDownloader(object):
def __init__(self, logger, node_id, can_context=DEFAULT_CAN_CONTEXT,
program_number=DEFAULT_PROGRAM_NUMBER,
sdo_retries=DEFAULT_SDO_RETRIES, sdo_timeout=DEFAULT_SDO_TIMEOUT,
block_transfer=False):
download_buffer_size=DEFAULT_PROGRAM_DOWNLOAD_BUFFER_SIZE, block_transfer=False):
super(CANopenProgramDownloader, self).__init__()
self.logger = logger
self.node_id = node_id
@ -201,6 +203,7 @@ class CANopenProgramDownloader(object):
self.ctrl_sdo = self.node.sdo[H1F51_PROGRAM_CTRL][self.program_number]
self.swid_sdo = self.node.sdo[H1F56_PROGRAM_SWID][self.program_number]
self.flash_sdo = self.node.sdo[H1F57_FLASH_STATUS][self.program_number]
self.download_buffer_size = download_buffer_size
self.node.sdo.MAX_RETRIES = sdo_retries
self.node.sdo.RESPONSE_TIMEOUT = sdo_timeout
@ -276,12 +279,12 @@ class CANopenProgramDownloader(object):
try:
size = os.path.getsize(bin_file)
infile = open(bin_file, 'rb')
outfile = self.data_sdo.open('wb', buffering=PROGRAM_DOWNLOAD_BUFFER_SIZE,
outfile = self.data_sdo.open('wb', buffering=self.download_buffer_size,
size=size, block_transfer=self.block_transfer)
progress = Bar('%(percent)d%%', max=size, suffix='%(index)d/%(max)dB')
while True:
chunk = infile.read(PROGRAM_DOWNLOAD_CHUNK_SIZE)
chunk = infile.read(self.download_buffer_size // 2)
if not chunk:
break
outfile.write(chunk)

9
scripts/west_commands/tests/test_canopen_program.py
View file

@ -22,7 +22,7 @@ TEST_ALT_CONTEXT = 'alternate'
# Test cases
#
TEST_CASES = [(n, x, p, c, o, t, r, s, b)
TEST_CASES = [(n, x, p, c, o, t, r, s, d, b)
for n in range(1, 3)
for x in (None, TEST_ALT_CONTEXT)
for p in range(1, 3)
@ -31,6 +31,7 @@ TEST_CASES = [(n, x, p, c, o, t, r, s, b)
for t in range(1, 3)
for r in range(1, 3)
for s in range(1, 3)
for d in [256, 1024]
for b in range(False, True)]
os_path_isfile = os.path.isfile
@ -44,7 +45,7 @@ def os_path_isfile_patch(filename):
@patch('runners.canopen_program.CANopenProgramDownloader')
def test_canopen_program_create(cpd, test_case, runner_config):
'''Test CANopen runner created from command line parameters.'''
node_id, context, program_number, confirm, confirm_only, timeout, sdo_retries, sdo_timeout, block_transfer = test_case
node_id, context, program_number, confirm, confirm_only, timeout, sdo_retries, sdo_timeout, download_buffer_size, block_transfer = test_case
args = ['--node-id', str(node_id)]
if context is not None:
@ -61,6 +62,8 @@ def test_canopen_program_create(cpd, test_case, runner_config):
args.extend(['--sdo-retries', str(sdo_retries)])
if sdo_timeout:
args.extend(['--sdo-timeout', str(sdo_timeout)])
if download_buffer_size:
args.extend(['--download-buffer-size', str(download_buffer_size)])
if block_transfer:
args.append('--block_transfer')
@ -84,6 +87,7 @@ def test_canopen_program_create(cpd, test_case, runner_config):
program_number=program_number,
sdo_retries=sdo_retries,
sdo_timeout=sdo_timeout,
download_buffer_size=download_buffer_size,
block_transfer=block_transfer)
else:
assert cpd.call_args == call(node_id=node_id,
@ -92,6 +96,7 @@ def test_canopen_program_create(cpd, test_case, runner_config):
program_number=program_number,
sdo_retries=sdo_retries,
sdo_timeout=sdo_timeout,
download_buffer_size=download_buffer_size,
block_transfer=block_transfer)
mock.connect.assert_called_once()