wirenboard / wb-hwconf-manager / 2

Committed 12 Nov 2025 10:44AM UTC coverage: 80.567% (+10.4%) from 70.142%

Build # 2

Build Type

Pull #147

github

Specific Base 6f43ff

Committed by

taraant

Commit Message

Add tests

Pull Request Pull Request #147: Features/soft 6124 add audio

Run Details

22 of 78 new or added lines in 3 files covered. (28.21%)

28 existing lines in 1 file now uncovered.

597 of 741 relevant lines covered (80.57%)

1.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.15

/config.py

#!/usr/bin/env python3

import argparse
import glob
import json
import logging
import os
import re
import sys
from pathlib import Path
from typing import List

import hdmi

MODULES_DIR = "/usr/share/wb-hwconf-manager/modules"
CONFIG_PATH = "/etc/wb-hardware.conf"
VENDOR_CONFIG_PATH = "/usr/share/wb-hwconf-manager/vendor-modules.json"


def get_compatible_boards_list() -> List[str]:
    """
    Returns a list of compatible board identifiers from the device tree.

    Returns:
        List[str]: A list of compatible board names (strings).
    """
    root_node = os.readlink("/proc/device-tree")
    with open(root_node + "/compatible", "r", encoding="utf-8") as file:
        return file.read().split("\x00")


def get_board_config_path() -> str:
    """
    Determines the board configuration file path based on the detected hardware.

    Returns:
        str: Path to the appropriate board configuration file.
    """
    boards = [
        ("wirenboard,wirenboard-85xm", "wb85xm"),
        ("wirenboard,wirenboard-85x", "wb85x"),
        ("wirenboard,wirenboard-84x", "wb84x"),
        ("wirenboard,wirenboard-74x", "wb74x"),
        ("wirenboard,wirenboard-731", "wb72x-73x"),
        ("wirenboard,wirenboard-730", "wb730"),
        ("wirenboard,wirenboard-73x", "wb72x-73x"),
        ("wirenboard,wirenboard-72x", "wb72x-73x"),
        ("wirenboard,wirenboard-720", "wb72x-73x"),
        ("contactless,imx6ul-wirenboard670", "wb67"),
        ("contactless,imx6ul-wirenboard61", "wb61"),
        ("contactless,imx6ul-wirenboard60", "wb60"),
    ]
    config_format = "/usr/share/wb-hwconf-manager/boards/{}.conf"
    compatible_boards = get_compatible_boards_list()
    for compatible, conf in boards:
        if compatible in compatible_boards:
            return config_format.format(conf)
    return config_format.format("default")


# board config structure
# {
#     "slots": [
#         {
#             "slot_id": "mod1",
#             "id": "wb67-mod1",
#             "compatible": ["wbe2", "wbe3-reduced"],
#             "name": "Internal slot 1",
#             "module": "",
#             "options": {}
#         },
#         ...,
#         {
#               "slot_id": "extio1",
#               "id": "wb6-extio1",
#               "compatible": ["wb5-extio"],
#               "name": "External I/O module 1",
#               "module": "",
#               "options": {}
#         },
#         ...
#     ]
# }

# config structure
# {
#     "mod1": {
#         "module": "MODULE_NAME1",
#         "options": {}
#     },
#     ...,
#     "extio1": {
#         "module": "MODULE_NAME2",
#         "options": {
#             "param": "value"
#         }
#     },
#     ...
# }


# combined config structure
# {
#     "slots": [
#         {
#             "slot_id": "mod1",
#             "id": "wb67-mod1",
#             "compatible": ["wbe2", "wbe3-reduced"],
#             "name": "Internal slot 1",
#             "module": "MODULE_NAME1",
#             "options": {}
#         },
#         ...,
#         {
#             "slot_id": "extio1",
#             "id": "wb6-extio1",
#             "compatible": ["wb5-extio"],
#             "name": "External I/O module 1",
#             "module": "MODULE_NAME2",
#             "options": {
#                 "param": "value"
#             }
#         },
#         ...
#     ]
# }


def merge_config_and_slots(config: dict, board_slots: dict) -> dict:
    """
    Merges user-defined configuration with board slot definitions.

    Args:
        config (dict): The user-defined configuration.
        board_slots (dict): The board slot definitions.

    Returns:
        dict: Combined configuration with modules and options set.
    """
    merged_config_slots = []
    for slot in board_slots["slots"]:
        slot_config = config.get(slot["slot_id"])
        if slot_config:
            slot["module"] = slot_config.get("module", "")
            slot["options"] = slot_config.get("options", {})
            merged_config_slots.append(slot["slot_id"])
        del slot["slot_id"]
    for slot_id in config.keys():
        if slot_id not in merged_config_slots:
            logging.warning("Slot %s is not supported by board", slot_id)
    return board_slots


def has_unsupported_module(combined_slot: dict, modules_by_id: dict) -> bool:
    """
    Checks whether a module is unsupported in a given slot.

    Args:
        combined_slot (dict): The slot configuration including module.
        modules_by_id (dict): Mapping of module IDs to compatible slots.

    Returns:
        bool: True if the module is unsupported in the slot, False otherwise.
    """
    module = combined_slot.get("module")
    if module:
        return set(combined_slot.get("compatible", [])).isdisjoint(modules_by_id.get(module, set()))
    return False


def remove_unsupported_modules(combined_config: dict, modules: List[dict]) -> None:
    """
    Removes unsupported modules from a combined configuration.

    Args:
        combined_config (dict): The combined configuration object (with slots).
        modules (List[dict]): List of all available module descriptions.
    """
    modules_by_id = {module["id"]: set(module.get("compatible_slots", [])) for module in modules}
    for slot in combined_config["slots"]:
        if has_unsupported_module(slot, modules_by_id):
            logging.warning("Module %s is not supported by slot %s", slot.get("module"), slot.get("id"))
            slot["module"] = ""
            slot["options"] = {}


def make_combined_config(config: dict, board_slots: dict, modules: List[dict]) -> dict:
    """
    Creates a combined configuration by merging slots and removing unsupported modules.

    Args:
        config (dict): User-defined configuration or combined config.
        board_slots (dict): Board slot definitions.
        modules (List[dict]): Available modules with compatibility info.

    Returns:
        dict: Final combined configuration with valid modules.
    """
    # Config has slots property, it is an old combined config format,
    # convert it to normal config format
    if "slots" in config:
        return merge_config_and_slots(extract_config(config, board_slots, modules), board_slots)

    combined_config = merge_config_and_slots(config, board_slots)
    remove_unsupported_modules(combined_config, modules)
    return combined_config


def module_configs_are_different(slot1: dict, slot2: dict) -> bool:
    """
    Compares two slot configurations to determine if they differ.

    Args:
        slot1 (dict): First slot configuration.
        slot2 (dict): Second slot configuration.

    Returns:
        bool: True if configurations differ, False otherwise.
    """

    return slot1.get("module") != slot2.get("module") or slot1.get("options") != slot2.get("options")


def extract_config(combined_config: dict, board_slots: dict, modules: List[dict]) -> dict:
    """
    Extracts a simple hardware config from a combined config structure.

    Args:
        combined_config (dict): Combined slot configuration.
        board_slots (dict): Board slot definitions.
        modules (List[dict]): List of module info with compatibility data.

    Returns:
        dict: Simplified config structure keyed by slot_id.
    """
    config = {}
    id_to_slots_id = {slot["id"]: slot for slot in board_slots["slots"]}
    modules_by_id = {module["id"]: set(module.get("compatible_slots", [])) for module in modules}

    for config_slot in combined_config["slots"]:
        board_slot = id_to_slots_id.get(config_slot["id"])
        if board_slot is None:
            logging.warning("Slot %s is not supported by board", config_slot["id"])
            continue
        slot_id = board_slot.get("slot_id")
        if slot_id is None:
            continue
        if module_configs_are_different(board_slot, config_slot):
            if has_unsupported_module(config_slot, modules_by_id):
                logging.warning(
                    "Module %s is not supported by slot %s", config_slot.get("module"), config_slot.get("id")
                )
            else:
                config[slot_id] = {
                    "module": config_slot.get("module", ""),
                    "options": config_slot.get("options", {}),
                }
    return config


def to_confed(config_path: str, board_slots_path: str, modules_dir: str, vendor_config_path: str) -> dict:
    """
    Converts the current hardware configuration to a format compatible with wb-mqtt-confed.

    Args:
        config_path (str): Path to the hardware configuration file.
        board_slots_path (str): Path to the board slot definitions.
        modules_dir (str): Directory containing module .dtso files.
        vendor_config_path (str): Path to vendor module descriptions.

    Returns:
        dict: Configuration formatted for wb-mqtt-confed.
    """
    with open(config_path, "r", encoding="utf-8") as config_file:
        config = json.load(config_file)
    modules = make_modules_list(modules_dir, vendor_config_path)
    with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:
        board_slots = json.load(board_slots_file)

    config = make_combined_config(config, board_slots, modules)

    # Provide HDMI modes only when HDMI module is present
    if "wbe2-hdmi" in {slot.get("module") for slot in config["slots"]}:
        config["available_hdmi_modes"] = hdmi.get_hdmi_modes()
        slot_with_hdmi = next((slot for slot in config["slots"] if slot.get("module") == "wbe2-hdmi"), None)
        if slot_with_hdmi is not None:
            slot_with_hdmi.setdefault("options", {})
            slot_with_hdmi["options"]["monitor_info"] = hdmi.get_monitor_info()

    config["modules"] = modules
    return config


def from_confed(
    confed_config_str: str, board_slots_path: str, modules_dir: str, vendor_config_path: str
) -> dict:
    """
    Converts a wb-mqtt-confed-style JSON config back to the simplified hardware config.

    Args:
        confed_config_str (str): Confed JSON string from stdin.
        board_slots_path (str): Path to board slot definitions.
        modules_dir (str): Directory containing module .dtso files.
        vendor_config_path (str): Path to vendor module descriptions.

    Returns:
        dict: Simplified configuration.
    """
    confed_config = json.loads(confed_config_str)
    modules = make_modules_list(modules_dir, vendor_config_path)
    with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:
        board_slots = json.load(board_slots_file)
    return extract_config(confed_config, board_slots, modules)


def to_combined_config(
    config_str: str, board_slots_path: str, modules_dir: str, vendor_config_path: str
) -> dict:
    """
    Converts a simplified config JSON string to a full combined configuration.

    Args:
        config_str (str): JSON string of the simplified config.
        board_slots_path (str): Path to board slot definitions.
        modules_dir (str): Directory containing module .dtso files.
        vendor_config_path (str): Path to vendor module descriptions.

    Returns:
        dict: Combined slot-based configuration.
    """
    config = json.loads(config_str)
    modules = make_modules_list(modules_dir, vendor_config_path)
    with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:
        board_slots = json.load(board_slots_file)
    return make_combined_config(config, board_slots, modules)


# Build list of json description of all modules in form
# {
#         "id": "mod-foo",
#         "description": "Foo Module",
#         "compatible_slots": ["bar", "baz"]
# }
# Vendor config looks like
# {
#         "mod-foo":"vendor_description"
# }
def make_modules_list(modules_dir: str, vendor_config_path: str) -> List[dict]:
    """
    Builds a list of available modules from .dtso files and vendor descriptions.

    Args:
        modules_dir (str): Directory with .dtso files describing modules.
        vendor_config_path (str): Path to vendor descriptions JSON.

    Returns:
        List[dict]: List of module definitions with ID, description, and compatible slots.
    """
    modules = []
    compatible_slots_pattern = re.compile(r"compatible-slots\s*=\s*\"(.*)\";")
    description_pattern = re.compile(r"description\s*=\s*\"(.*)\";")
    for dtso_filename in glob.glob(modules_dir + "/*.dtso"):
        with open(dtso_filename, "r", encoding="utf-8") as file:
            module = {"id": Path(dtso_filename).stem}
            for line in file:
                description = description_pattern.search(line)
                if description:
                    module["description"] = description.group(1)
                else:
                    compatible_slots = compatible_slots_pattern.search(line)
                    if compatible_slots:
                        module["compatible_slots"] = [compatible_slots.group(1)]
                if module.get("compatible_slots") and module.get("description"):
                    modules.append(module)
                    break

    modules.sort(key=lambda item: item["id"])
    if not os.path.exists(vendor_config_path):
        return modules

    with open(vendor_config_path, "r", encoding="utf-8") as vendor_config_file:
        vendor_config = json.load(vendor_config_file)
        vendor_modules = []
        wb_modules = []
        for module in modules:
            if module["id"] in vendor_config:
                module["description"] = vendor_config[module["id"]]
                vendor_modules.append(module)
            else:
                wb_modules.append(module)

        return vendor_modules + wb_modules


def main(args=None):
    parser = argparse.ArgumentParser(description="Config generator/updater for wb-hwconf-manager")
    parser.add_argument(
        "-j",
        "--to-confed",
        help="make JSON for wb-mqtt-confed from /etc/wb-hardware.conf",
        action="store_true",
    )
    parser.add_argument(
        "-J",
        "--from-confed",
        help="make /etc/wb-hardware.conf from wb-mqtt-confed output",
        action="store_true",
    )
    parser.add_argument(
        "-o",
        "--to-combined-config",
        help="convert stdin to combined configuration file",
        action="store_true",
    )
    args = parser.parse_args()

    logging.basicConfig(format="%(levelname)s: %(message)s")

    if args.to_confed:
        print(json.dumps(to_confed(CONFIG_PATH, get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH)))
        return

    if args.from_confed:
        print(
            json.dumps(
                from_confed(sys.stdin.read(), get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH),
                indent=4,
            )
        )
        return

    if args.to_combined_config:
        print(
            json.dumps(
                to_combined_config(
                    sys.stdin.read(), get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH
                ),
                indent=4,
            )
        )
        return

    parser.print_usage()


if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2
3	import argparse	2✔
4	import glob	2✔
5	import json	2✔
6	import logging	2✔
7	import os	2✔
8	import re	2✔
9	import sys	2✔
10	from pathlib import Path	2✔
11	from typing import List	2✔
12
13	import hdmi	2✔
14
15	MODULES_DIR = "/usr/share/wb-hwconf-manager/modules"	2✔
16	CONFIG_PATH = "/etc/wb-hardware.conf"	2✔
17	VENDOR_CONFIG_PATH = "/usr/share/wb-hwconf-manager/vendor-modules.json"	2✔
18
19
20	def get_compatible_boards_list() -> List[str]:	2✔
21	"""
22	Returns a list of compatible board identifiers from the device tree.
23
24	Returns:
25	List[str]: A list of compatible board names (strings).
26	"""
27	root_node = os.readlink("/proc/device-tree")	×
UNCOV 28	with open(root_node + "/compatible", "r", encoding="utf-8") as file:	×
UNCOV 29	return file.read().split("\x00")	×
30
31
32	def get_board_config_path() -> str:	2✔
33	"""
34	Determines the board configuration file path based on the detected hardware.
35
36	Returns:
37	str: Path to the appropriate board configuration file.
38	"""
UNCOV 39	boards = [	×
40	("wirenboard,wirenboard-85xm", "wb85xm"),
41	("wirenboard,wirenboard-85x", "wb85x"),
42	("wirenboard,wirenboard-84x", "wb84x"),
43	("wirenboard,wirenboard-74x", "wb74x"),
44	("wirenboard,wirenboard-731", "wb72x-73x"),
45	("wirenboard,wirenboard-730", "wb730"),
46	("wirenboard,wirenboard-73x", "wb72x-73x"),
47	("wirenboard,wirenboard-72x", "wb72x-73x"),
48	("wirenboard,wirenboard-720", "wb72x-73x"),
49	("contactless,imx6ul-wirenboard670", "wb67"),
50	("contactless,imx6ul-wirenboard61", "wb61"),
51	("contactless,imx6ul-wirenboard60", "wb60"),
52	]
53	config_format = "/usr/share/wb-hwconf-manager/boards/{}.conf"	×
54	compatible_boards = get_compatible_boards_list()	×
55	for compatible, conf in boards:	×
56	if compatible in compatible_boards:	×
UNCOV 57	return config_format.format(conf)	×
UNCOV 58	return config_format.format("default")	×
59
60
61	# board config structure
62	# {
63	# "slots": [
64	# {
65	# "slot_id": "mod1",
66	# "id": "wb67-mod1",
67	# "compatible": ["wbe2", "wbe3-reduced"],
68	# "name": "Internal slot 1",
69	# "module": "",
70	# "options": {}
71	# },
72	# ...,
73	# {
74	# "slot_id": "extio1",
75	# "id": "wb6-extio1",
76	# "compatible": ["wb5-extio"],
77	# "name": "External I/O module 1",
78	# "module": "",
79	# "options": {}
80	# },
81	# ...
82	# ]
83	# }
84
85	# config structure
86	# {
87	# "mod1": {
88	# "module": "MODULE_NAME1",
89	# "options": {}
90	# },
91	# ...,
92	# "extio1": {
93	# "module": "MODULE_NAME2",
94	# "options": {
95	# "param": "value"
96	# }
97	# },
98	# ...
99	# }
100
101
102	# combined config structure
103	# {
104	# "slots": [
105	# {
106	# "slot_id": "mod1",
107	# "id": "wb67-mod1",
108	# "compatible": ["wbe2", "wbe3-reduced"],
109	# "name": "Internal slot 1",
110	# "module": "MODULE_NAME1",
111	# "options": {}
112	# },
113	# ...,
114	# {
115	# "slot_id": "extio1",
116	# "id": "wb6-extio1",
117	# "compatible": ["wb5-extio"],
118	# "name": "External I/O module 1",
119	# "module": "MODULE_NAME2",
120	# "options": {
121	# "param": "value"
122	# }
123	# },
124	# ...
125	# ]
126	# }
127
128
129	def merge_config_and_slots(config: dict, board_slots: dict) -> dict:	2✔
130	"""
131	Merges user-defined configuration with board slot definitions.
132
133	Args:
134	config (dict): The user-defined configuration.
135	board_slots (dict): The board slot definitions.
136
137	Returns:
138	dict: Combined configuration with modules and options set.
139	"""
140	merged_config_slots = []	2✔
141	for slot in board_slots["slots"]:	2✔
142	slot_config = config.get(slot["slot_id"])	2✔
143	if slot_config:	2✔
144	slot["module"] = slot_config.get("module", "")	2✔
145	slot["options"] = slot_config.get("options", {})	2✔
146	merged_config_slots.append(slot["slot_id"])	2✔
147	del slot["slot_id"]	2✔
148	for slot_id in config.keys():	2✔
149	if slot_id not in merged_config_slots:	2✔
150	logging.warning("Slot %s is not supported by board", slot_id)	2✔
151	return board_slots	2✔
152
153
154	def has_unsupported_module(combined_slot: dict, modules_by_id: dict) -> bool:	2✔
155	"""
156	Checks whether a module is unsupported in a given slot.
157
158	Args:
159	combined_slot (dict): The slot configuration including module.
160	modules_by_id (dict): Mapping of module IDs to compatible slots.
161
162	Returns:
163	bool: True if the module is unsupported in the slot, False otherwise.
164	"""
165	module = combined_slot.get("module")	2✔
166	if module:	2✔
167	return set(combined_slot.get("compatible", [])).isdisjoint(modules_by_id.get(module, set()))	2✔
168	return False	2✔
169
170
171	def remove_unsupported_modules(combined_config: dict, modules: List[dict]) -> None:	2✔
172	"""
173	Removes unsupported modules from a combined configuration.
174
175	Args:
176	combined_config (dict): The combined configuration object (with slots).
177	modules (List[dict]): List of all available module descriptions.
178	"""
179	modules_by_id = {module["id"]: set(module.get("compatible_slots", [])) for module in modules}	2✔
180	for slot in combined_config["slots"]:	2✔
181	if has_unsupported_module(slot, modules_by_id):	2✔
182	logging.warning("Module %s is not supported by slot %s", slot.get("module"), slot.get("id"))	2✔
183	slot["module"] = ""	2✔
184	slot["options"] = {}	2✔
185
186
187	def make_combined_config(config: dict, board_slots: dict, modules: List[dict]) -> dict:	2✔
188	"""
189	Creates a combined configuration by merging slots and removing unsupported modules.
190
191	Args:
192	config (dict): User-defined configuration or combined config.
193	board_slots (dict): Board slot definitions.
194	modules (List[dict]): Available modules with compatibility info.
195
196	Returns:
197	dict: Final combined configuration with valid modules.
198	"""
199	# Config has slots property, it is an old combined config format,
200	# convert it to normal config format
201	if "slots" in config:	2✔
202	return merge_config_and_slots(extract_config(config, board_slots, modules), board_slots)	2✔
203
204	combined_config = merge_config_and_slots(config, board_slots)	2✔
205	remove_unsupported_modules(combined_config, modules)	2✔
206	return combined_config	2✔
207
208
209	def module_configs_are_different(slot1: dict, slot2: dict) -> bool:	2✔
210	"""
211	Compares two slot configurations to determine if they differ.
212
213	Args:
214	slot1 (dict): First slot configuration.
215	slot2 (dict): Second slot configuration.
216
217	Returns:
218	bool: True if configurations differ, False otherwise.
219	"""
220
221	return slot1.get("module") != slot2.get("module") or slot1.get("options") != slot2.get("options")	2✔
222
223
224	def extract_config(combined_config: dict, board_slots: dict, modules: List[dict]) -> dict:	2✔
225	"""
226	Extracts a simple hardware config from a combined config structure.
227
228	Args:
229	combined_config (dict): Combined slot configuration.
230	board_slots (dict): Board slot definitions.
231	modules (List[dict]): List of module info with compatibility data.
232
233	Returns:
234	dict: Simplified config structure keyed by slot_id.
235	"""
236	config = {}	2✔
237	id_to_slots_id = {slot["id"]: slot for slot in board_slots["slots"]}	2✔
238	modules_by_id = {module["id"]: set(module.get("compatible_slots", [])) for module in modules}	2✔
239
240	for config_slot in combined_config["slots"]:	2✔
241	board_slot = id_to_slots_id.get(config_slot["id"])	2✔
242	if board_slot is None:	2✔
243	logging.warning("Slot %s is not supported by board", config_slot["id"])	2✔
244	continue	2✔
245	slot_id = board_slot.get("slot_id")	2✔
246	if slot_id is None:	2✔
UNCOV 247	continue	×
248	if module_configs_are_different(board_slot, config_slot):	2✔
249	if has_unsupported_module(config_slot, modules_by_id):	2✔
250	logging.warning(	2✔
251	"Module %s is not supported by slot %s", config_slot.get("module"), config_slot.get("id")
252	)
253	else:
254	config[slot_id] = {	2✔
255	"module": config_slot.get("module", ""),
256	"options": config_slot.get("options", {}),
257	}
258	return config	2✔
259
260
261	def to_confed(config_path: str, board_slots_path: str, modules_dir: str, vendor_config_path: str) -> dict:	2✔
262	"""
263	Converts the current hardware configuration to a format compatible with wb-mqtt-confed.
264
265	Args:
266	config_path (str): Path to the hardware configuration file.
267	board_slots_path (str): Path to the board slot definitions.
268	modules_dir (str): Directory containing module .dtso files.
269	vendor_config_path (str): Path to vendor module descriptions.
270
271	Returns:
272	dict: Configuration formatted for wb-mqtt-confed.
273	"""
274	with open(config_path, "r", encoding="utf-8") as config_file:	2✔
275	config = json.load(config_file)	2✔
276	modules = make_modules_list(modules_dir, vendor_config_path)	2✔
277	with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:	2✔
278	board_slots = json.load(board_slots_file)	2✔
279
280	config = make_combined_config(config, board_slots, modules)	2✔
281
282	# Provide HDMI modes only when HDMI module is present
283	if "wbe2-hdmi" in {slot.get("module") for slot in config["slots"]}:	2✔
284	config["available_hdmi_modes"] = hdmi.get_hdmi_modes()	2✔
285	slot_with_hdmi = next((slot for slot in config["slots"] if slot.get("module") == "wbe2-hdmi"), None)	2✔
286	if slot_with_hdmi is not None:	2✔
287	slot_with_hdmi.setdefault("options", {})	2✔
288	slot_with_hdmi["options"]["monitor_info"] = hdmi.get_monitor_info()	2✔
289
290	config["modules"] = modules	2✔
291	return config	2✔
292
293
294	def from_confed(	2✔
295	confed_config_str: str, board_slots_path: str, modules_dir: str, vendor_config_path: str
296	) -> dict:
297	"""
298	Converts a wb-mqtt-confed-style JSON config back to the simplified hardware config.
299
300	Args:
301	confed_config_str (str): Confed JSON string from stdin.
302	board_slots_path (str): Path to board slot definitions.
303	modules_dir (str): Directory containing module .dtso files.
304	vendor_config_path (str): Path to vendor module descriptions.
305
306	Returns:
307	dict: Simplified configuration.
308	"""
309	confed_config = json.loads(confed_config_str)	2✔
310	modules = make_modules_list(modules_dir, vendor_config_path)	2✔
311	with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:	2✔
312	board_slots = json.load(board_slots_file)	2✔
313	return extract_config(confed_config, board_slots, modules)	2✔
314
315
316	def to_combined_config(	2✔
317	config_str: str, board_slots_path: str, modules_dir: str, vendor_config_path: str
318	) -> dict:
319	"""
320	Converts a simplified config JSON string to a full combined configuration.
321
322	Args:
323	config_str (str): JSON string of the simplified config.
324	board_slots_path (str): Path to board slot definitions.
325	modules_dir (str): Directory containing module .dtso files.
326	vendor_config_path (str): Path to vendor module descriptions.
327
328	Returns:
329	dict: Combined slot-based configuration.
330	"""
UNCOV 331	config = json.loads(config_str)	×
UNCOV 332	modules = make_modules_list(modules_dir, vendor_config_path)	×
UNCOV 333	with open(board_slots_path, "r", encoding="utf-8") as board_slots_file:	×
UNCOV 334	board_slots = json.load(board_slots_file)	×
UNCOV 335	return make_combined_config(config, board_slots, modules)	×
336
337
338	# Build list of json description of all modules in form
339	# {
340	# "id": "mod-foo",
341	# "description": "Foo Module",
342	# "compatible_slots": ["bar", "baz"]
343	# }
344	# Vendor config looks like
345	# {
346	# "mod-foo":"vendor_description"
347	# }
348	def make_modules_list(modules_dir: str, vendor_config_path: str) -> List[dict]:	2✔
349	"""
350	Builds a list of available modules from .dtso files and vendor descriptions.
351
352	Args:
353	modules_dir (str): Directory with .dtso files describing modules.
354	vendor_config_path (str): Path to vendor descriptions JSON.
355
356	Returns:
357	List[dict]: List of module definitions with ID, description, and compatible slots.
358	"""
359	modules = []	2✔
360	compatible_slots_pattern = re.compile(r"compatible-slots\s=\s\"(.*)\";")	2✔
361	description_pattern = re.compile(r"description\s=\s\"(.*)\";")	2✔
362	for dtso_filename in glob.glob(modules_dir + "/*.dtso"):	2✔
363	with open(dtso_filename, "r", encoding="utf-8") as file:	2✔
364	module = {"id": Path(dtso_filename).stem}	2✔
365	for line in file:	2✔
366	description = description_pattern.search(line)	2✔
367	if description:	2✔
368	module["description"] = description.group(1)	2✔
369	else:
370	compatible_slots = compatible_slots_pattern.search(line)	2✔
371	if compatible_slots:	2✔
372	module["compatible_slots"] = [compatible_slots.group(1)]	2✔
373	if module.get("compatible_slots") and module.get("description"):	2✔
374	modules.append(module)	2✔
375	break	2✔
376
377	modules.sort(key=lambda item: item["id"])	2✔
378	if not os.path.exists(vendor_config_path):	2✔
379	return modules	2✔
380
381	with open(vendor_config_path, "r", encoding="utf-8") as vendor_config_file:	2✔
382	vendor_config = json.load(vendor_config_file)	2✔
383	vendor_modules = []	2✔
384	wb_modules = []	2✔
385	for module in modules:	2✔
386	if module["id"] in vendor_config:	2✔
387	module["description"] = vendor_config[module["id"]]	2✔
388	vendor_modules.append(module)	2✔
389	else:
390	wb_modules.append(module)	2✔
391
392	return vendor_modules + wb_modules	2✔
393
394
395	def main(args=None):	2✔
UNCOV 396	parser = argparse.ArgumentParser(description="Config generator/updater for wb-hwconf-manager")	×
UNCOV 397	parser.add_argument(	×
398	"-j",
399	"--to-confed",
400	help="make JSON for wb-mqtt-confed from /etc/wb-hardware.conf",
401	action="store_true",
402	)
UNCOV 403	parser.add_argument(	×
404	"-J",
405	"--from-confed",
406	help="make /etc/wb-hardware.conf from wb-mqtt-confed output",
407	action="store_true",
408	)
UNCOV 409	parser.add_argument(	×
410	"-o",
411	"--to-combined-config",
412	help="convert stdin to combined configuration file",
413	action="store_true",
414	)
UNCOV 415	args = parser.parse_args()	×
416
UNCOV 417	logging.basicConfig(format="%(levelname)s: %(message)s")	×
418
UNCOV 419	if args.to_confed:	×
UNCOV 420	print(json.dumps(to_confed(CONFIG_PATH, get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH)))	×
UNCOV 421	return	×
422
UNCOV 423	if args.from_confed:	×
UNCOV 424	print(	×
425	json.dumps(
426	from_confed(sys.stdin.read(), get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH),
427	indent=4,
428	)
429	)
UNCOV 430	return	×
431
UNCOV 432	if args.to_combined_config:	×
UNCOV 433	print(	×
434	json.dumps(
435	to_combined_config(
436	sys.stdin.read(), get_board_config_path(), MODULES_DIR, VENDOR_CONFIG_PATH
437	),
438	indent=4,
439	)
440	)
UNCOV 441	return	×
442
UNCOV 443	parser.print_usage()	×
444
445
446	if __name__ == "__main__":	2✔
UNCOV 447	main()	×

wirenboard / wb-hwconf-manager / 2

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous