diff --git a/tools/tcat_ble_client/cli/dataset_commands.py b/tools/tcat_ble_client/cli/dataset_commands.py index bac050ddf..8fd268ea8 100644 --- a/tools/tcat_ble_client/cli/dataset_commands.py +++ b/tools/tcat_ble_client/cli/dataset_commands.py @@ -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(), diff --git a/tools/tcat_ble_client/dataset/dataset.py b/tools/tcat_ble_client/dataset/dataset.py index 384bd2652..6c3d6be7d 100644 --- a/tools/tcat_ble_client/dataset/dataset.py +++ b/tools/tcat_ble_client/dataset/dataset.py @@ -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.') diff --git a/tools/tcat_ble_client/dataset/dataset_entries.py b/tools/tcat_ble_client/dataset/dataset_entries.py index 20d470e36..d68e65c5c 100644 --- a/tools/tcat_ble_client/dataset/dataset_entries.py +++ b/tools/tcat_ble_client/dataset/dataset_entries.py @@ -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}")