Merge pull request #392 from tzeikob/bluetoothctl-parser

Add parser for the bluetoothctl utility

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 <ctrl>`
+- `bluetoothctl devices`
+- `bluetoothctl info <dev>`
+
+Usage (cli):
+
+    $ bluetoothctl info <dev> | jc --bluetoothctl 
+or
+
+    $ jc bluetoothctl info <dev>
+
+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<address>([0-9A-F]{2}:){5}[0-9A-F]{2}) (?P<name>.+)"
+
+_controller_line_pattern = (
+    r"(\s*Name:\s*(?P<name>.+)"
+    + r"|\s*Alias:\s*(?P<alias>.+)"
+    + r"|\s*Class:\s*(?P<class>.+)"
+    + r"|\s*Powered:\s*(?P<powered>.+)"
+    + r"|\s*Discoverable:\s*(?P<discoverable>.+)"
+    + r"|\s*DiscoverableTimeout:\s*(?P<discoverable_timeout>.+)"
+    + r"|\s*Pairable:\s*(?P<pairable>.+)"
+    + r"|\s*Modalias:\s*(?P<modalias>.+)"
+    + r"|\s*Discovering:\s*(?P<discovering>.+)"
+    + r"|\s*UUID:\s*(?P<uuid>.+))"
+)
+
+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<address>([0-9A-F]{2}:){5}[0-9A-F]{2}) (?P<name>.+)"
+
+_device_line_pattern = (
+    r"(\s*Name:\s*(?P<name>.+)"
+    + r"|\s*Alias:\s*(?P<alias>.+)"
+    + r"|\s*Class:\s*(?P<class>.+)"
+    + r"|\s*Icon:\s*(?P<icon>.+)"
+    + r"|\s*Paired:\s*(?P<paired>.+)"
+    + r"|\s*Bonded:\s*(?P<bonded>.+)"
+    + r"|\s*Trusted:\s*(?P<trusted>.+)"
+    + r"|\s*Blocked:\s*(?P<blocked>.+)"
+    + r"|\s*Connected:\s*(?P<connected>.+)"
+    + r"|\s*LegacyPairing:\s*(?P<legacy_pairing>.+)"
+    + r"|\s*Modalias:\s*(?P<modalias>.+)"
+    + r"|\s*RSSI:\s*(?P<rssi>.+)"
+    + r"|\s*TxPower:\s*(?P<txpower>.+)"
+    + r"|\s*UUID:\s*(?P<uuid>.+))"
+)
+
+
+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

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

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

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()