| 1 | """
|
| 2 | prompt.py: A LIBRARY for prompt evaluation.
|
| 3 |
|
| 4 | User interface details should go in core/ui.py.
|
| 5 | """
|
| 6 | from __future__ import print_function
|
| 7 |
|
| 8 | import time as time_
|
| 9 |
|
| 10 | from _devbuild.gen.id_kind_asdl import Id, Id_t
|
| 11 | from _devbuild.gen.syntax_asdl import (loc, command_t, source, CompoundWord)
|
| 12 | from _devbuild.gen.value_asdl import (value, value_e, value_t, Obj)
|
| 13 | from core import alloc
|
| 14 | from core import main_loop
|
| 15 | from core import error
|
| 16 | from core import pyos
|
| 17 | from core import state
|
| 18 | from display import ui
|
| 19 | from frontend import consts
|
| 20 | from frontend import match
|
| 21 | from frontend import reader
|
| 22 | from mycpp import mylib
|
| 23 | from mycpp.mylib import log, tagswitch
|
| 24 | from osh import word_
|
| 25 | from pylib import os_path
|
| 26 |
|
| 27 | import libc # gethostname()
|
| 28 | import posix_ as posix
|
| 29 |
|
| 30 | from typing import Dict, List, Tuple, Optional, cast, TYPE_CHECKING
|
| 31 | if TYPE_CHECKING:
|
| 32 | from core.state import Mem
|
| 33 | from frontend.parse_lib import ParseContext
|
| 34 | from osh import cmd_eval
|
| 35 | from osh import word_eval
|
| 36 | from ysh import expr_eval
|
| 37 |
|
| 38 | _ = log
|
| 39 |
|
| 40 | #
|
| 41 | # Prompt Evaluation
|
| 42 | #
|
| 43 |
|
| 44 | _ERROR_FMT = '<Error: %s> '
|
| 45 | _UNBALANCED_ERROR = r'Unbalanced \[ and \]'
|
| 46 |
|
| 47 |
|
| 48 | class _PromptEvaluatorCache(object):
|
| 49 | """Cache some values we don't expect to change for the life of a
|
| 50 | process."""
|
| 51 |
|
| 52 | def __init__(self):
|
| 53 | # type: () -> None
|
| 54 | self.cache = {} # type: Dict[str, str]
|
| 55 | self.euid = -1 # invalid value
|
| 56 |
|
| 57 | def _GetEuid(self):
|
| 58 | # type: () -> int
|
| 59 | """Cached lookup."""
|
| 60 | if self.euid == -1:
|
| 61 | self.euid = posix.geteuid()
|
| 62 | return self.euid
|
| 63 |
|
| 64 | def Get(self, name):
|
| 65 | # type: (str) -> str
|
| 66 | if name in self.cache:
|
| 67 | return self.cache[name]
|
| 68 |
|
| 69 | if name == '$': # \$
|
| 70 | value = '#' if self._GetEuid() == 0 else '$'
|
| 71 |
|
| 72 | elif name == 'hostname': # for \h and \H
|
| 73 | value = libc.gethostname()
|
| 74 |
|
| 75 | elif name == 'user': # for \u
|
| 76 | # recursive call for caching
|
| 77 | value = pyos.GetUserName(self._GetEuid())
|
| 78 |
|
| 79 | else:
|
| 80 | raise AssertionError(name)
|
| 81 |
|
| 82 | self.cache[name] = value
|
| 83 | return value
|
| 84 |
|
| 85 |
|
| 86 | class Evaluator(object):
|
| 87 | """Evaluate the prompt mini-language.
|
| 88 |
|
| 89 | bash has a very silly algorithm:
|
| 90 | 1. replace backslash codes, except any $ in those values get quoted into \$.
|
| 91 | 2. Parse the word as if it's in a double quoted context, and then evaluate
|
| 92 | the word.
|
| 93 |
|
| 94 | Haven't done this from POSIX: POSIX:
|
| 95 | http://pubs.opengroup.org/onlinepubs/9699919799/utilities/V3_chap02.html
|
| 96 |
|
| 97 | The shell shall replace each instance of the character '!' in PS1 with the
|
| 98 | history file number of the next command to be typed. Escaping the '!' with
|
| 99 | another '!' (that is, "!!" ) shall place the literal character '!' in the
|
| 100 | prompt.
|
| 101 | """
|
| 102 |
|
| 103 | def __init__(self, lang, version_str, parse_ctx, mem):
|
| 104 | # type: (str, str, ParseContext, Mem) -> None
|
| 105 | self.word_ev = None # type: word_eval.AbstractWordEvaluator
|
| 106 | self.expr_ev = None # type: expr_eval.ExprEvaluator
|
| 107 | self.global_io = None # type: Obj
|
| 108 |
|
| 109 | assert lang in ('osh', 'ysh'), lang
|
| 110 | self.lang = lang
|
| 111 | self.version_str = version_str
|
| 112 | self.parse_ctx = parse_ctx
|
| 113 | self.mem = mem
|
| 114 | # Cache to save syscalls / libc calls.
|
| 115 | self.cache = _PromptEvaluatorCache()
|
| 116 |
|
| 117 | # These caches should reduce memory pressure a bit. We don't want to
|
| 118 | # reparse the prompt twice every time you hit enter.
|
| 119 | self.tokens_cache = {} # type: Dict[str, List[Tuple[Id_t, str]]]
|
| 120 | self.parse_cache = {} # type: Dict[str, CompoundWord]
|
| 121 |
|
| 122 | def CheckCircularDeps(self):
|
| 123 | # type: () -> None
|
| 124 | assert self.word_ev is not None
|
| 125 |
|
| 126 | def PromptVal(self, what):
|
| 127 | # type: (str) -> str
|
| 128 | """
|
| 129 | _io->promptVal('$')
|
| 130 | """
|
| 131 | if what == 'D':
|
| 132 | # TODO: wrap strftime(), time(), localtime(), etc. so users can do
|
| 133 | # it themselves
|
| 134 | return _ERROR_FMT % '\D{} not in promptVal()'
|
| 135 | else:
|
| 136 | # Could make hostname -> h alias, etc.
|
| 137 | return self.PromptSubst(what)
|
| 138 |
|
| 139 | def PromptSubst(self, ch, arg=None):
|
| 140 | # type: (str, Optional[str]) -> str
|
| 141 |
|
| 142 | if ch == '$': # So the user can tell if they're root or not.
|
| 143 | r = self.cache.Get('$')
|
| 144 |
|
| 145 | elif ch == 'u':
|
| 146 | r = self.cache.Get('user')
|
| 147 |
|
| 148 | elif ch == 'h':
|
| 149 | hostname = self.cache.Get('hostname')
|
| 150 | # foo.com -> foo
|
| 151 | r, _ = mylib.split_once(hostname, '.')
|
| 152 |
|
| 153 | elif ch == 'H':
|
| 154 | r = self.cache.Get('hostname')
|
| 155 |
|
| 156 | elif ch == 's':
|
| 157 | r = self.lang
|
| 158 |
|
| 159 | elif ch == 'v':
|
| 160 | r = self.version_str
|
| 161 |
|
| 162 | elif ch == 'A':
|
| 163 | now = time_.time()
|
| 164 | r = time_.strftime('%H:%M', time_.localtime(now))
|
| 165 |
|
| 166 | elif ch == 'D': # \D{%H:%M} is the only one with a suffix
|
| 167 | now = time_.time()
|
| 168 | assert arg is not None
|
| 169 | if len(arg) == 0:
|
| 170 | # In bash y.tab.c uses %X when string is empty
|
| 171 | # This doesn't seem to match exactly, but meh for now.
|
| 172 | fmt = '%X'
|
| 173 | else:
|
| 174 | fmt = arg
|
| 175 | r = time_.strftime(fmt, time_.localtime(now))
|
| 176 |
|
| 177 | elif ch == 'w':
|
| 178 | # HOME doesn't have to exist
|
| 179 | home = state.MaybeString(self.mem, 'HOME')
|
| 180 |
|
| 181 | # Shorten /home/andy/mydir -> ~/mydir
|
| 182 | # Note: could also call sh_init.GetWorkingDir()?
|
| 183 | r = ui.PrettyDir(self.mem.pwd, home)
|
| 184 |
|
| 185 | elif ch == 'W':
|
| 186 | # Note: could also call sh_init.GetWorkingDir()?
|
| 187 | r = os_path.basename(self.mem.pwd)
|
| 188 |
|
| 189 | else:
|
| 190 | # e.g. \e \r \n \\
|
| 191 | r = consts.LookupCharPrompt(ch)
|
| 192 |
|
| 193 | # TODO: Handle more codes
|
| 194 | # R(r'\\[adehHjlnrstT@AuvVwW!#$\\]', Id.PS_Subst),
|
| 195 | if r is None:
|
| 196 | r = _ERROR_FMT % (r'\%s is invalid or unimplemented in $PS1' %
|
| 197 | ch)
|
| 198 |
|
| 199 | return r
|
| 200 |
|
| 201 | def _ReplaceBackslashCodes(self, tokens):
|
| 202 | # type: (List[Tuple[Id_t, str]]) -> str
|
| 203 | ret = [] # type: List[str]
|
| 204 | non_printing = 0
|
| 205 | for id_, s in tokens:
|
| 206 | # BadBacklash means they should have escaped with \\. TODO: Make it an error.
|
| 207 | # 'echo -e' has a similar issue.
|
| 208 | if id_ in (Id.PS_Literals, Id.PS_BadBackslash):
|
| 209 | ret.append(s)
|
| 210 |
|
| 211 | elif id_ == Id.PS_Octal3:
|
| 212 | i = int(s[1:], 8)
|
| 213 | ret.append(chr(i % 256))
|
| 214 |
|
| 215 | elif id_ == Id.PS_LBrace:
|
| 216 | non_printing += 1
|
| 217 | ret.append('\x01')
|
| 218 |
|
| 219 | elif id_ == Id.PS_RBrace:
|
| 220 | non_printing -= 1
|
| 221 | if non_printing < 0: # e.g. \]\[
|
| 222 | return _ERROR_FMT % _UNBALANCED_ERROR
|
| 223 |
|
| 224 | ret.append('\x02')
|
| 225 |
|
| 226 | elif id_ == Id.PS_Subst: # \u \h \w etc.
|
| 227 | ch = s[1]
|
| 228 | arg = None # type: Optional[str]
|
| 229 | if ch == 'D':
|
| 230 | arg = s[3:-1] # \D{%H:%M}
|
| 231 | r = self.PromptSubst(ch, arg=arg)
|
| 232 |
|
| 233 | # See comment above on bash hack for $.
|
| 234 | ret.append(r.replace('$', '\\$'))
|
| 235 |
|
| 236 | else:
|
| 237 | raise AssertionError('Invalid token %r %r' % (id_, s))
|
| 238 |
|
| 239 | # mismatched brackets, see https://github.com/oilshell/oil/pull/256
|
| 240 | if non_printing != 0:
|
| 241 | return _ERROR_FMT % _UNBALANCED_ERROR
|
| 242 |
|
| 243 | return ''.join(ret)
|
| 244 |
|
| 245 | def EvalPrompt(self, s):
|
| 246 | # type: (str) -> str
|
| 247 | """Perform the two evaluations that bash does.
|
| 248 |
|
| 249 | Used by $PS1 and ${x@P}.
|
| 250 | """
|
| 251 |
|
| 252 | # Parse backslash escapes (cached)
|
| 253 | tokens = self.tokens_cache.get(s)
|
| 254 | if tokens is None:
|
| 255 | tokens = match.Ps1Tokens(s)
|
| 256 | self.tokens_cache[s] = tokens
|
| 257 |
|
| 258 | # Replace values.
|
| 259 | ps1_str = self._ReplaceBackslashCodes(tokens)
|
| 260 |
|
| 261 | # Parse it like a double-quoted word (cached). TODO: This could be done on
|
| 262 | # mem.SetValue(), so we get the error earlier.
|
| 263 | # NOTE: This is copied from the PS4 logic in Tracer.
|
| 264 | ps1_word = self.parse_cache.get(ps1_str)
|
| 265 | if ps1_word is None:
|
| 266 | w_parser = self.parse_ctx.MakeWordParserForPlugin(ps1_str)
|
| 267 | try:
|
| 268 | ps1_word = w_parser.ReadForPlugin()
|
| 269 | except error.Parse as e:
|
| 270 | ps1_word = word_.ErrorWord("<ERROR: Can't parse PS1: %s>" %
|
| 271 | e.UserErrorString())
|
| 272 | self.parse_cache[ps1_str] = ps1_word
|
| 273 |
|
| 274 | # Evaluate, e.g. "${debian_chroot}\u" -> '\u'
|
| 275 | val2 = self.word_ev.EvalForPlugin(ps1_word)
|
| 276 | return val2.s
|
| 277 |
|
| 278 | def EvalFirstPrompt(self):
|
| 279 | # type: () -> str
|
| 280 |
|
| 281 | # First try calling renderPrompt()
|
| 282 | UP_func_val = self.mem.GetValue('renderPrompt')
|
| 283 | if UP_func_val.tag() == value_e.Func:
|
| 284 | func_val = cast(value.Func, UP_func_val)
|
| 285 |
|
| 286 | assert self.global_io is not None
|
| 287 | pos_args = [self.global_io] # type: List[value_t]
|
| 288 | val = self.expr_ev.PluginCall(func_val, pos_args)
|
| 289 |
|
| 290 | UP_val = val
|
| 291 | with tagswitch(val) as case:
|
| 292 | if case(value_e.Str):
|
| 293 | val = cast(value.Str, UP_val)
|
| 294 | return val.s
|
| 295 | else:
|
| 296 | msg = 'renderPrompt() should return Str, got %s' % ui.ValType(
|
| 297 | val)
|
| 298 | return _ERROR_FMT % msg
|
| 299 |
|
| 300 | # Now try evaluating $PS1
|
| 301 | ps1_val = self.mem.env_config.GetVal('PS1')
|
| 302 | #log('ps1_val %s', ps1_val)
|
| 303 | UP_ps1_val = ps1_val
|
| 304 | if UP_ps1_val.tag() == value_e.Str:
|
| 305 | ps1_val = cast(value.Str, UP_ps1_val)
|
| 306 | return self.EvalPrompt(ps1_val.s)
|
| 307 | else:
|
| 308 | return '' # e.g. if the user does 'unset PS1'
|
| 309 |
|
| 310 |
|
| 311 | PROMPT_COMMAND = 'PROMPT_COMMAND'
|
| 312 |
|
| 313 |
|
| 314 | class UserPlugin(object):
|
| 315 | """For executing PROMPT_COMMAND and caching its parse tree.
|
| 316 |
|
| 317 | Similar to core/dev.py:Tracer, which caches $PS4.
|
| 318 | """
|
| 319 |
|
| 320 | def __init__(self, mem, parse_ctx, cmd_ev, errfmt):
|
| 321 | # type: (Mem, ParseContext, cmd_eval.CommandEvaluator, ui.ErrorFormatter) -> None
|
| 322 | self.mem = mem
|
| 323 | self.parse_ctx = parse_ctx
|
| 324 | self.cmd_ev = cmd_ev
|
| 325 | self.errfmt = errfmt
|
| 326 |
|
| 327 | self.arena = parse_ctx.arena
|
| 328 | self.parse_cache = {} # type: Dict[str, command_t]
|
| 329 |
|
| 330 | def Run(self):
|
| 331 | # type: () -> None
|
| 332 | val = self.mem.GetValue(PROMPT_COMMAND)
|
| 333 | if val.tag() != value_e.Str:
|
| 334 | return
|
| 335 |
|
| 336 | # PROMPT_COMMAND almost never changes, so we try to cache its parsing.
|
| 337 | # This avoids memory allocations.
|
| 338 | prompt_cmd = cast(value.Str, val).s
|
| 339 | node = self.parse_cache.get(prompt_cmd)
|
| 340 | if node is None:
|
| 341 | line_reader = reader.StringLineReader(prompt_cmd, self.arena)
|
| 342 | c_parser = self.parse_ctx.MakeOshParser(line_reader)
|
| 343 |
|
| 344 | # NOTE: This is similar to CommandEvaluator.ParseTrapCode().
|
| 345 | src = source.Variable(PROMPT_COMMAND, loc.Missing)
|
| 346 | with alloc.ctx_SourceCode(self.arena, src):
|
| 347 | try:
|
| 348 | node = main_loop.ParseWholeFile(c_parser)
|
| 349 | except error.Parse as e:
|
| 350 | self.errfmt.PrettyPrintError(e)
|
| 351 | return # don't execute
|
| 352 |
|
| 353 | self.parse_cache[prompt_cmd] = node
|
| 354 |
|
| 355 | # Save this so PROMPT_COMMAND can't set $?
|
| 356 | with state.ctx_Registers(self.mem):
|
| 357 | # Catches fatal execution error
|
| 358 | self.cmd_ev.ExecuteAndCatch(node, 0)
|