rc.xsh 9.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. $VI_MODE = True
  2. $AUTO_PUSHD = True
  3. $XONSH_AUTOPAIR = True
  4. # tab selection: do not execute cmd when pressing enter
  5. $COMPLETIONS_CONFIRM = True
  6. $_Z_EXCLUDE_DIRS = ['/tmp']
  7. xontrib load vox z
  8. def _last_exit_status():
  9. try:
  10. exit_status = __xonsh_history__.rtns[-1]
  11. return exit_status if exit_status != 0 else None
  12. except IndexError:
  13. return None
  14. $PROMPT_FIELDS['last_exit_status'] = _last_exit_status
  15. $SHLVL = int($SHLVL) + 1 if 'SHLVL' in ${...} else 1
  16. $XONSH_STDERR_PREFIX = '{RED}'
  17. $XONSH_STDERR_POSTFIX = '{NO_COLOR}'
  18. $DYNAMIC_CWD_WIDTH = '30%'
  19. $DYNAMIC_CWD_ELISION_CHAR = '…'
  20. $PROMPT = ''.join([
  21. '{RED}{last_exit_status:[{}] }',
  22. '{BOLD_GREEN}{user}@{hostname} ',
  23. '{YELLOW}{cwd} ',
  24. '{{BLUE}}{} '.format('{prompt_end}' * $SHLVL),
  25. '{NO_COLOR}',
  26. ])
  27. $RIGHT_PROMPT = '{gitstatus}{env_name: {}}'
  28. $XONSH_APPEND_NEWLINE = True
  29. import datetime as dt
  30. import io
  31. import os
  32. import re
  33. import shutil
  34. import stat
  35. import subprocess
  36. import sys
  37. os.umask(stat.S_IWGRP | stat.S_IRWXO) # 027
  38. # default locale
  39. # will be used for all non-explicitly set LC_* variables
  40. $LANG = 'en_US.UTF-8'
  41. # fallback locales
  42. # GNU gettext gives preference to LANGUAGE over LC_ALL and LANG
  43. # for the purpose of message handling
  44. # https://www.gnu.org/software/gettext/manual/html_node/The-LANGUAGE-variable.html
  45. # cave: if this list contains 'de(_.*)?' at any (sic!) position
  46. # vim 7.4.1689 will switch to german
  47. $LANGUAGE = ':'.join(['en_US', 'en'])
  48. $LC_COLLATE = 'C.UTF-8'
  49. # char classification, case conversion & other char attrs
  50. $LC_CTYPE = 'de_AT.UTF-8'
  51. # $ locale currency_symbol
  52. $LC_MONETARY = 'de_AT.UTF-8'
  53. # $ locale -k LC_NUMERIC | head -n 3
  54. # decimal_point="."
  55. # thousands_sep=""
  56. # grouping=-1
  57. $LC_NUMERIC = 'C.UTF-8'
  58. # A4
  59. $LC_PAPER = 'de_AT.UTF-8'
  60. USER_BIN_PATH = os.path.join($HOME, '.local', 'bin')
  61. if os.path.isdir(USER_BIN_PATH):
  62. $PATH.insert(0, USER_BIN_PATH)
  63. $PAGER = 'less'
  64. $EDITOR = 'vim'
  65. # i3-sensible-terminal
  66. $TERMINAL = 'termite'
  67. # required by pinentry-tty when using gpg command:
  68. $GPG_TTY = $(tty)
  69. if shutil.which('gpgconf'):
  70. # required by scute
  71. $GPG_AGENT_INFO = $(gpgconf --list-dir agent-socket).rstrip() + ':0:1'
  72. if not 'SSH_CLIENT' in ${...}:
  73. # in gnupg 2.1.13 the location of agents socket changed
  74. $SSH_AUTH_SOCK = $(gpgconf --list-dir agent-ssh-socket).rstrip()
  75. # wrapper for termite required when launching termite from ranger:
  76. $TERMCMD = os.path.join(os.path.dirname(__file__), 'ranger-termite-termcmd')
  77. # https://docs.docker.com/engine/security/trust/content_trust/
  78. $DOCKER_CONTENT_TRUST = 1
  79. class DockerImage:
  80. def __init__(self, image):
  81. import json
  82. attrs, = json.loads(subprocess.check_output(['sudo', 'docker', 'image', 'inspect', image])
  83. .decode(sys.stdout.encoding))
  84. self._id = attrs['Id']
  85. self._tags = attrs['RepoTags']
  86. def __repr__(self):
  87. return '{}(id={!r}, tags={!r})'.format(type(self).__name__, self._id, self._tags)
  88. @classmethod
  89. def build(cls, dockerfile_or_path):
  90. out = io.BytesIO()
  91. with StdoutTee(out) as tee:
  92. if os.path.exists(dockerfile_or_path):
  93. p = subprocess.Popen(['sudo', 'docker', 'build', dockerfile_or_path],
  94. stdin=None, stdout=tee)
  95. else:
  96. p = subprocess.Popen(['sudo', 'docker', 'build', '-'],
  97. stdin=subprocess.PIPE, stdout=tee)
  98. p.stdin.write(dockerfile_or_path.encode())
  99. p.stdin.close()
  100. assert p.wait() == 0, 'docker build failed'
  101. image_id, = re.search(rb'^Successfully built (\S+)$', out.getvalue(), re.MULTILINE).groups()
  102. return cls(image_id.decode(sys.stdout.encoding))
  103. @classmethod
  104. def pull(cls, image):
  105. out = io.BytesIO()
  106. with StdoutTee(out) as tee:
  107. subprocess.run(['sudo', 'docker', 'image', 'pull', image], stdout=tee)
  108. repo_digest, = re.search(rb'^Digest: (sha\S+:\S+)$', out.getvalue(), re.MULTILINE).groups()
  109. return cls('{}@{}'.format(image, repo_digest.decode()))
  110. def run(self, args=[], name=None, detach=False, env={},
  111. network=None, publish_ports=[], volumes=[], caps=[]):
  112. params = ['sudo', 'docker', 'run', '--rm']
  113. if name:
  114. params.extend(['--name', name])
  115. if detach:
  116. params.append('--detach')
  117. else:
  118. params.extend(['--interactive', '--tty'])
  119. params.extend([a for k, v in env.items() for a in ['--env', '{}={}'.format(k, v)]])
  120. if network:
  121. params.extend(['--network', network])
  122. params.extend([a for v in volumes for a in ['--volume', ':'.join(v)]])
  123. params.extend('--publish=' + ':'.join([str(a) for a in p]) for p in publish_ports)
  124. params.extend(['--security-opt=no-new-privileges', '--cap-drop=all'])
  125. params.extend(['--cap-add={}'.format(c) for c in caps])
  126. params.append(self._id)
  127. params.extend(args)
  128. sys.stderr.write('{}\n'.format(shlex_join(params)))
  129. subprocess.run(params, check=True)
  130. class FuseMount:
  131. def __init__(self, mount_arg_patterns, mount_point_path, cd=False):
  132. self._mount_arg_patterns = mount_arg_patterns
  133. self._mount_point_path = mount_point_path
  134. self._cd = cd
  135. def __enter__(self):
  136. import shlex
  137. mount_args = [a.format(mp=shlex.quote(self._mount_point_path))
  138. for a in self._mount_arg_patterns]
  139. sys.stderr.write('{}\n'.format(shlex_join(mount_args)))
  140. subprocess.check_call(mount_args)
  141. if self._cd:
  142. self._previous_work_dir_path = os.getcwd()
  143. os.chdir(self._mount_point_path)
  144. return self._mount_point_path
  145. def __exit__(self, exc_type, exc_value, traceback):
  146. umount_args = ['fusermount', '-u', '-z', self._mount_point_path]
  147. sys.stderr.write('{}\n'.format(shlex_join(umount_args)))
  148. subprocess.check_call(umount_args)
  149. if self._cd:
  150. os.chdir(self._previous_work_dir_path)
  151. class EncFSMount(FuseMount):
  152. def __init__(self, root_dir_path, mount_point_path, cd=False, extpass=None):
  153. mount_arg_patterns = ['encfs', root_dir_path, mount_point_path]
  154. if extpass:
  155. mount_arg_patterns.extend(['--extpass', shlex_join(extpass)])
  156. super().__init__(
  157. mount_arg_patterns=mount_arg_patterns,
  158. mount_point_path=mount_point_path,
  159. cd=cd,
  160. )
  161. class StdoutTee:
  162. def __init__(self, sink):
  163. self._sink = sink
  164. def __enter__(self):
  165. self._read_fd, self._write_fd = os.pipe()
  166. import threading
  167. self._thread = threading.Thread(target=self._loop)
  168. self._thread.start()
  169. return self
  170. def _loop(self):
  171. while True:
  172. try:
  173. data = os_read_non_blocking(self._read_fd)
  174. except OSError: # fd closed
  175. return
  176. if data:
  177. self._sink.write(data)
  178. sys.stdout.buffer.write(data)
  179. sys.stdout.flush()
  180. def fileno(self):
  181. return self._write_fd
  182. def __exit__(self, exc_type, exc_value, traceback):
  183. os.close(self._read_fd)
  184. os.close(self._write_fd)
  185. self._thread.join()
  186. def dpkg_listfiles(pkg_name):
  187. assert isinstance(pkg_name, str)
  188. paths = $(dpkg --listfiles @(pkg_name)).split('\n')[:-1]
  189. assert len(paths) > 0, 'pkg {!r} not installed'.format(pkg_name)
  190. return paths
  191. def dpkg_search(path_search_pattern):
  192. assert isinstance(path_search_pattern, str)
  193. return re.findall(
  194. '^(\S+): (.*)$\n',
  195. $(dpkg --search @(path_search_pattern)),
  196. flags=re.MULTILINE,
  197. )
  198. def dpkg_welse(cmd):
  199. pkg_name, cmd_path = dpkg_which(cmd)
  200. return dpkg_listfiles(pkg_name)
  201. def dpkg_which(cmd):
  202. cmd_path = shutil.which(cmd)
  203. assert cmd_path, 'cmd {!r} not found'.format(cmd)
  204. matches = dpkg_search(cmd_path)
  205. assert len(matches) != 0, '{!r} not installed via dpkg'.format(cmd_path)
  206. assert len(matches) == 1
  207. return matches[0]
  208. def gpg_decrypt(path, verify=False):
  209. import gpg
  210. with gpg.Context() as gpg_ctx:
  211. with open(path, 'rb') as f:
  212. data, decrypt_result, verify_result = gpg_ctx.decrypt(f, verify=verify)
  213. return data
  214. def locate(*patterns, match_all=True, ignore_case=True):
  215. params = []
  216. if match_all:
  217. params.insert(0, '--all')
  218. if ignore_case:
  219. params.insert(0, '--ignore-case')
  220. return $(locate @(params) -- @(patterns)).split('\n')[:-1]
  221. def os_read_non_blocking(fd, buffer_size_bytes=8*1024, timeout_seconds=0.1):
  222. import select
  223. if fd in select.select([fd], [], [], timeout_seconds)[0]:
  224. return os.read(fd, buffer_size_bytes)
  225. else:
  226. return None
  227. def shlex_join(params):
  228. import shlex
  229. assert isinstance(params, list) or isinstance(params, tuple), params
  230. return ' '.join(shlex.quote(p) for p in params)
  231. def timestamp_now_utc():
  232. return dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)
  233. def timestamp_now_local():
  234. # if called without tz argument astimezone() assumes
  235. # the system local timezone for the target timezone
  236. return timestamp_now_utc().astimezone()
  237. def timestamp_iso_local():
  238. # if called without tz argument astimezone() assumes
  239. # the system local timezone for the target timezone
  240. return timestamp_now_local().strftime('%Y%m%dT%H%M%S%z')
  241. def yaml_load(path):
  242. import yaml
  243. with open(path, 'r') as f:
  244. return yaml.load(f.read())
  245. def yaml_write(path, data):
  246. import yaml
  247. with open(path, 'w') as f:
  248. f.write(yaml.dump(data, default_flow_style=False))
  249. aliases['d'] = ['sudo', 'docker']
  250. aliases['dpkg-welse'] = lambda args: '\n'.join(dpkg_welse(args[0]))
  251. aliases['dpkg-which'] = lambda args: '\t'.join(dpkg_which(args[0]))
  252. aliases['g'] = ['git']
  253. aliases['ll'] = ['ls', '-l', '--all', '--indicator-style=slash',
  254. '--human-readable', '--time-style=long-iso', '--color=auto']
  255. if shutil.which('startx') and $(tty).rstrip() == '/dev/tty1':
  256. startx
  257. # vim: filetype=python