peppa

Author: marcin-usielski

moler/config/connections.py

@@ -228,7 +228,6 @@ def terminal_nofork_thd_conn_st(name=None):
         io_conn = ThreadedTerminalNoForkPTY(moler_connection=mlr_conn)  # TODO: add name, logger
         return io_conn
 
-
     # TODO: unify passing logger to io_conn (logger/logger_name)
     connection_factory.register_construction(io_type="terminal",
                                              variant="threaded",
@@ -244,7 +243,6 @@ def terminal_nofork_thd_conn_st(name=None):
                                              variant="single-threaded",
                                              constructor=terminal_thd_conn_st)
 
-
     # TODO: unify passing logger to io_conn (logger/logger_name)
     connection_factory.register_construction(io_type="terminal_no_forkpty",
                                              variant="threaded",
@@ -260,6 +258,7 @@ def terminal_nofork_thd_conn_st(name=None):
                                              variant="single-threaded",
                                              constructor=terminal_nofork_thd_conn_st)
 
+
 def _register_builtin_py3_unix_connections(connection_factory, moler_conn_class):
     from moler.io.asyncio.terminal import AsyncioTerminal, AsyncioInThreadTerminal
 

moler/io/raw/terminal_no_forkpty.py

@@ -28,18 +28,18 @@
 
 class PtyProcessUnicodeNotFork:
     """PtyProcessUnicode without forking process."""
-    def __init__(self, cmd: str = "/bin/bash", dimensions: Tuple[int, int]=(25, 120), buffer_size: int=4096):
+    def __init__(self, cmd: str = "/bin/bash", dimensions: Tuple[int, int] =( 25, 120), buffer_size: int = 4096):
         self.cmd = cmd
         self.dimensions = dimensions
         self.buffer_size = buffer_size
         self.delayafterclose = 0.2
         encoding = "utf-8"
         self.decoder = codecs.getincrementaldecoder(encoding)(errors='strict')
-        self.fd : int = -1  # File descriptor for pty master
-        self.pid : int = -1  # Process ID of the child process
-        self.slave_fd : int = -1  # File descriptor for pty slave
-        self.process : Optional[subprocess.Popen] = None  # Subprocess.Popen object
-        self._closed : bool = True
+        self.fd: int = -1  # File descriptor for pty master
+        self.pid: int = -1  # Process ID of the child process
+        self.slave_fd: int = -1  # File descriptor for pty slave
+        self.process: Optional[subprocess.Popen] = None  # Subprocess.Popen object
+        self._closed: bool = True
 
     def create_pty_process(self):
         """Create PtyProcessUnicode without forking process."""