parameters: (asok_path: str,
                   cmd: List[str],
                   format: Optional[str] = '')
return_statement: return ret
docstring: Send a daemon (--admin-daemon) command 'cmd'.  asok_path is the
      path to the admin socket; cmd is a list of strings; format may be
      set to one of the formatted forms to get output in that form
      (daemon commands don't support 'plain' output).
docstring_summary: Send a daemon (--admin-daemon) command 'cmd'.  asok_path is the
      path to the admin socket; cmd is a list of strings; format may be
      set to one of the formatted forms to get output in that form
      (daemon commands don't support 'plain' output).
docstring_tokens: Send a daemon ( -- admin - daemon ) command cmd . asok_path is the path to the admin socket ; cmd is a list of strings ; format may be set to one of the formatted forms to get output in that form ( daemon commands don t support plain output ) .
function: def admin_socket(asok_path: str,
                   cmd: List[str],
                   format: Optional[str] = '') -> bytes:
      """
      Send a daemon (--admin-daemon) command 'cmd'.  asok_path is the
      path to the admin socket; cmd is a list of strings; format may be
      set to one of the formatted forms to get output in that form
      (daemon commands don't support 'plain' output).
      """
  
      def do_sockio(path: str, cmd_bytes: bytes) -> bytes:
          """ helper: do all the actual low-level stream I/O """
          sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
          sock.connect(path)
          try:
              sock.sendall(cmd_bytes + b'\0')
              len_str = sock.recv(4)
              if len(len_str) < 4:
                  raise RuntimeError("no data returned from admin socket")
              l, = struct.unpack(">I", len_str)
              sock_ret = b''
  
              got = 0
              while got < l:
                  # recv() receives signed int, i.e max 2GB
                  # workaround by capping READ_CHUNK_SIZE per call.
                  want = min(l - got, READ_CHUNK_SIZE)
                  bit = sock.recv(want)
                  sock_ret += bit
                  got += len(bit)
  
          except Exception as sock_e:
              raise RuntimeError('exception: ' + str(sock_e))
          return sock_ret
  
      try:
          cmd_json = do_sockio(asok_path,
                               b'{"prefix": "get_command_descriptions"}')
      except Exception as e:
          raise RuntimeError('exception getting command descriptions: ' + str(e))
  
      sigdict = parse_json_funcsigs(cmd_json.decode('utf-8'), 'cli')
      valid_dict = validate_command(sigdict, cmd)
      if not valid_dict:
          raise RuntimeError('invalid command')
  
      if format:
          valid_dict['format'] = format
  
      try:
          ret = do_sockio(asok_path, json.dumps(valid_dict).encode('utf-8'))
      except Exception as e:
          raise RuntimeError('exception: ' + str(e))
  
      return ret
function_tokens: def admin_socket ( asok_path : str , cmd : List [ str ] , format : Optional [ str ] = '' ) -> bytes : def do_sockio ( path : str , cmd_bytes : bytes ) -> bytes : """ helper: do all the actual low-level stream I/O """ sock = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) sock . connect ( path ) try : sock . sendall ( cmd_bytes + b'\0' ) len_str = sock . recv ( 4 ) if len ( len_str ) < 4 : raise RuntimeError ( "no data returned from admin socket" ) l , = struct . unpack ( ">I" , len_str ) sock_ret = b'' got = 0 while got < l : # recv() receives signed int, i.e max 2GB # workaround by capping READ_CHUNK_SIZE per call. want = min ( l - got , READ_CHUNK_SIZE ) bit = sock . recv ( want ) sock_ret += bit got += len ( bit ) except Exception as sock_e : raise RuntimeError ( 'exception: ' + str ( sock_e ) ) return sock_ret try : cmd_json = do_sockio ( asok_path , b'{"prefix": "get_command_descriptions"}' ) except Exception as e : raise RuntimeError ( 'exception getting command descriptions: ' + str ( e ) ) sigdict = parse_json_funcsigs ( cmd_json . decode ( 'utf-8' ) , 'cli' ) valid_dict = validate_command ( sigdict , cmd ) if not valid_dict : raise RuntimeError ( 'invalid command' ) if format : valid_dict [ 'format' ] = format try : ret = do_sockio ( asok_path , json . dumps ( valid_dict ) . encode ( 'utf-8' ) ) except Exception as e : raise RuntimeError ( 'exception: ' + str ( e ) ) return ret