Format code with black.

.github/workflows/ci.yml

@@ -44,3 +44,5 @@ jobs:
           pip install -r lint_requirements.txt
       - name: Run Flake8
         run: flake8 .
+      - name: Run black
+        run: black --check .

README.rst

@@ -1,4 +1,4 @@
-|Build Status| |PyPi Version| |Issue Count| |Coverage Status|
+|Build Status| |PyPi Version| |Issue Count| |Coverage Status| |Black|
 
 ``pyCEC``
 =========
@@ -36,6 +36,9 @@ HDMI port on client's machine. Just use ``TcpAdapter`` instead of
 
 You can also connect to ``9526`` by `NetCat <https://www.wikiwand.com/en/Netcat>`_ and send CEC commands directly.
 
+
+.. |Black| image:: https://img.shields.io/badge/code%20style-black-000000.svg
+   :target: https://github.com/psf/black
 .. |PyPi Version| image:: https://img.shields.io/pypi/v/pyCEC
    :target: https://pypi.org/project/pyCEC/
 .. |Build Status| image:: https://github.com/konikvranik/pyCEC/workflows/Tests/badge.svg

lint_requirements.txt

@@ -1,3 +1,4 @@
-flake8==3.8.3
+black==20.8b1
 flake8-bugbear==20.1.4
+flake8==3.8.4
 pep8-naming==0.11.1

pycec/__init__.py

@@ -4,4 +4,4 @@
 _LOGGER = logging.getLogger(__name__)
 
 DEFAULT_PORT = 9526
-DEFAULT_HOST = '0.0.0.0'
+DEFAULT_HOST = "0.0.0.0"

pycec/__main__.py

@@ -27,42 +27,51 @@ def main():
 
     transports = set()
     loop = asyncio.get_event_loop()
-    network = HDMINetwork(CecAdapter("pyCEC", activate_source=False),
-                          loop=loop)
+    network = HDMINetwork(
+        CecAdapter("pyCEC", activate_source=False), loop=loop
+    )
 
     class CECServerProtocol(asyncio.Protocol):
         transport = None
-        buffer = ''
+        buffer = ""
 
         def connection_made(self, transport):
-            _LOGGER.info("Connection opened by %s",
-                         transport.get_extra_info('peername'))
+            _LOGGER.info(
+                "Connection opened by %s", transport.get_extra_info("peername")
+            )
             self.transport = transport
             transports.add(transport)
 
         def data_received(self, data):
             self.buffer += bytes.decode(data)
             for line in self.buffer.splitlines(keepends=True):
-                if line.endswith('\n'):
+                if line.endswith("\n"):
                     line = line.rstrip()
                     if len(line) == 2:
-                        _LOGGER.info("Received poll %s from %s", line,
-                                     self.transport.get_extra_info('peername'))
+                        _LOGGER.info(
+                            "Received poll %s from %s",
+                            line,
+                            self.transport.get_extra_info("peername"),
+                        )
                         d = CecCommand(line).dst
                         t = network._adapter.poll_device(d)
-                        t.add_done_callback(
-                            functools.partial(_after_poll, d))
+                        t.add_done_callback(functools.partial(_after_poll, d))
                     else:
-                        _LOGGER.info("Received command %s from %s", line,
-                                     self.transport.get_extra_info('peername'))
+                        _LOGGER.info(
+                            "Received command %s from %s",
+                            line,
+                            self.transport.get_extra_info("peername"),
+                        )
                         network.send_command(CecCommand(line))
-                    self.buffer = ''
+                    self.buffer = ""
                 else:
                     self.buffer = line
 
         def connection_lost(self, exc):
-            _LOGGER.info("Connection with %s lost",
-                         self.transport.get_extra_info('peername'))
+            _LOGGER.info(
+                "Connection with %s lost",
+                self.transport.get_extra_info("peername"),
+            )
             transports.remove(self.transport)
 
     def _after_poll(d, f):
@@ -72,20 +81,24 @@ def _after_poll(d, f):
 
     def _send_command_to_tcp(command):
         for t in transports:
-            _LOGGER.info("Sending %s to %s", command,
-                         t.get_extra_info('peername'))
+            _LOGGER.info(
+                "Sending %s to %s", command, t.get_extra_info("peername")
+            )
             t.write(str.encode("%s\n" % command.raw))
 
     network.set_command_callback(_send_command_to_tcp)
     loop.run_until_complete(network.async_init())
 
     _LOGGER.info("CEC initialized... Starting server.")
     # Each client connection will create a new protocol instance
-    coro = loop.create_server(CECServerProtocol, config['DEFAULT']['host'],
-                              int(config['DEFAULT']['port']))
+    coro = loop.create_server(
+        CECServerProtocol,
+        config["DEFAULT"]["host"],
+        int(config["DEFAULT"]["port"]),
+    )
     server = loop.run_until_complete(coro)
     # Serve requests until Ctrl+C is pressed
-    _LOGGER.info('Serving on {}'.format(server.sockets[0].getsockname()))
+    _LOGGER.info("Serving on {}".format(server.sockets[0].getsockname()))
     if _LOGGER.level >= logging.DEBUG:
         loop.create_task(async_show_devices(network, loop))
     try:
@@ -101,37 +114,63 @@ def _send_command_to_tcp(command):
 
 def configure():
     parser = OptionParser()
-    parser.add_option("-i", "--interface", dest="host", action="store",
-                      type="string", default=DEFAULT_HOST,
-                      help=("Address of interface to bind to. Default is '%s'."
-                            % DEFAULT_HOST))
-    parser.add_option("-p", "--port", dest="port", action="store", type="int",
-                      default=DEFAULT_PORT,
-                      help=("Port to bind to. Default is '%s'."
-                            % DEFAULT_PORT))
-    parser.add_option("-v", "--verbose", dest="verbose", action="count",
-                      default=0, help="Increase verbosity.")
-    parser.add_option("-q", "--quiet", dest="quiet", action="count",
-                      default=0, help="Decrease verbosity.")
+    parser.add_option(
+        "-i",
+        "--interface",
+        dest="host",
+        action="store",
+        type="string",
+        default=DEFAULT_HOST,
+        help=(
+            "Address of interface to bind to. Default is '%s'." % DEFAULT_HOST
+        ),
+    )
+    parser.add_option(
+        "-p",
+        "--port",
+        dest="port",
+        action="store",
+        type="int",
+        default=DEFAULT_PORT,
+        help=("Port to bind to. Default is '%s'." % DEFAULT_PORT),
+    )
+    parser.add_option(
+        "-v",
+        "--verbose",
+        dest="verbose",
+        action="count",
+        default=0,
+        help="Increase verbosity.",
+    )
+    parser.add_option(
+        "-q",
+        "--quiet",
+        dest="quiet",
+        action="count",
+        default=0,
+        help="Decrease verbosity.",
+    )
     (options, args) = parser.parse_args()
     script_dir = os.path.dirname(os.path.realpath(__file__))
     config = configparser.ConfigParser()
-    config['DEFAULT'] = {'host': options.host, 'port': options.port,
-                         'logLevel': logging.INFO + (
-                             (options.quiet - options.verbose) * 10)}
-    paths = ['/etc/pycec.conf', script_dir + '/pycec.conf']
-    if 'HOME' in os.environ:
-        paths.append(os.environ['HOME'] + '/.pycec')
+    config["DEFAULT"] = {
+        "host": options.host,
+        "port": options.port,
+        "logLevel": logging.INFO + ((options.quiet - options.verbose) * 10),
+    }
+    paths = ["/etc/pycec.conf", script_dir + "/pycec.conf"]
+    if "HOME" in os.environ:
+        paths.append(os.environ["HOME"] + "/.pycec")
     config.read(paths)
 
     return config
 
 
 def setup_logger(config):
     try:
-        log_level = int(config['DEFAULT']['logLevel'])
+        log_level = int(config["DEFAULT"]["logLevel"])
     except ValueError:
-        log_level = config['DEFAULT']['logLevel']
+        log_level = config["DEFAULT"]["logLevel"]
     _LOGGER.setLevel(log_level)
     ch = logging.StreamHandler()
     ch.setLevel(log_level)
@@ -143,16 +182,17 @@ def setup_logger(config):
             datefmt=None,
             reset=True,
             log_colors={
-                'DEBUG': 'cyan',
-                'INFO': 'green',
-                'WARNING': 'yellow',
-                'ERROR': 'red',
-                'CRITICAL': 'red',
-            }
+                "DEBUG": "cyan",
+                "INFO": "green",
+                "WARNING": "yellow",
+                "ERROR": "red",
+                "CRITICAL": "red",
+            },
         )
     except ImportError:
         formatter = logging.Formatter(
-            '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+            "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+        )
     ch.setFormatter(formatter)
     _LOGGER.addHandler(ch)
 

pycec/cec.py

@@ -11,13 +11,18 @@
 
 # pragma: no cover
 class CecAdapter(AbstractCecAdapter):
-    def __init__(self, name: str = None, monitor_only: bool = None,
-                 activate_source: bool = None,
-                 device_type=ADDR_RECORDINGDEVICE1):
+    def __init__(
+        self,
+        name: str = None,
+        monitor_only: bool = None,
+        activate_source: bool = None,
+        device_type=ADDR_RECORDINGDEVICE1,
+    ):
         super().__init__()
         self._adapter = None
         self._io_executor = ThreadPoolExecutor(1)
         import cec
+
         self._cecconfig = cec.libcec_configuration()
         if monitor_only is not None:
             self._cecconfig.bMonitorOnly = 1 if monitor_only else 0
@@ -28,16 +33,19 @@ def __init__(self, name: str = None, monitor_only: bool = None,
 
     def set_command_callback(self, callback):
         self._cecconfig.SetKeyPressCallback(
-            lambda key, delay: callback(KeyPressCommand(key).raw))
+            lambda key, delay: callback(KeyPressCommand(key).raw)
+        )
         self._cecconfig.SetCommandCallback(callback)
 
     def standby_devices(self):
-        self._loop.run_in_executor(self._io_executor,
-                                   self._adapter.StandbyDevices)
+        self._loop.run_in_executor(
+            self._io_executor, self._adapter.StandbyDevices
+        )
 
     def poll_device(self, device):
         return self._loop.run_in_executor(
-            self._io_executor, self._adapter.PollDevice, device)
+            self._io_executor, self._adapter.PollDevice, device
+        )
 
     def shutdown(self):
         self._io_executor.shutdown()
@@ -48,20 +56,25 @@ def get_logical_address(self):
         return self._adapter.GetLogicalAddresses().primary
 
     def power_on_devices(self):
-        self._loop.run_in_executor(self._io_executor,
-                                   self._adapter.PowerOnDevices)
+        self._loop.run_in_executor(
+            self._io_executor, self._adapter.PowerOnDevices
+        )
 
     def transmit(self, command: CecCommand):
         self._loop.run_in_executor(
-            self._io_executor, self._adapter.Transmit,
-            self._adapter.CommandFromString(command.raw))
+            self._io_executor,
+            self._adapter.Transmit,
+            self._adapter.CommandFromString(command.raw),
+        )
 
     def init(self, callback: callable = None):
-        return self._loop.run_in_executor(self._io_executor, self._init,
-                                          callback)
+        return self._loop.run_in_executor(
+            self._io_executor, self._init, callback
+        )
 
     def _init(self, callback: callable = None):
         import cec
+
         if not self._cecconfig.clientVersion:
             self._cecconfig.clientVersion = cec.LIBCEC_VERSION_CURRENT
         _LOGGER.debug("Initializing CEC...")
@@ -72,9 +85,14 @@ def _init(self, callback: callable = None):
         for a in adapters:
             _LOGGER.info("found a CEC adapter:")
             _LOGGER.info("port:     " + a.strComName)
-            _LOGGER.info("vendor:   " + (
-                VENDORS[a.iVendorId] if a.iVendorId in VENDORS else hex(
-                    a.iVendorId)))
+            _LOGGER.info(
+                "vendor:   "
+                + (
+                    VENDORS[a.iVendorId]
+                    if a.iVendorId in VENDORS
+                    else hex(a.iVendorId)
+                )
+            )
             _LOGGER.info("product:  " + hex(a.iProductId))
             a = a.strComName
         if a is None:

pycec/commands.py

@@ -4,8 +4,14 @@
 
 
 class CecCommand:
-    def __init__(self, cmd, dst: int = None, src: int = None,
-                 att: List[int] = None, raw: str = None):
+    def __init__(
+        self,
+        cmd,
+        dst: int = None,
+        src: int = None,
+        att: List[int] = None,
+        raw: str = None,
+    ):
 
         self._src = src
         self._dst = dst
@@ -52,13 +58,16 @@ def _att(self, value: List[int]):  # pragma: no cover
     @property
     def raw(self) -> str:
         atts = "".join(((":%02x" % i) for i in self.att))
-        cmd = ("" if self.cmd is None else (":%02x" % self.cmd))
-        return "%1x%1x%s%s" % (self.src if self.src is not None else 0xf,
-                               self.dst if self.dst is not None else 0xf,
-                               cmd, atts)
+        cmd = "" if self.cmd is None else (":%02x" % self.cmd)
+        return "%1x%1x%s%s" % (
+            self.src if self.src is not None else 0xF,
+            self.dst if self.dst is not None else 0xF,
+            cmd,
+            atts,
+        )
 
     def _raw(self, value: str):
-        atts = value.split(':')
+        atts = value.split(":")
         self.src = int(atts[0][0], 16)
         self.dst = int(atts[0][1], 16)
         if len(atts) > 1: