diff --git a/jc/parsers/bluetoothctl.py b/jc/parsers/bluetoothctl.py new file mode 100644 index 00000000..07b4c6d3 --- /dev/null +++ b/jc/parsers/bluetoothctl.py @@ -0,0 +1,374 @@ +"""jc - JSON Convert `bluetoothctl` command output parser + +Supports the following `bluetoothctl` subcommands: +- `bluetoothctl list` +- `bluetoothctl show` +- `bluetoothctl show ` +- `bluetoothctl devices` +- `bluetoothctl info ` + +Usage (cli): + + $ bluetoothctl info | jc --bluetoothctl +or + + $ jc bluetoothctl info + +Usage (module): + + import jc + result = jc.parse('bluetoothctl', bluetoothctl_command_output) + +Schema: + +Becasuse bluetoothctl is handling two main entities, controllers and devices, +the schema is shared between them. The most of the fields are common between +a controller and a device but there might be fields corresponding to one entity. + + Controller: + [ + { + "name": string, + "is_default": boolean, + "is_public": boolean, + "address": string, + "alias": string, + "class": string, + "powered": string, + "discoverable": string, + "discoverable_timeout": string, + "pairable": string, + "modalias": string, + "discovering": string, + "uuids": array + } + ] + + Device: + [ + { + "name": string, + "is_public": boolean, + "address": string, + "alias": string, + "class": string, + "icon": string, + "paired": string, + "bonded": string, + "trusted": string, + "blocked": string, + "connected": string, + "legacy_pairing": string, + "rssi": int, + "txpower": int, + "uuids": array + } + ] + +Examples: + + $ bluetoothctl info EB:06:EF:62:B3:19 | jc --bluetoothctl -p + [ + { + "address": "22:06:33:62:B3:19", + "is_public": true, + "name": "TaoTronics TT-BH336", + "alias": "TaoTronics TT-BH336", + "class": "0x00240455", + "icon": "audio-headset", + "paired": "no", + "bonded": "no", + "trusted": "no", + "blocked": "no", + "connected": "no", + "legacy_pairing": "no", + "uuids": [ + "Advanced Audio Distribu.. (0000120d-0000-1000-8000-00805f9b34fb)", + "Audio Sink (0000130b-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control (0000140e-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control Cont.. (0000150f-0000-1000-8000-00805f9b34fb)", + "Handsfree (0000161e-0000-1000-8000-00805f9b34fb)", + "Headset (00001708-0000-1000-8000-00805f9b34fb)", + "Headset HS (00001831-0000-1000-8000-00805f9b34fb)" + ], + "rssi": -52, + "txpower": 4 + } + ] +""" +import re +from typing import List, Dict, Union +from jc.jc_types import JSONDictType +import jc.utils + + +class info(): + """Provides parser metadata (version, author, etc.)""" + version = '1.0' + description = '`bluetoothctl` command parser' + author = 'Jake Ob' + author_email = 'iakopap at gmail.com' + compatible = ["linux", "darwin", "cygwin", "aix", "freebsd"] + magic_commands = ["bluetoothctl"] + tags = ['command'] + + +__version__ = info.version + +try: + from typing import TypedDict + + Controller = TypedDict( + "Controller", + { + "name": str, + "is_default": bool, + "is_public": bool, + "address": str, + "alias": str, + "class": str, + "powered": str, + "discoverable": str, + "discoverable_timeout": str, + "pairable": str, + "modalias": str, + "discovering": str, + "uuids": List[str], + }, + ) + Device = TypedDict( + "Device", + { + "name": str, + "is_public": bool, + "address": str, + "alias": str, + "class": str, + "icon": str, + "paired": str, + "bonded": str, + "trusted": str, + "blocked": str, + "connected": str, + "legacy_pairing": str, + "rssi": int, + "txpower": int, + "uuids": List[str], + }, + ) +except ImportError: + Controller = Dict[str, Union[str, bool, List[str]]] + Device = Dict[str, Union[str, bool, int, List[str]]] + + +_controller_head_pattern = r"Controller (?P
([0-9A-F]{2}:){5}[0-9A-F]{2}) (?P.+)" + +_controller_line_pattern = ( + r"(\s*Name:\s*(?P.+)" + + r"|\s*Alias:\s*(?P.+)" + + r"|\s*Class:\s*(?P.+)" + + r"|\s*Powered:\s*(?P.+)" + + r"|\s*Discoverable:\s*(?P.+)" + + r"|\s*DiscoverableTimeout:\s*(?P.+)" + + r"|\s*Pairable:\s*(?P.+)" + + r"|\s*Modalias:\s*(?P.+)" + + r"|\s*Discovering:\s*(?P.+)" + + r"|\s*UUID:\s*(?P.+))" +) + +def _parse_controller(next_lines: List[str]) -> Controller: + next_line = next_lines.pop() + result = re.match(_controller_head_pattern, next_line) + + if not result: + next_lines.append(next_line) + return None + + matches = result.groupdict() + + name = matches["name"] + + if name.endswith("not available"): + return None + + controller = Controller = { + "address": matches["address"], + } + + if name.endswith("[default]"): + controller["is_default"] = True + name = name.replace("[default]", "") + + if name.endswith("(public)"): + controller["is_public"] = True + name = name.replace("(public)", "") + + controller["name"] = name.strip() + + while next_lines: + next_line = next_lines.pop() + result = re.match(_controller_line_pattern, next_line) + + if not result: + next_lines.append(next_line) + return controller + + matches = result.groupdict() + + if matches["name"]: + controller["name"] = matches["name"] + elif matches["alias"]: + controller["alias"] = matches["alias"] + elif matches["class"]: + controller["class"] = matches["class"] + elif matches["powered"]: + controller["powered"] = matches["powered"] + elif matches["discoverable"]: + controller["discoverable"] = matches["discoverable"] + elif matches["discoverable_timeout"]: + controller["discoverable_timeout"] = matches["discoverable_timeout"] + elif matches["pairable"]: + controller["pairable"] = matches["pairable"] + elif matches["modalias"]: + controller["modalias"] = matches["modalias"] + elif matches["discovering"]: + controller["discovering"] = matches["discovering"] + elif matches["uuid"]: + if not "uuids" in controller: + controller["uuids"] = [] + controller["uuids"].append(matches["uuid"]) + + return controller + +_device_head_pattern = r"Device (?P
([0-9A-F]{2}:){5}[0-9A-F]{2}) (?P.+)" + +_device_line_pattern = ( + r"(\s*Name:\s*(?P.+)" + + r"|\s*Alias:\s*(?P.+)" + + r"|\s*Class:\s*(?P.+)" + + r"|\s*Icon:\s*(?P.+)" + + r"|\s*Paired:\s*(?P.+)" + + r"|\s*Bonded:\s*(?P.+)" + + r"|\s*Trusted:\s*(?P.+)" + + r"|\s*Blocked:\s*(?P.+)" + + r"|\s*Connected:\s*(?P.+)" + + r"|\s*LegacyPairing:\s*(?P.+)" + + r"|\s*Modalias:\s*(?P.+)" + + r"|\s*RSSI:\s*(?P.+)" + + r"|\s*TxPower:\s*(?P.+)" + + r"|\s*UUID:\s*(?P.+))" +) + + +def _parse_device(next_lines: List[str]) -> Device: + next_line = next_lines.pop() + result = re.match(_device_head_pattern, next_line) + + if not result: + next_lines.append(next_line) + return None + + matches = result.groupdict() + + name = matches["name"] + + if name.endswith("not available"): + return None + + device = Device = { + "address": matches["address"], + } + + if name.endswith("(public)"): + device["is_public"] = True + name = name.replace("(public)", "") + + device["name"] = name.strip() + + while next_lines: + next_line = next_lines.pop() + result = re.match(_device_line_pattern, next_line) + + if not result: + next_lines.append(next_line) + return device + + matches = result.groupdict() + + if matches["name"]: + device["name"] = matches["name"] + elif matches["alias"]: + device["alias"] = matches["alias"] + elif matches["class"]: + device["class"] = matches["class"] + elif matches["icon"]: + device["icon"] = matches["icon"] + elif matches["paired"]: + device["paired"] = matches["paired"] + elif matches["bonded"]: + device["bonded"] = matches["bonded"] + elif matches["trusted"]: + device["trusted"] = matches["trusted"] + elif matches["blocked"]: + device["blocked"] = matches["blocked"] + elif matches["connected"]: + device["connected"] = matches["connected"] + elif matches["legacy_pairing"]: + device["legacy_pairing"] = matches["legacy_pairing"] + elif matches["modalias"]: + device["modalias"] = matches["modalias"] + elif matches["rssi"]: + rssi = matches["rssi"] + try: + device["rssi"] = int(rssi) + except ValueError and not quiet: + jc.utils.warning_message([f"{next_line} : rssi - {rssi} is not int-able"]) + elif matches["txpower"]: + txpower = matches["txpower"] + try: + device["txpower"] = int(txpower) + except ValueError and not quiet: + jc.utils.warning_message([f"{next_line} : txpower - {txpower} is not int-able"]) + elif matches["uuid"]: + if not "uuids" in device: + device["uuids"] = [] + device["uuids"].append(matches["uuid"]) + + return device + +def parse(data: str, raw: bool = False, quiet: bool = False) -> List[JSONDictType]: + """ + Main text parsing function + + Parameters: + + data: (string) text data to parse + raw: (boolean) unprocessed output if True + quiet: (boolean) suppress warning messages if True + + Returns: + + List of Dictionaries. Raw or processed structured data. + """ + result: List[TypeDict] = [] + + if jc.utils.has_data(data): + jc.utils.compatibility(__name__, info.compatible, quiet) + jc.utils.input_type_check(data) + + linedata = data.splitlines() + linedata.reverse() + + while linedata: + element = None + if data.startswith("Controller"): + element = _parse_controller(linedata) + elif data.startswith("Device"): + element = _parse_device(linedata) + + if element: + result.append(element) + else: + break + + return result diff --git a/tests/fixtures/generic/bluetoothctl_controller.out b/tests/fixtures/generic/bluetoothctl_controller.out new file mode 100644 index 00000000..78d58bc5 --- /dev/null +++ b/tests/fixtures/generic/bluetoothctl_controller.out @@ -0,0 +1,17 @@ +Controller CC:BB:AF:27:6A:E4 (public) + Name: arch + Alias: arch + Class: 0x006c010c + Powered: yes + Discoverable: no + DiscoverableTimeout: 0x000000b4 + Pairable: no + UUID: Handsfree (0000111e-0000-1000-8000-00805f9b34fb) + UUID: Audio Source (0000110a-0000-1000-8000-00805f9b34fb) + UUID: Audio Sink (0000110b-0000-1000-8000-00805f9b34fb) + UUID: PnP Information (00001200-0000-1000-8000-00805f9b34fb) + UUID: A/V Remote Control Target (0000110c-0000-1000-8000-00805f9b34fb) + UUID: A/V Remote Control (0000110e-0000-1000-8000-00805f9b34fb) + UUID: Handsfree Audio Gateway (0000111f-0000-1000-8000-00805f9b34fb) + Modalias: usb:v1D6Bp0246d0542 + Discovering: no diff --git a/tests/fixtures/generic/bluetoothctl_device.out b/tests/fixtures/generic/bluetoothctl_device.out new file mode 100644 index 00000000..5aafcd66 --- /dev/null +++ b/tests/fixtures/generic/bluetoothctl_device.out @@ -0,0 +1,20 @@ +Device EB:06:EF:62:B3:19 (public) + Name: TaoTronics TT-BH026 + Alias: TaoTronics TT-BH026 + Class: 0x00240404 + Icon: audio-headset + Paired: no + Bonded: no + Trusted: no + Blocked: no + Connected: no + LegacyPairing: no + UUID: Advanced Audio Distribu.. (0000110d-0000-1000-8000-00805f9b34fb) + UUID: Audio Sink (0000110b-0000-1000-8000-00805f9b34fb) + UUID: A/V Remote Control (0000110e-0000-1000-8000-00805f9b34fb) + UUID: A/V Remote Control Cont.. (0000110f-0000-1000-8000-00805f9b34fb) + UUID: Handsfree (0000111e-0000-1000-8000-00805f9b34fb) + UUID: Headset (00001108-0000-1000-8000-00805f9b34fb) + UUID: Headset HS (00001131-0000-1000-8000-00805f9b34fb) + RSSI: -52 + TxPower: 4 \ No newline at end of file diff --git a/tests/test_bluetoothctl.py b/tests/test_bluetoothctl.py new file mode 100644 index 00000000..2cf1d09f --- /dev/null +++ b/tests/test_bluetoothctl.py @@ -0,0 +1,218 @@ +import json +import re +import unittest + +from jc.parsers.bluetoothctl import ( + _controller_head_pattern, + _controller_line_pattern, + _device_head_pattern, + _device_line_pattern, + _parse_controller, + _parse_device, + Controller, + Device, + parse, +) + + +class BluetoothctlTests(unittest.TestCase): + def test_bluetoothctl_nodata(self): + """ + Test 'bluetoothctl' with no data + """ + + output='' + self.assertEqual(parse(output, quiet=True), []) + + def test_bluetoothctl_invalid_call(self): + """ + Test 'bluetoothctl' with output from invalid call + """ + + output='Invalid command in menu main: foo' + self.assertEqual(parse(output, quiet=True), []) + + def test_bluetoothctl_with_invalid_args(self): + """ + Test 'bluetoothctl' with output from invalid arguments + """ + + output='Too many arguments: 2 > 1' + self.assertEqual(parse(output, quiet=True), []) + + def test_bluetoothctl_no_controller(self): + """ + Test 'bluetoothctl' with no controller + """ + + output='No default controller available' + self.assertEqual(parse(output, quiet=True), []) + + + def test_bluetoothctl_no_controller_found(self): + """ + Test 'bluetoothctl' with no controller found + """ + + output='Controller EB:06:EF:62:B3:33 not available' + self.assertEqual(parse(output, quiet=True), []) + + def test_bluetoothctl_no_device_found(self): + """ + Test 'bluetoothctl' with no device found + """ + + output='Device EB:06:EF:62:B3:33 not available' + self.assertEqual(parse(output, quiet=True), []) + + def test_bluetoothctl_controller(self): + """ + Test 'bluetoothctl' with controller + """ + + with open("tests/fixtures/generic/bluetoothctl_controller.out", "r") as f: + output = f.read() + + actual = parse(output, quiet=True) + + self.assertIsNotNone(actual) + self.assertIsNotNone(actual[0]) + + expected = { + "address": "CC:BB:AF:27:6A:E4", + "is_public": True, + "name": "arch", + "alias": "arch", + "class": "0x006c010c", + "powered": "yes", + "discoverable": "no", + "discoverable_timeout": "0x000000b4", + "pairable": "no", + "uuids": [ + "Handsfree (0000111e-0000-1000-8000-00805f9b34fb)", + "Audio Source (0000110a-0000-1000-8000-00805f9b34fb)", + "Audio Sink (0000110b-0000-1000-8000-00805f9b34fb)", + "PnP Information (00001200-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control Target (0000110c-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control (0000110e-0000-1000-8000-00805f9b34fb)", + "Handsfree Audio Gateway (0000111f-0000-1000-8000-00805f9b34fb)" + ], + "modalias": "usb:v1D6Bp0246d0542", + "discovering": "no" + } + + if actual: + for k, v in expected.items(): + self.assertEqual(v, actual[0][k], f"Controller regex failed on {k}") + + def test_bluetoothctl_controllers(self): + """ + Test 'bluetoothctl' with controllers + """ + + output='Controller CC:52:AF:A4:6A:E4 arch [default]\n' + output+='Controller CC:53:AF:17:6A:34 logi' + + actual = parse(output, quiet=True) + + self.assertIsNotNone(actual) + self.assertIsNotNone(actual[0]) + self.assertIsNotNone(actual[1]) + + expected = [ + { + "address": "CC:52:AF:A4:6A:E4", + "is_default": True, + "name": "arch" + }, + { + "address": "CC:53:AF:17:6A:34", + "name": "logi" + }, + ] + + if actual: + for k, v in expected[0].items(): + self.assertEqual(v, actual[0][k], f"Controller regex failed on {k}") + + for k, v in expected[1].items(): + self.assertEqual(v, actual[1][k], f"Controller regex failed on {k}") + + def test_bluetoothctl_device(self): + """ + Test 'bluetoothctl' with device + """ + + with open("tests/fixtures/generic/bluetoothctl_device.out", "r") as f: + output = f.read() + + actual = parse(output, quiet=True) + + self.assertIsNotNone(actual) + self.assertIsNotNone(actual[0]) + + expected = { + "address": "EB:06:EF:62:B3:19", + "is_public": True, + "name": "TaoTronics TT-BH026", + "alias": "TaoTronics TT-BH026", + "class": "0x00240404", + "icon": "audio-headset", + "paired": "no", + "bonded": "no", + "trusted": "no", + "blocked": "no", + "connected": "no", + "legacy_pairing": "no", + "uuids": [ + "Advanced Audio Distribu.. (0000110d-0000-1000-8000-00805f9b34fb)", + "Audio Sink (0000110b-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control (0000110e-0000-1000-8000-00805f9b34fb)", + "A/V Remote Control Cont.. (0000110f-0000-1000-8000-00805f9b34fb)", + "Handsfree (0000111e-0000-1000-8000-00805f9b34fb)", + "Headset (00001108-0000-1000-8000-00805f9b34fb)", + "Headset HS (00001131-0000-1000-8000-00805f9b34fb)" + ], + "rssi": -52, + "txpower": 4 + } + + if actual: + for k, v in expected.items(): + self.assertEqual(v, actual[0][k], f"Device regex failed on {k}") + + def test_bluetoothctl_devices(self): + """ + Test 'bluetoothctl' with devices + """ + + output='Device EB:06:EF:62:13:19 TaoTronics TT-BH026\n' + output+='Device AC:1F:EA:F8:AA:A1 wacom' + + actual = parse(output, quiet=True) + + self.assertIsNotNone(actual) + self.assertIsNotNone(actual[0]) + self.assertIsNotNone(actual[1]) + + expected = [ + { + "address": "EB:06:EF:62:13:19", + "name": "TaoTronics TT-BH026" + }, + { + "address": "AC:1F:EA:F8:AA:A1", + "name": "wacom" + } + ] + + if actual: + for k, v in expected[0].items(): + self.assertEqual(v, actual[0][k], f"Device regex failed on {k}") + + for k, v in expected[1].items(): + self.assertEqual(v, actual[1][k], f"Device regex failed on {k}") + + +if __name__ == '__main__': + unittest.main()