mirror of
https://github.com/espressif/openthread.git
synced 2026-08-11 05:07:47 +00:00
[tcat] feat: extend dataset hex command (#10831)
This commit extends the 'dataset hex' command in the bbtc.py script by allowing dataset TLVs to be set using a hex-encoded format. Till now the 'dataset hex' command was only printing the 'ThreadDataset' object values in hex-encoded format, there was no functionality to set the TLVs using hex-encoded format. The 'dataset hex' command has been modified so the user can pass dataset TLVs in hex-encoded format as an argument to this command. This enables the script to set desired dataset TLVs in one command, instead of calling dataset commands individually. Example usage: 'dataset hex <hex-encoded TLVs>'
This commit is contained in:
@@ -29,6 +29,7 @@
|
||||
from cli.command import Command, CommandResultNone
|
||||
from dataset.dataset import ThreadDataset, initial_dataset
|
||||
from tlv.dataset_tlv import MeshcopTlvType
|
||||
from copy import deepcopy
|
||||
|
||||
|
||||
def handle_dataset_entry_command(type: MeshcopTlvType, args, context):
|
||||
@@ -71,13 +72,23 @@ class DatasetHelpCommand(Command):
|
||||
return CommandResultNone()
|
||||
|
||||
|
||||
class PrintDatasetHexCommand(Command):
|
||||
class DatasetHexCommand(Command):
|
||||
|
||||
def get_help_string(self) -> str:
|
||||
return 'Print current dataset as a hexadecimal string.'
|
||||
return 'Get or set dataset as hex-encoded TLVs.'
|
||||
|
||||
async def execute_default(self, args, context):
|
||||
ds: ThreadDataset = context['dataset']
|
||||
if args:
|
||||
try:
|
||||
ds_tmp = deepcopy(ds)
|
||||
tlvs_str: str = args[0]
|
||||
tlvs_bytes = bytes.fromhex(tlvs_str)
|
||||
ds_tmp.set_from_bytes(tlvs_bytes)
|
||||
ds.clear()
|
||||
ds.set_from_bytes(tlvs_bytes)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print(ds.to_bytes().hex())
|
||||
return CommandResultNone()
|
||||
|
||||
@@ -207,7 +218,7 @@ class DatasetCommand(Command):
|
||||
self._subcommands = {
|
||||
'clear': DatasetClearCommand(),
|
||||
'help': DatasetHelpCommand(),
|
||||
'hex': PrintDatasetHexCommand(),
|
||||
'hex': DatasetHexCommand(),
|
||||
'reload': ReloadDatasetCommand(),
|
||||
'activetimestamp': ActiveTimestampCommand(),
|
||||
'pendingtimestamp': PendingTimestampCommand(),
|
||||
|
||||
@@ -18,7 +18,7 @@ from typing import Dict, List
|
||||
|
||||
from tlv.tlv import TLV
|
||||
from tlv.dataset_tlv import MeshcopTlvType
|
||||
from dataset.dataset_entries import DatasetEntry, create_dataset_entry
|
||||
from dataset.dataset_entries import DatasetEntry, create_dataset_entry, ENTRY_CLASSES
|
||||
|
||||
initial_dataset = bytes([
|
||||
0x0E, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x12, 0x35, 0x06, 0x00, 0x04,
|
||||
@@ -58,8 +58,11 @@ class ThreadDataset:
|
||||
return self.entries[type]
|
||||
|
||||
def set_entry(self, type: MeshcopTlvType, args: List[str]):
|
||||
if type in self.entries:
|
||||
self.entries[type].set(args)
|
||||
if type in ENTRY_CLASSES:
|
||||
if type in self.entries:
|
||||
self.entries[type].set(args)
|
||||
else:
|
||||
self.entries[type] = create_dataset_entry(type, args)
|
||||
return
|
||||
raise KeyError(f'Key {type} not available in the dataset.')
|
||||
|
||||
|
||||
@@ -476,23 +476,24 @@ class ChannelMaskEntry(DatasetEntry):
|
||||
return TLV.from_bytes(tlv)
|
||||
|
||||
|
||||
def create_dataset_entry(type: MeshcopTlvType, args=None):
|
||||
entry_classes = {
|
||||
MeshcopTlvType.ACTIVETIMESTAMP: ActiveTimestamp,
|
||||
MeshcopTlvType.PENDINGTIMESTAMP: PendingTimestamp,
|
||||
MeshcopTlvType.NETWORKKEY: NetworkKey,
|
||||
MeshcopTlvType.NETWORKNAME: NetworkName,
|
||||
MeshcopTlvType.EXTPANID: ExtPanID,
|
||||
MeshcopTlvType.MESHLOCALPREFIX: MeshLocalPrefix,
|
||||
MeshcopTlvType.DELAYTIMER: DelayTimer,
|
||||
MeshcopTlvType.PANID: PanID,
|
||||
MeshcopTlvType.CHANNEL: Channel,
|
||||
MeshcopTlvType.PSKC: Pskc,
|
||||
MeshcopTlvType.SECURITYPOLICY: SecurityPolicy,
|
||||
MeshcopTlvType.CHANNELMASK: ChannelMask
|
||||
}
|
||||
ENTRY_CLASSES = {
|
||||
MeshcopTlvType.ACTIVETIMESTAMP: ActiveTimestamp,
|
||||
MeshcopTlvType.PENDINGTIMESTAMP: PendingTimestamp,
|
||||
MeshcopTlvType.NETWORKKEY: NetworkKey,
|
||||
MeshcopTlvType.NETWORKNAME: NetworkName,
|
||||
MeshcopTlvType.EXTPANID: ExtPanID,
|
||||
MeshcopTlvType.MESHLOCALPREFIX: MeshLocalPrefix,
|
||||
MeshcopTlvType.DELAYTIMER: DelayTimer,
|
||||
MeshcopTlvType.PANID: PanID,
|
||||
MeshcopTlvType.CHANNEL: Channel,
|
||||
MeshcopTlvType.PSKC: Pskc,
|
||||
MeshcopTlvType.SECURITYPOLICY: SecurityPolicy,
|
||||
MeshcopTlvType.CHANNELMASK: ChannelMask
|
||||
}
|
||||
|
||||
entry_class = entry_classes.get(type)
|
||||
|
||||
def create_dataset_entry(type: MeshcopTlvType, args=None):
|
||||
entry_class = ENTRY_CLASSES.get(type)
|
||||
if not entry_class:
|
||||
raise ValueError(f"Invalid configuration type: {type}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user