Coverage for src/lcdoc/lprunner.py: 24.57%
Shortcuts on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Runs the literate programming blocks within a markdown document.
4We open tmux (requirement) and show in one pane the markdown plus the commands to run.
5If the commands open a session themselves, we attach in split panes, so that you can
6follow the output.
8"""
10import os lppytest
11import sys lppytest
12import time lppytest
13import traceback lppytest
14import httpx lppytest
16from lcdoc.const import lprunner_sep as sep lppytest
17from lcdoc.mkdocs.tools import split_off_fenced_blocks lppytest
18from lcdoc.tools import app, dirname, exists, read_file, write_file lppytest
20env = os.environ lppytest
23class FLG: lppytest
24 class tmx_ctrl_color: lppytest
25 n = 'Color of the controlling TMUX pane (fg/bg)' lppytest
26 v = '7;16' lppytest
27 s = 'cc' lppytest
29 class tmx_lp_color: lppytest
30 n = 'Color of the LP TMUX pane(s)' lppytest
31 v = '243;17' lppytest
32 s = 'cl' lppytest
34 class confirm_all: lppytest
35 n = 'Assume yes on all questions' lppytest
36 v = False lppytest
37 s = 'y' lppytest
40class S: lppytest
41 # fmt:off
42 auto_confirm_all = False lppytest
43 exit_after_run = False lppytest
44 fn = None lppytest
45 help_shown = False lppytest
46 # fmt:on
49a = sys.argv lppytest
50# we are started like this inside tmux:
51if len(a) > 1 and a[1] == 'TMXID': 51 ↛ 52line 51 didn't jump to line 52, because the condition on line 51 was never truelppytest
52 our_tmx_id = a[2]
53 a.pop(1)
54 a.pop(1)
55else:
56 our_tmx_id = str(os.getpid()) lppytest
57OurTmux = 'tmux -S /tmp/lprunner.tmux.%s ' % our_tmx_id lppytest
58hl = lambda msg: '\x1b[32m%s\x1b[0m' % msg 58 ↛ exitline 58 didn't run the lambda on line 58lppytest
61def restart_in_tmux(): lppytest
62 if env.get('TMUX'):
63 del env['TMUX'] # otherwise we could not attach to lp tmux from inside
64 return # we are started inside
65 cmd = ['"%s"' % s for s in sys.argv]
66 cmd.insert(1, str(os.getpid()))
67 cmd.insert(1, 'TMXID')
68 err = os.system(OurTmux + 'new -s lprunner ' + ' '.join(cmd))
69 sys.exit(err)
72class Unconfirmed(Exception): lppytest
73 pass lppytest
76have_sessions = set() lppytest
79def show_tmux(session_name): lppytest
80 """
81 In lp a tmux session was created or is attached to
82 We have our session, and attach to that in a split pane,
83 then get focus back to this app here
84 """
85 s = session_name
86 if s in have_sessions:
87 return
88 # found no other way to get focus back to original pane:
89 os.system('( sleep 0.1 && %s select-pane -t 1 ) &' % OurTmux)
90 cmd = OurTmux + 'split-window -h \'unset TMUX; tmux att -t "%s"\'' % s
91 os.system(cmd)
92 set_tmx_col('tmux', FLG.tmx_lp_color)
93 time.sleep(0.2)
94 have_sessions.add(s)
97def set_tmx_col(tmx, col): lppytest
98 c = col.v
99 if not c:
100 return
101 if ';' in c:
102 s = 'fg=colour%s,bg=colour%s' % tuple(c.split(';', 1))
103 else:
104 s = 'bg=colour%s' % c
105 os.system(tmx + ' set -g window-style %s' % s)
108help_msg = 'a:yes for all e:yes for all, then exit q:quit s:shell y:confirm' lppytest
111def confirm(msg): lppytest
112 if S.auto_confirm_all:
113 return
114 print(hl(msg))
115 try:
116 if not S.help_shown:
117 os.system(OurTmux + 'display-message -p "%s"' % help_msg)
118 S.help_shown = True
119 r = input('[a|e|h|q|s|Y(default)] ? ')
120 except KeyboardInterrupt as ex:
121 r = 'q'
122 r = r.lower()
123 print()
124 if r == 's':
125 print('entering shell - ctrl+d to exit')
126 os.system('bash')
127 return confirm(msg)
128 if r == 'a':
129 S.auto_confirm_all = True
130 elif r == 'e':
131 S.auto_confirm_all = True
132 S.exit_after_run = True
133 elif r in ('q', 'n'):
134 raise Unconfirmed()
137def extract_md_src_from_url(url): lppytest
138 a = url
139 a = (a + '/') if not a.endswith('/') else a
140 a += 'runner.md'
141 src = httpx.get(a).text
142 S.fn = a.split('/', 3)[3].replace('/', '_')
143 return src
146def exit_help(flgs, shts): lppytest
147 print(__doc__)
148 for k, v in flgs.items():
149 print('%4s | %s: %s [%s]' % ('-' + v.s, k.ljust(18), v.n, v.v))
151 sys.exit(0)
154def chck_book(flg, flags): lppytest
155 if not flags[flg].v in (True, False):
156 return False
157 flags[flg].v = not flags[flg].v
158 return True
161def parse_cli(g=getattr): lppytest
162 args = list(sys.argv[1:])
163 flgs = {'--' + k: g(FLG, k) for k in dir(FLG) if not k.startswith('_')}
164 shts = {'-' + v.s: v for k, v in flgs.items()}
165 if '-h' in args or '--help' in args or not args:
166 exit_help(flgs, shts)
167 while args:
168 a = args.pop(0)
169 if a in flgs:
170 if not chck_book(a, flgs):
171 flgs[a].v = args.pop(0)
172 elif a in shts:
173 if not chck_book(a, shts):
174 shts[a].v = args.pop(0)
175 elif exists(a):
176 S.fn = a
177 elif a.startswith('http'):
178 src = extract_md_src_from_url(a)
179 write_file(S.fn, src)
180 S.fn = os.path.abspath(S.fn)
181 app.name = os.path.basename(S.fn)
182 S.auto_confirm_all = FLG.confirm_all.v
183 from lcdoc import log
185 f = open('/tmp/tmux.log.%s' % our_tmx_id, 'w')
186 w = lambda msg, f=f: f.write(msg + '\n')
187 log.outputter[0] = w
190class mock: lppytest
191 def page(): lppytest
192 class page:
193 src_path = abs_src_path = S.fn
194 stats = {}
196 return page
198 def config(): lppytest
199 return {'docs_dir': dirname(S.fn)}
202def show_md(md): lppytest
203 md = [md] if isinstance(md, str) else md
204 for m in md:
205 print(m)
208def run_lp(spec): lppytest
209 src = spec['kwargs'].get('source', '')
210 show_md(src)
211 if not spec['kwargs'].get('runner'):
212 return
213 try:
214 res = LP.run_block(spec)
215 except Unconfirmed as ex:
216 print('unconfirmed - bye')
217 sys.exit(1)
220def run_fn(): lppytest
221 markdown = read_file(S.fn)
222 LP.page = mock.page()
223 LP.config = mock.config()
224 LP.previous_results = {}
225 LP.cur_results = {}
226 LP.fn_lp = S.fn
227 mds, lp_blocks = split_off_fenced_blocks(
228 markdown, fc_crit=LP.is_lp_block, fc_process=LP.parse_lp_block
229 )
230 l = lp_blocks
231 while mds:
232 show_md(mds.pop(0))
233 run_lp(lp_blocks.pop(0)) if lp_blocks else None
236def setup_our_tmux(): lppytest
237 set_tmx_col(OurTmux, FLG.tmx_ctrl_color)
240def main(): lppytest
241 if '-h' in sys.argv or '--help' in sys.argv or len(sys.argv) == 1:
242 parse_cli() # will exit
243 restart_in_tmux()
244 try:
245 run_in_tmux()
246 except SystemExit as ex:
247 print('Tmux runner exitted with Error')
248 breakpoint()
249 print('Hit q to exit')
252def run_in_tmux(): lppytest
253 # we are in a subprocess now, running within tmux:
254 global LP
255 from lcdoc.mkdocs.lp import LP
256 from lcdoc import lp
258 lp.is_lprunner[0] = True
259 try:
260 parse_cli()
261 setup_our_tmux()
262 if not S.fn:
263 app.die('Require filename or URL')
264 run_fn()
265 if not S.exit_after_run:
266 S.auto_confirm_all = False
267 except Exception as ex:
268 traceback.print_exc()
269 print(ex)
270 print('Error - entering debug mode')
271 breakpoint() # FIXME BREAKPOINT
272 print('? for help')
273 sys.exit(1)
274 try:
275 confirm('All done. Exit Tmux.')
276 while True:
277 if os.system('tmux detach'):
278 break
279 os.system(OurTmux + 'kill-session')
280 finally:
281 os.unlink(OurTmux.rsplit(' ', 1)[-1])
284if __name__ == '__main__': 284 ↛ 285line 284 didn't jump to line 285, because the condition on line 284 was never truelppytest
285 main()