text: import io, os, re, serial, struct, time
  from errno import EPERM
  from .console import VT_ENABLED
  
  try:
      from .pyboard import Pyboard, PyboardError, stdout_write_bytes, filesystem_command
  except ImportError:
      import sys
  
      sys.path.append(os.path.dirname(__file__) + "/../..")
      from pyboard import Pyboard, PyboardError, stdout_write_bytes, filesystem_command
  
  fs_hook_cmds = {
      "CMD_STAT": 1,
      "CMD_ILISTDIR_START": 2,
      "CMD_ILISTDIR_NEXT": 3,
      "CMD_OPEN": 4,
      "CMD_CLOSE": 5,
      "CMD_READ": 6,
      "CMD_WRITE": 7,
      "CMD_SEEK": 8,
      "CMD_REMOVE": 9,
      "CMD_RENAME": 10,
      "CMD_MKDIR": 11,
      "CMD_RMDIR": 12,
  }
  
  fs_hook_code = """\
  import uos, uio, ustruct, micropython
  
  SEEK_SET = 0
  
  class RemoteCommand:
      def __init__(self):
          import uselect, usys
          self.buf4 = bytearray(4)
          self.fout = usys.stdout.buffer
          self.fin = usys.stdin.buffer
          self.poller = uselect.poll()
          self.poller.register(self.fin, uselect.POLLIN)
  
      def poll_in(self):
          for _ in self.poller.ipoll(1000):
              return
          self.end()
          raise Exception('timeout waiting for remote')
  
      def rd(self, n):
          buf = bytearray(n)
          self.rd_into(buf, n)
          return buf
  
      def rd_into(self, buf, n):
          # implement reading with a timeout in case other side disappears
          if n == 0:
              return
          self.poll_in()
          r = self.fin.readinto(buf, n)
          if r < n:
              mv = memoryview(buf)
              while r < n:
                  self.poll_in()
                  r += self.fin.readinto(mv[r:], n - r)
  
      def begin(self, type):
          micropython.kbd_intr(-1)
          buf4 = self.buf4
          buf4[0] = 0x18
          buf4[1] = type
          self.fout.write(buf4, 2)
          # Wait for sync byte 0x18, but don't get stuck forever
          for i in range(30):
              self.poller.poll(1000)
              self.fin.readinto(buf4, 1)
              if buf4[0] == 0x18:
                  break
  
      def end(self):
          micropython.kbd_intr(3)
  
      def rd_s8(self):
          self.rd_into(self.buf4, 1)
          n = self.buf4[0]
          if n & 0x80:
              n -= 0x100
          return n
  
      def rd_s32(self):
          buf4 = self.buf4
          self.rd_into(buf4, 4)
          n = buf4[0] | buf4[1] << 8 | buf4[2] << 16 | buf4[3] << 24
          if buf4[3] & 0x80:
              n -= 0x100000000
          return n
  
      def rd_u32(self):
          buf4 = self.buf4
          self.rd_into(buf4, 4)
          return buf4[0] | buf4[1] << 8 | buf4[2] << 16 | buf4[3] << 24
  
      def rd_bytes(self, buf):
          # TODO if n is large (eg >256) then we may miss bytes on stdin
          n = self.rd_s32()
          if buf is None:
              ret = buf = bytearray(n)
          else:
              ret = n
          self.rd_into(buf, n)
          return ret
  
      def rd_str(self):
          n = self.rd_s32()
          if n == 0:
              return ''
          else:
              return str(self.rd(n), 'utf8')
  
      def wr_s8(self, i):
          self.buf4[0] = i
          self.fout.write(self.buf4, 1)
  
      def wr_s32(self, i):
          ustruct.pack_into('<i', self.buf4, 0, i)
          self.fout.write(self.buf4)
  
      def wr_bytes(self, b):
          self.wr_s32(len(b))
          self.fout.write(b)
  
      # str and bytes act the same in MicroPython
      wr_str = wr_bytes
  
  
  class RemoteFile(uio.IOBase):
      def __init__(self, cmd, fd, is_text):
          self.cmd = cmd
          self.fd = fd
          self.is_text = is_text
  
      def __enter__(self):
          return self
  
      def __exit__(self, a, b, c):
          self.close()
  
      def ioctl(self, request, arg):
          if request == 1:  # FLUSH
              self.flush()
          elif request == 2:  # SEEK
              # This assumes a 32-bit bare-metal machine.
              import machine
              machine.mem32[arg] = self.seek(machine.mem32[arg], machine.mem32[arg + 4])
          elif request == 4:  # CLOSE
              self.close()
          return 0
  
      def flush(self):
          pass
  
      def close(self):
          if self.fd is None:
              return
          c = self.cmd
          c.begin(CMD_CLOSE)
          c.wr_s8(self.fd)
          c.end()
          self.fd = None
  
      def read(self, n=-1):
          c = self.cmd
          c.begin(CMD_READ)
          c.wr_s8(self.fd)
          c.wr_s32(n)
          data = c.rd_bytes(None)
          c.end()
          if self.is_text:
              data = str(data, 'utf8')
          else:
              data = bytes(data)
          return data
  
      def readinto(self, buf):
          c = self.cmd
          c.begin(CMD_READ)
          c.wr_s8(self.fd)
          c.wr_s32(len(buf))
          n = c.rd_bytes(buf)
          c.end()
          return n
  
      def readline(self):
          l = ''
          while 1:
              c = self.read(1)
              l += c
              if c == '\\n' or c == '':
                  return l
  
      def readlines(self):
          ls = []
          while 1:
              l = self.readline()
              if not l:
                  return ls
              ls.append(l)
  
      def write(self, buf):
          c = self.cmd
          c.begin(CMD_WRITE)
          c.wr_s8(self.fd)
          c.wr_bytes(buf)
          n = c.rd_s32()
          c.end()
          return n
  
      def seek(self, n, whence=SEEK_SET):
          c = self.cmd
          c.begin(CMD_SEEK)
          c.wr_s8(self.fd)
          c.wr_s32(n)
          c.wr_s8(whence)
          n = c.rd_s32()
          c.end()
          if n < 0:
              raise OSError(n)
          return n
  
  
  class RemoteFS:
      def __init__(self, cmd):
          self.cmd = cmd
  
      def mount(self, readonly, mkfs):
          pass
  
      def umount(self):
          pass
  
      def chdir(self, path):
          if not path.startswith("/"):
              path = self.path + path
          if not path.endswith("/"):
              path += "/"
          if path != "/":
              self.stat(path)
          self.path = path
  
      def getcwd(self):
          return self.path
  
      def remove(self, path):
          c = self.cmd
          c.begin(CMD_REMOVE)
          c.wr_str(self.path + path)
          res = c.rd_s32()
          c.end()
          if res < 0:
              raise OSError(-res)
  
      def rename(self, old, new):
          c = self.cmd
          c.begin(CMD_RENAME)
          c.wr_str(self.path + old)
          c.wr_str(self.path + new)
          res = c.rd_s32()
          c.end()
          if res < 0:
              raise OSError(-res)
  
      def mkdir(self, path):
          c = self.cmd
          c.begin(CMD_MKDIR)
          c.wr_str(self.path + path)
          res = c.rd_s32()
          c.end()
          if res < 0:
              raise OSError(-res)
  
      def rmdir(self, path):
          c = self.cmd
          c.begin(CMD_RMDIR)
          c.wr_str(self.path + path)
          res = c.rd_s32()
          c.end()
          if res < 0:
              raise OSError(-res)
  
      def stat(self, path):
          c = self.cmd
          c.begin(CMD_STAT)
          c.wr_str(self.path + path)
          res = c.rd_s8()
          if res < 0:
              c.end()
              raise OSError(-res)
          mode = c.rd_u32()
          size = c.rd_u32()
          atime = c.rd_u32()
          mtime = c.rd_u32()
          ctime = c.rd_u32()
          c.end()
          return mode, 0, 0, 0, 0, 0, size, atime, mtime, ctime
  
      def ilistdir(self, path):
          c = self.cmd
          c.begin(CMD_ILISTDIR_START)
          c.wr_str(self.path + path)
          res = c.rd_s8()
          c.end()
          if res < 0:
              raise OSError(-res)
          def next():
              while True:
                  c.begin(CMD_ILISTDIR_NEXT)
                  name = c.rd_str()
                  if name:
                      type = c.rd_u32()
                      c.end()
                      yield (name, type, 0)
                  else:
                      c.end()
                      break
          return next()
  
      def open(self, path, mode):
          c = self.cmd
          c.begin(CMD_OPEN)
          c.wr_str(self.path + path)
          c.wr_str(mode)
          fd = c.rd_s8()
          c.end()
          if fd < 0:
              raise OSError(-fd)
          return RemoteFile(c, fd, mode.find('b') == -1)
  
  
  def __mount():
      uos.mount(RemoteFS(RemoteCommand()), '/remote')
      uos.chdir('/remote')
  """
  
  # Apply basic compression on hook code.
  for key, value in fs_hook_cmds.items():
      fs_hook_code = re.sub(key, str(value), fs_hook_code)
  fs_hook_code = re.sub(" *#.*$", "", fs_hook_code, flags=re.MULTILINE)
  fs_hook_code = re.sub("\n\n+", "\n", fs_hook_code)
  fs_hook_code = re.sub("    ", " ", fs_hook_code)
  fs_hook_code = re.sub("rd_", "r", fs_hook_code)
  fs_hook_code = re.sub("wr_", "w", fs_hook_code)
  fs_hook_code = re.sub("buf4", "b4", fs_hook_code)
  
  
  class PyboardCommand:
      def __init__(self, fin, fout, path):
          self.fin = fin
          self.fout = fout
          self.root = path + "/"
          self.data_ilistdir = ["", []]
          self.data_files = []
  
      def rd_s8(self):
          return struct.unpack("<b", self.fin.read(1))[0]
  
      def rd_s32(self):
          return struct.unpack("<i", self.fin.read(4))[0]
  
      def rd_bytes(self):
          n = self.rd_s32()
          return self.fin.read(n)
  
      def rd_str(self):
          n = self.rd_s32()
          if n == 0:
              return ""
          else:
              return str(self.fin.read(n), "utf8")
  
      def wr_s8(self, i):
          self.fout.write(struct.pack("<b", i))
  
      def wr_s32(self, i):
          self.fout.write(struct.pack("<i", i))
  
      def wr_u32(self, i):
          self.fout.write(struct.pack("<I", i))
  
      def wr_bytes(self, b):
          self.wr_s32(len(b))
          self.fout.write(b)
  
      def wr_str(self, s):
          b = bytes(s, "utf8")
          self.wr_s32(len(b))
          self.fout.write(b)
  
      def log_cmd(self, msg):
          print(f"[{msg}]", end="\r\n")
  
      def path_check(self, path):
          parent = os.path.realpath(self.root)
          child = os.path.realpath(path)
          if parent != os.path.commonpath([parent, child]):
              raise OSError(EPERM, "")  # File is outside mounted dir
  
      def do_stat(self):
          path = self.root + self.rd_str()
          # self.log_cmd(f"stat {path}")
          try:
              self.path_check(path)
              stat = os.stat(path)
          except OSError as er:
              self.wr_s8(-abs(er.errno))
          else:
              self.wr_s8(0)
              # Note: st_ino would need to be 64-bit if added here
              self.wr_u32(stat.st_mode)
              self.wr_u32(stat.st_size)
              self.wr_u32(int(stat.st_atime))
              self.wr_u32(int(stat.st_mtime))
              self.wr_u32(int(stat.st_ctime))
  
      def do_ilistdir_start(self):
          path = self.root + self.rd_str()
          try:
              self.path_check(path)
              self.wr_s8(0)
          except OSError as er:
              self.wr_s8(-abs(er.errno))
          else:
              self.data_ilistdir[0] = path
              self.data_ilistdir[1] = os.listdir(path)
  
      def do_ilistdir_next(self):
          if self.data_ilistdir[1]:
              entry = self.data_ilistdir[1].pop(0)
              try:
                  stat = os.lstat(self.data_ilistdir[0] + "/" + entry)
                  mode = stat.st_mode & 0xC000
              except OSError as er:
                  mode = 0
              self.wr_str(entry)
              self.wr_u32(mode)
          else:
              self.wr_str("")
  
      def do_open(self):
          path = self.root + self.rd_str()
          mode = self.rd_str()
          # self.log_cmd(f"open {path} {mode}")
          try:
              self.path_check(path)
              f = open(path, mode)
          except OSError as er:
              self.wr_s8(-abs(er.errno))
          else:
              is_text = mode.find("b") == -1
              try:
                  fd = self.data_files.index(None)
                  self.data_files[fd] = (f, is_text)
              except ValueError:
                  fd = len(self.data_files)
                  self.data_files.append((f, is_text))
              self.wr_s8(fd)
  
      def do_close(self):
          fd = self.rd_s8()
          # self.log_cmd(f"close {fd}")
          self.data_files[fd][0].close()
          self.data_files[fd] = None
  
      def do_read(self):
          fd = self.rd_s8()
          n = self.rd_s32()
          buf = self.data_files[fd][0].read(n)
          if self.data_files[fd][1]:
              buf = bytes(buf, "utf8")
          self.wr_bytes(buf)
          # self.log_cmd(f"read {fd} {n} -> {len(buf)}")
  
      def do_seek(self):
          fd = self.rd_s8()
          n = self.rd_s32()
          whence = self.rd_s8()
          # self.log_cmd(f"seek {fd} {n}")
          try:
              n = self.data_files[fd][0].seek(n, whence)
          except io.UnsupportedOperation:
              n = -1
          self.wr_s32(n)
  
      def do_write(self):
          fd = self.rd_s8()
          buf = self.rd_bytes()
          if self.data_files[fd][1]:
              buf = str(buf, "utf8")
          n = self.data_files[fd][0].write(buf)
          self.wr_s32(n)
          # self.log_cmd(f"write {fd} {len(buf)} -> {n}")
  
      def do_remove(self):
          path = self.root + self.rd_str()
          # self.log_cmd(f"remove {path}")
          try:
              self.path_check(path)
              os.remove(path)
              ret = 0
          except OSError as er:
              ret = -abs(er.errno)
          self.wr_s32(ret)
  
      def do_rename(self):
          old = self.root + self.rd_str()
          new = self.root + self.rd_str()
          # self.log_cmd(f"rename {old} {new}")
          try:
              self.path_check(old)
              self.path_check(new)
              os.rename(old, new)
              ret = 0
          except OSError as er:
              ret = -abs(er.errno)
          self.wr_s32(ret)
  
      def do_mkdir(self):
          path = self.root + self.rd_str()
          # self.log_cmd(f"mkdir {path}")
          try:
              self.path_check(path)
              os.mkdir(path)
              ret = 0
          except OSError as er:
              ret = -abs(er.errno)
          self.wr_s32(ret)
  
      def do_rmdir(self):
          path = self.root + self.rd_str()
          # self.log_cmd(f"rmdir {path}")
          try:
              self.path_check(path)
              os.rmdir(path)
              ret = 0
          except OSError as er:
              ret = -abs(er.errno)
          self.wr_s32(ret)
  
      cmd_table = {
          fs_hook_cmds["CMD_STAT"]: do_stat,
          fs_hook_cmds["CMD_ILISTDIR_START"]: do_ilistdir_start,
          fs_hook_cmds["CMD_ILISTDIR_NEXT"]: do_ilistdir_next,
          fs_hook_cmds["CMD_OPEN"]: do_open,
          fs_hook_cmds["CMD_CLOSE"]: do_close,
          fs_hook_cmds["CMD_READ"]: do_read,
          fs_hook_cmds["CMD_WRITE"]: do_write,
          fs_hook_cmds["CMD_SEEK"]: do_seek,
          fs_hook_cmds["CMD_REMOVE"]: do_remove,
          fs_hook_cmds["CMD_RENAME"]: do_rename,
          fs_hook_cmds["CMD_MKDIR"]: do_mkdir,
          fs_hook_cmds["CMD_RMDIR"]: do_rmdir,
      }
  
  
  class SerialIntercept:
      def __init__(self, serial, cmd):
          self.orig_serial = serial
          self.cmd = cmd
          self.buf = b""
          self.orig_serial.timeout = 5.0
  
      def _check_input(self, blocking):
          if blocking or self.orig_serial.inWaiting() > 0:
              c = self.orig_serial.read(1)
              if c == b"\x18":
                  # a special command
                  c = self.orig_serial.read(1)[0]
                  self.orig_serial.write(b"\x18")  # Acknowledge command
                  PyboardCommand.cmd_table[c](self.cmd)
              elif not VT_ENABLED and c == b"\x1b":
                  # ESC code, ignore these on windows
                  esctype = self.orig_serial.read(1)
                  if esctype == b"[":  # CSI
                      while not (0x40 < self.orig_serial.read(1)[0] < 0x7E):
                          # Looking for "final byte" of escape sequence
                          pass
              else:
                  self.buf += c
  
      @property
      def fd(self):
          return self.orig_serial.fd
  
      def close(self):
          self.orig_serial.close()
  
      def inWaiting(self):
          self._check_input(False)
          return len(self.buf)
  
      def read(self, n):
          while len(self.buf) < n:
              self._check_input(True)
          out = self.buf[:n]
          self.buf = self.buf[n:]
          return out
  
      def write(self, buf):
          self.orig_serial.write(buf)
  
  
  class PyboardExtended(Pyboard):
      def __init__(self, dev, *args, **kwargs):
          super().__init__(dev, *args, **kwargs)
          self.device_name = dev
          self.mounted = False
  
      def mount_local(self, path):
          fout = self.serial
          if self.eval('"RemoteFS" in globals()') == b"False":
              self.exec_(fs_hook_code)
          self.exec_("__mount()")
          self.mounted = True
          self.cmd = PyboardCommand(self.serial, fout, path)
          self.serial = SerialIntercept(self.serial, self.cmd)
  
      def soft_reset_with_mount(self, out_callback):
          self.serial.write(b"\x04")
          if not self.mounted:
              return
  
          # Clear flag while board reboots, it will be re-set once mounted.
          self.mounted = False
  
          # Wait for a response to the soft-reset command.
          for i in range(10):
              if self.serial.inWaiting():
                  break
              time.sleep(0.05)
          else:
              # Device didn't respond so it wasn't in a state to do a soft reset.
              return
  
          out_callback(self.serial.read(1))
          self.serial = self.serial.orig_serial
          n = self.serial.inWaiting()
          while n > 0:
              buf = self.serial.read(n)
              out_callback(buf)
              time.sleep(0.2)
              n = self.serial.inWaiting()
          self.serial.write(b"\x01")
          self.exec_(fs_hook_code)
          self.exec_("__mount()")
          self.mounted = True
          self.exit_raw_repl()
          self.read_until(4, b">>> ")
          self.serial = SerialIntercept(self.serial, self.cmd)
  
      def umount_local(self):
          if self.mounted:
              self.exec_('uos.umount("/remote")')
              self.mounted = False
  