OILS / builtin / io_ysh.py View on Github | oilshell.org

256 lines, 153 significant
1#!/usr/bin/env python2
2"""
3builtin/io_ysh.py - YSH builtins that perform I/O
4"""
5from __future__ import print_function
6
7from _devbuild.gen import arg_types
8from _devbuild.gen.runtime_asdl import cmd_value
9from _devbuild.gen.syntax_asdl import command_e, BraceGroup, loc
10from asdl import format as fmt
11from core import error
12from core.error import e_usage
13from core import state
14from core import ui
15from core import vm
16from data_lang import j8
17from frontend import flag_util
18from frontend import match
19from frontend import typed_args
20from mycpp import mylib
21from mycpp.mylib import log
22
23from typing import TYPE_CHECKING, cast
24if TYPE_CHECKING:
25 from core.alloc import Arena
26 from core.ui import ErrorFormatter
27 from osh import cmd_eval
28
29_ = log
30
31
32class _Builtin(vm._Builtin):
33
34 def __init__(self, mem, errfmt):
35 # type: (state.Mem, ErrorFormatter) -> None
36 self.mem = mem
37 self.errfmt = errfmt
38
39
40class Pp(_Builtin):
41 """Given a list of variable names, print their values.
42
43 'pp cell a' is a lot easier to type than 'argv.py "${a[@]}"'.
44 """
45
46 def __init__(self, mem, errfmt, procs, arena):
47 # type: (state.Mem, ErrorFormatter, state.Procs, Arena) -> None
48 _Builtin.__init__(self, mem, errfmt)
49 self.procs = procs
50 self.arena = arena
51 self.stdout_ = mylib.Stdout()
52
53 def Run(self, cmd_val):
54 # type: (cmd_value.Argv) -> int
55 arg, arg_r = flag_util.ParseCmdVal('pp',
56 cmd_val,
57 accept_typed_args=True)
58
59 action, action_loc = arg_r.Peek2()
60
61 # pp (x) prints in the same way that '= x' does
62 # TODO: We also need pp [x], which shows the expression
63 if action is None:
64 rd = typed_args.ReaderForProc(cmd_val)
65 val = rd.PosValue()
66 rd.Done()
67
68 # IOError caught by caller
69 ui.PrettyPrintValue('', val, mylib.Stdout())
70 return 0
71
72 arg_r.Next()
73
74 # Actions that print unstable formats start with '.'
75 if action == 'cell':
76 argv, locs = arg_r.Rest2()
77
78 status = 0
79 for i, name in enumerate(argv):
80 if name.startswith(':'):
81 name = name[1:]
82
83 if not match.IsValidVarName(name):
84 raise error.Usage('got invalid variable name %r' % name,
85 locs[i])
86
87 cell = self.mem.GetCell(name)
88 if cell is None:
89 self.errfmt.Print_("Couldn't find a variable named %r" %
90 name,
91 blame_loc=locs[i])
92 status = 1
93 else:
94 self.stdout_.write('%s = ' % name)
95 pretty_f = fmt.DetectConsoleOutput(self.stdout_)
96 fmt.PrintTree(cell.PrettyTree(), pretty_f)
97 self.stdout_.write('\n')
98
99 elif action == 'asdl':
100 # TODO: could be pp asdl (x, y, z)
101 rd = typed_args.ReaderForProc(cmd_val)
102 val = rd.PosValue()
103 rd.Done()
104
105 tree = val.PrettyTree()
106 #tree = val.AbbreviatedTree() # I used this to test cycle detection
107
108 # TODO: ASDL should print the IDs. And then they will be
109 # line-wrapped.
110 # The IDs should also be used to detect cycles, and omit values
111 # already printed.
112 #id_str = vm.ValueIdString(val)
113 #f.write(' <%s%s>\n' % (ysh_type, id_str))
114
115 pretty_f = fmt.DetectConsoleOutput(self.stdout_)
116 fmt.PrintTree(tree, pretty_f)
117 self.stdout_.write('\n')
118
119 status = 0
120
121 elif action == 'line':
122 # Print format for unit tests
123
124 # TODO: could be pp line (x, y, z)
125 rd = typed_args.ReaderForProc(cmd_val)
126 val = rd.PosValue()
127 rd.Done()
128
129 if ui.TypeNotPrinted(val):
130 ysh_type = ui.ValType(val)
131 self.stdout_.write('(%s) ' % ysh_type)
132
133 j8.PrintLine(val, self.stdout_)
134
135 status = 0
136
137 elif action == 'gc-stats':
138 print('TODO')
139 status = 0
140
141 elif action == 'proc':
142 names, locs = arg_r.Rest2()
143 if len(names):
144 for i, name in enumerate(names):
145 node = self.procs.Get(name)
146 if node is None:
147 self.errfmt.Print_('Invalid proc %r' % name,
148 blame_loc=locs[i])
149 return 1
150 else:
151 names = self.procs.GetNames()
152
153 # TSV8 header
154 print('proc_name\tdoc_comment')
155 for name in names:
156 proc = self.procs.Get(name) # must exist
157 #log('Proc %s', proc)
158 body = proc.body
159
160 # TODO: not just command.ShFunction, but command.Proc!
161 doc = ''
162 if body.tag() == command_e.BraceGroup:
163 bgroup = cast(BraceGroup, body)
164 if bgroup.doc_token:
165 token = bgroup.doc_token
166 # 1 to remove leading space
167 doc = token.line.content[token.col + 1:token.col +
168 token.length]
169
170 # Note: these should be attributes on value.Proc
171 buf = mylib.BufWriter()
172 j8.EncodeString(name, buf, unquoted_ok=True)
173 buf.write('\t')
174 j8.EncodeString(doc, buf, unquoted_ok=True)
175 print(buf.getvalue())
176
177 status = 0
178
179 else:
180 e_usage('got invalid action %r' % action, action_loc)
181
182 return status
183
184
185class Write(_Builtin):
186 """
187 write -- @strs
188 write --sep ' ' --end '' -- @strs
189 write -n -- @
190 write --j8 -- @strs # argv serialization
191 write --j8 --sep $'\t' -- @strs # this is like TSV8
192 """
193
194 def __init__(self, mem, errfmt):
195 # type: (state.Mem, ErrorFormatter) -> None
196 _Builtin.__init__(self, mem, errfmt)
197 self.stdout_ = mylib.Stdout()
198
199 def Run(self, cmd_val):
200 # type: (cmd_value.Argv) -> int
201 attrs, arg_r = flag_util.ParseCmdVal('write', cmd_val)
202 arg = arg_types.write(attrs.attrs)
203 #print(arg)
204
205 i = 0
206 while not arg_r.AtEnd():
207 if i != 0:
208 self.stdout_.write(arg.sep)
209 s = arg_r.Peek()
210
211 if arg.json:
212 s = j8.MaybeEncodeJsonString(s)
213
214 elif arg.j8:
215 s = j8.MaybeEncodeString(s)
216
217 self.stdout_.write(s)
218
219 arg_r.Next()
220 i += 1
221
222 if arg.n:
223 pass
224 elif len(arg.end):
225 self.stdout_.write(arg.end)
226
227 return 0
228
229
230class Fopen(vm._Builtin):
231 """fopen does nothing but run a block.
232
233 It's used solely for its redirects.
234 fopen >out.txt { echo hi }
235
236 It's a subset of eval
237 eval >out.txt { echo hi }
238 """
239
240 def __init__(self, mem, cmd_ev):
241 # type: (state.Mem, cmd_eval.CommandEvaluator) -> None
242 self.mem = mem
243 self.cmd_ev = cmd_ev # To run blocks
244
245 def Run(self, cmd_val):
246 # type: (cmd_value.Argv) -> int
247 _, arg_r = flag_util.ParseCmdVal('fopen',
248 cmd_val,
249 accept_typed_args=True)
250
251 cmd = typed_args.OptionalBlock(cmd_val)
252 if not cmd:
253 raise error.Usage('expected a block', loc.Missing)
254
255 unused = self.cmd_ev.EvalCommand(cmd)
256 return 0