| 1 | """
|
| 2 | builtin/pure_ysh.py - YSH builtins that don't do I/O.
|
| 3 | """
|
| 4 | from __future__ import print_function
|
| 5 |
|
| 6 | from _devbuild.gen.runtime_asdl import cmd_value
|
| 7 | from _devbuild.gen.syntax_asdl import command_t, loc, loc_t
|
| 8 | from _devbuild.gen.value_asdl import value, value_e, value_t
|
| 9 | from core import error
|
| 10 | from core import state
|
| 11 | from core import vm
|
| 12 | from frontend import flag_util
|
| 13 | from frontend import typed_args
|
| 14 | from mycpp.mylib import tagswitch
|
| 15 |
|
| 16 | from typing import TYPE_CHECKING, cast, Any, Dict, List
|
| 17 | if TYPE_CHECKING:
|
| 18 | from display import ui
|
| 19 | from osh.cmd_eval import CommandEvaluator
|
| 20 |
|
| 21 |
|
| 22 | class ctx_Context(object):
|
| 23 | """For ctx push (context) { ... }"""
|
| 24 |
|
| 25 | def __init__(self, mem, context):
|
| 26 | # type: (state.Mem, Dict[str, value_t]) -> None
|
| 27 | self.mem = mem
|
| 28 | self.mem.PushContextStack(context)
|
| 29 |
|
| 30 | def __enter__(self):
|
| 31 | # type: () -> None
|
| 32 | pass
|
| 33 |
|
| 34 | def __exit__(self, type, value, traceback):
|
| 35 | # type: (Any, Any, Any) -> None
|
| 36 | self.mem.PopContextStack()
|
| 37 |
|
| 38 |
|
| 39 | class Ctx(vm._Builtin):
|
| 40 |
|
| 41 | def __init__(self, mem, cmd_ev):
|
| 42 | # type: (state.Mem, CommandEvaluator) -> None
|
| 43 | self.mem = mem
|
| 44 | self.cmd_ev = cmd_ev # To run blocks
|
| 45 |
|
| 46 | def _GetContext(self):
|
| 47 | # type: () -> Dict[str, value_t]
|
| 48 | ctx = self.mem.GetContext()
|
| 49 | if ctx is None:
|
| 50 | raise error.Expr(
|
| 51 | "Could not find a context. Did you forget to 'ctx push'?",
|
| 52 | loc.Missing)
|
| 53 | return ctx
|
| 54 |
|
| 55 | def _Push(self, context, block):
|
| 56 | # type: (Dict[str, value_t], command_t) -> int
|
| 57 | with ctx_Context(self.mem, context):
|
| 58 | return self.cmd_ev.EvalCommand(block)
|
| 59 |
|
| 60 | def _Set(self, updates):
|
| 61 | # type: (Dict[str, value_t]) -> int
|
| 62 | ctx = self._GetContext()
|
| 63 | ctx.update(updates)
|
| 64 | return 0
|
| 65 |
|
| 66 | def _Emit(self, field, item, blame):
|
| 67 | # type: (str, value_t, loc_t) -> int
|
| 68 | ctx = self._GetContext()
|
| 69 |
|
| 70 | if field not in ctx:
|
| 71 | ctx[field] = value.List([])
|
| 72 |
|
| 73 | UP_arr = ctx[field]
|
| 74 | if UP_arr.tag() != value_e.List:
|
| 75 | raise error.TypeErr(
|
| 76 | UP_arr,
|
| 77 | "Expected the context item '%s' to be a List" % (field), blame)
|
| 78 |
|
| 79 | arr = cast(value.List, UP_arr)
|
| 80 | arr.items.append(item)
|
| 81 |
|
| 82 | return 0
|
| 83 |
|
| 84 | def Run(self, cmd_val):
|
| 85 | # type: (cmd_value.Argv) -> int
|
| 86 | rd = typed_args.ReaderForProc(cmd_val)
|
| 87 | _, arg_r = flag_util.ParseCmdVal('ctx',
|
| 88 | cmd_val,
|
| 89 | accept_typed_args=True)
|
| 90 |
|
| 91 | verb, verb_loc = arg_r.ReadRequired2(
|
| 92 | 'Expected a verb (push, set, emit)')
|
| 93 |
|
| 94 | if verb == "push":
|
| 95 | context = rd.PosDict()
|
| 96 | block = rd.RequiredBlock()
|
| 97 | rd.Done()
|
| 98 | arg_r.AtEnd()
|
| 99 |
|
| 100 | return self._Push(context, block)
|
| 101 |
|
| 102 | elif verb == "set":
|
| 103 | updates = rd.RestNamed()
|
| 104 | rd.Done()
|
| 105 | arg_r.AtEnd()
|
| 106 |
|
| 107 | return self._Set(updates)
|
| 108 |
|
| 109 | elif verb == "emit":
|
| 110 | field, field_loc = arg_r.ReadRequired2(
|
| 111 | "A target field is required")
|
| 112 | item = rd.PosValue()
|
| 113 | rd.Done()
|
| 114 | arg_r.AtEnd()
|
| 115 |
|
| 116 | return self._Emit(field, item, field_loc)
|
| 117 |
|
| 118 | else:
|
| 119 | raise error.Usage("Unknown verb '%s'" % verb, verb_loc)
|
| 120 |
|
| 121 |
|
| 122 | class PushRegisters(vm._Builtin):
|
| 123 |
|
| 124 | def __init__(self, mem, cmd_ev):
|
| 125 | # type: (state.Mem, CommandEvaluator) -> None
|
| 126 | self.mem = mem
|
| 127 | self.cmd_ev = cmd_ev # To run blocks
|
| 128 |
|
| 129 | def Run(self, cmd_val):
|
| 130 | # type: (cmd_value.Argv) -> int
|
| 131 | _, arg_r = flag_util.ParseCmdVal('push-registers',
|
| 132 | cmd_val,
|
| 133 | accept_typed_args=True)
|
| 134 |
|
| 135 | cmd = typed_args.OptionalBlock(cmd_val)
|
| 136 | if not cmd:
|
| 137 | raise error.Usage('expected a block', loc.Missing)
|
| 138 |
|
| 139 | with state.ctx_Registers(self.mem):
|
| 140 | unused = self.cmd_ev.EvalCommand(cmd)
|
| 141 |
|
| 142 | # make it "SILENT" in terms of not mutating $?
|
| 143 | # TODO: Revisit this. It might be better to provide the headless shell
|
| 144 | # with a way to SET $? instead. Needs to be tested/prototyped.
|
| 145 | return self.mem.last_status[-1]
|
| 146 |
|
| 147 |
|
| 148 | class Append(vm._Builtin):
|
| 149 | """Push word args onto an List.
|
| 150 |
|
| 151 | Not doing typed args since you can do
|
| 152 |
|
| 153 | :: mylist->append(42)
|
| 154 | """
|
| 155 |
|
| 156 | def __init__(self, mem, errfmt):
|
| 157 | # type: (state.Mem, ui.ErrorFormatter) -> None
|
| 158 | self.mem = mem
|
| 159 | self.errfmt = errfmt
|
| 160 |
|
| 161 | def Run(self, cmd_val):
|
| 162 | # type: (cmd_value.Argv) -> int
|
| 163 |
|
| 164 | # This means we ignore -- , which is consistent
|
| 165 | arg, arg_r = flag_util.ParseCmdVal('append',
|
| 166 | cmd_val,
|
| 167 | accept_typed_args=True)
|
| 168 |
|
| 169 | rd = typed_args.ReaderForProc(cmd_val)
|
| 170 | val = rd.PosValue()
|
| 171 | rd.Done()
|
| 172 |
|
| 173 | UP_val = val
|
| 174 | with tagswitch(val) as case:
|
| 175 | if case(value_e.BashArray):
|
| 176 | val = cast(value.BashArray, UP_val)
|
| 177 | val.strs.extend(arg_r.Rest())
|
| 178 | elif case(value_e.List):
|
| 179 | val = cast(value.List, UP_val)
|
| 180 | typed = [value.Str(s)
|
| 181 | for s in arg_r.Rest()] # type: List[value_t]
|
| 182 | val.items.extend(typed)
|
| 183 | else:
|
| 184 | raise error.TypeErr(val, 'expected List or BashArray',
|
| 185 | loc.Missing)
|
| 186 |
|
| 187 | return 0
|