Skip to content

Instantly share code, notes, and snippets.

@apieum
Created April 26, 2014 10:58
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save apieum/11317165 to your computer and use it in GitHub Desktop.
Save apieum/11317165 to your computer and use it in GitHub Desktop.
if [[ -z "$TMUXP_STARTED" ]]; then
pydev
fi
#!/bin/bash
source tmuxp.bash
export TMUXP_STARTED=0
TMUXP_SESSION=`basename $0`
TMUXP_EXEC=~/bin/tmuxp_session.py
function close_session {
$TMUXP_EXEC freeze $TMUXP_SESSION
$TMUXP_EXEC kill-session $TMUXP_SESSION
unset TMUXP_STARTED
}
trap close_session EXIT
$TMUXP_EXEC load ~/.config/tmuxp/$TMUXP_SESSION.json
#!/bin/env python
# -*- coding: utf-8 -*-
"""Command line tool for managing tmux workspaces and tmuxp configurations.
tmuxp.cli
~~~~~~~~~
"""
from __future__ import absolute_import, division, print_function, \
with_statement, unicode_literals
import os
import sys
import argparse
import re
import logging
from distutils.util import strtobool
import argcomplete
import kaptan
from tmuxp import log, util, exc, WorkspaceBuilder, Server, config
from tmuxp._compat import ascii_lowercase, input, string_types
logger = logging.getLogger(__name__)
config_dir = os.path.expanduser('~/.config/tmuxp/')
cwd_dir = os.getcwd() + '/'
class ConfigFileCompleter(argcomplete.completers.FilesCompleter):
"""argcomplete completer for tmuxp files."""
def __call__(self, prefix, **kwargs):
completion = argcomplete.completers.FilesCompleter.__call__(
self, prefix, **kwargs
)
completion += [os.path.join(config_dir, c)
for c in config.in_dir(config_dir)]
return completion
def SessionCompleter(prefix, parsed_args, **kwargs):
"""Return list of session names for argcomplete completer."""
t = Server(
socket_name=parsed_args.socket_name,
socket_path=parsed_args.socket_path
)
sessions_available = [
s.get('session_name') for s in t._sessions
if s.get('session_name').startswith(' '.join(prefix))
]
if parsed_args.session_name and sessions_available:
return []
return [
s.get('session_name') for s in t._sessions
if s.get('session_name').startswith(prefix)
]
def setup_logger(logger=None, level='INFO'):
"""Setup logging for CLI use.
:param logger: instance of logger
:type logger: :py:class:`Logger`
"""
if not logger:
logger = logging.getLogger()
if not logger.handlers:
channel = logging.StreamHandler()
channel.setFormatter(log.DebugLogFormatter())
# channel.setFormatter(log.LogFormatter())
logger.setLevel(level)
logger.addHandler(channel)
def startup(config_dir):
"""Initialize CLI.
:param config_dir: Config directory to search
:type config_dir: string
"""
if not os.path.exists(config_dir):
os.makedirs(config_dir)
def load_workspace(config_file, args):
"""Build config workspace.
:param config_file: full path to config file
:param type: string
"""
sconfig = kaptan.Kaptan()
sconfig = sconfig.import_config(config_file).get()
# expands configurations relative to config / profile file location
sconfig = config.expand(sconfig, os.path.dirname(config_file))
sconfig = config.trickle(sconfig)
t = Server(
socket_name=args.socket_name,
socket_path=args.socket_path,
colors=args.colors
)
try:
builder = WorkspaceBuilder(sconf=sconfig, server=t)
except exc.EmptyConfigException:
logger.error('%s is empty or parsed no config data' % config_file)
return
tmux_bin = util.which('tmux')
try:
logger.info('Loading %s.' % config_file)
builder.build()
if 'TMUX' in os.environ:
tmux_env = os.environ.pop('TMUX')
builder.session.switch_client()
os.environ['TMUX'] = tmux_env
return
if not args.detached:
builder.session.attach_session()
except exc.TmuxSessionExists as e:
if not args.detached:
if 'TMUX' in os.environ:
builder.session.switch_client()
else:
builder.session.attach_session()
return
except exc.TmuxpException as e:
import traceback
logger.error(traceback.format_exc())
logger.error(e)
builder.session.kill_session()
logger.info('Session killed.')
def freeze(session):
"""Freeze live tmux session and Return session config :py:obj:`dict`.
:param session: session object
:type session: :class:`Session`
:rtype: dict
"""
sconf = {}
sconf['session_name'] = session['session_name']
sconf['windows'] = []
for w in session.windows:
wconf = {}
wconf['options'] = w.show_window_options()
wconf['window_name'] = w.get('window_name')
wconf['layout'] = w.get('window_layout')
wconf['panes'] = []
if w.get('window_active', '0') == '1':
wconf['focus']='true'
# If all panes have same path, set 'start_directory' instead
# of using 'cd' shell commands.
pane_has_same_path = lambda p: (
w.panes[0].get('pane_current_path') ==
p.get('pane_current_path')
)
if (all(pane_has_same_path(p) for p in w.panes)):
wconf['start_directory'] = w.panes[0].get('pane_current_path')
for p in w.panes:
pconf = {}
pconf['shell_command'] = []
if 'start_directory' not in wconf:
pconf['shell_command'].append(
'cd ' + p.get('pane_current_path')
)
if p.get('pane_active', '0') == '1':
pconf['focus']='true'
current_cmd = p.get('pane_current_command')
def filter_interpretters_and_shells():
return (
current_cmd.startswith('-') or
any(
current_cmd.endswith(cmd)
for cmd in ['python', 'ruby', 'node', 'bash']
)
)
if (filter_interpretters_and_shells()):
current_cmd = None
if current_cmd:
pconf['shell_command'].append(current_cmd)
else:
if not len(pconf['shell_command']):
pconf = 'pane'
wconf['panes'].append(pconf)
sconf['windows'].append(wconf)
return sconf
def command_freeze(args):
"""Import teamocil config to tmuxp format."""
ctext = ' '.join(args.session_name)
t = Server(
socket_name=args.socket_name,
socket_path=args.socket_path,
colors=args.colors
)
try:
session = t.findWhere({
'session_name': ctext
})
if not session:
raise exc.TmuxpException('Session not found.')
except exc.TmuxpException as e:
logger.error(e)
return
sconf = freeze(session)
configparser = kaptan.Kaptan()
newconfig = config.inline(sconf)
configparser.import_config(newconfig)
newconfig = configparser.export('json', indent=2)
dest = os.path.join(
config_dir,
'%s.%s' % (sconf.get('session_name'), 'json')
)
dest = os.path.abspath(os.path.relpath(os.path.expanduser(dest)))
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
buf = open(dest, 'w')
buf.write(newconfig)
buf.close()
logger.info('Saved to %s.' % dest)
def command_load(args):
"""Load a session from a tmuxp session file."""
if isinstance(args.config, list):
args.config = ' '.join(args.config)
if '.' == args.config:
if config.in_cwd():
configfile = config.in_cwd()[0]
else:
sys.exit('No tmuxp configs found in current directory.')
else:
configfile = args.config
file_user = os.path.join(config_dir, configfile)
file_cwd = os.path.join(cwd_dir, configfile)
if os.path.exists(file_cwd) and os.path.isfile(file_cwd):
logger.info('load config %s' % file_cwd)
load_workspace(file_cwd, args)
elif os.path.exists(file_user) and os.path.isfile(file_user):
load_workspace(file_user, args)
else:
logger.error('%s not found.' % configfile)
def command_kill_session(args):
"""Command to kill a tmux session."""
commands = []
ctext = ' '.join(args.session_name)
t = Server(
socket_name=args.socket_name or None,
socket_path=args.socket_path or None
)
try:
session = next((s for s in t.sessions if s.get(
'session_name') == ctext), None)
if not session:
raise exc.TmuxpException('Session not found.')
except exc.TmuxpException as e:
logger.error(e)
return
try:
session.kill_session()
logger.info("Killed session %s." % ctext)
except exc.TmuxpException as e:
logger.error(e)
def get_parser():
"""Return :py:class:`argparse.ArgumentParser` instance for CLI."""
server_parser = argparse.ArgumentParser(add_help=False)
server_parser.add_argument(
'-L', dest='socket_name',
default=None,
help='socket name of tmux server. Same as tmux.',
metavar='socket-name'
)
server_parser.add_argument(
'-S',
dest='socket_path',
default=None,
help='socket path of tmux server. Same as tmux.',
metavar='socket-path'
)
server_parser.add_argument(
'-y',
dest='answer_yes',
default=None,
help='Always answer yes.',
action='store_true'
)
parser = argparse.ArgumentParser(
description='Launch tmux workspace. '
'Help documentation: <http://tmuxp.rtfd.org>.',
parents=[server_parser]
)
client_parser = argparse.ArgumentParser(add_help=False)
colorsgroup = client_parser.add_mutually_exclusive_group()
colorsgroup.add_argument(
'-2',
dest='colors',
action='store_const',
const=256,
help='Force tmux to assume the terminal supports 256 colours.',
)
colorsgroup.add_argument(
'-8',
dest='colors',
action='store_const',
const=88,
help='Like -2, but indicates that the terminal supports 88 colours.',
)
parser.set_defaults(colors=None)
subparsers = parser.add_subparsers(
title='commands',
description='valid commands',
)
kill_session = subparsers.add_parser(
'kill-session',
parents=[server_parser],
help='Kill tmux session by name.'
)
kill_session.set_defaults(callback=command_kill_session)
kill_session.add_argument(
dest='session_name',
type=str,
nargs='+',
default=None,
help='Name of session',
).completer = SessionCompleter
freeze = subparsers.add_parser(
'freeze',
parents=[server_parser],
help='Create a snapshot of a tmux session and save it to JSON or YAML.'
)
freeze.set_defaults(callback=command_freeze)
freeze.add_argument(
dest='session_name',
type=str,
nargs='+',
help='Name of session',
).completer = SessionCompleter
load = subparsers.add_parser(
'load',
parents=[server_parser, client_parser],
help='Load a configuration from file. Attach the session. If session '
'already exists, offer to attach instead.'
)
load.add_argument(
dest='config',
type=str,
nargs='+',
help='List config available in working directory and config folder.'
).completer = ConfigFileCompleter(
allowednames=('.yaml', '.json'), directories=False
)
load.set_defaults(callback=command_load)
load.add_argument(
'-d',
dest='detached',
default=None,
help='Load a session without attaching to it.',
action='store_true'
)
return parser
def main():
"""Main CLI application."""
parser = get_parser()
argcomplete.autocomplete(parser, always_complete_options=False)
args = parser.parse_args()
setup_logger(
level=args.log_level.upper() if 'log_level' in args else 'INFO'
)
try:
util.has_required_tmux_version()
except exc.TmuxpException as e:
logger.error(e)
sys.exit()
util.oh_my_zsh_auto_title()
t = Server(
socket_name=args.socket_name,
socket_path=args.socket_path,
colors=args.colors
)
try:
if not hasattr(args, 'callback'):
parser.print_help()
elif args.callback is command_load:
command_load(args)
elif args.callback is command_freeze:
command_freeze(args)
elif args.callback is command_kill_session:
command_kill_session(args)
except KeyboardInterrupt:
pass
if __name__ == '__main__':
"""Assure tmuxp is in python path's and available as a package. """
base = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, base)
main()
if exit:
sys.exit(0)
@apieum
Copy link
Author

apieum commented Apr 26, 2014

Need to install tmuxp: pip install tmuxp

Code in tmuxp_session.py is crapy (mostly copied from tmuxp.cli and tmuxp.workspacebuilder.freeze without prompts)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment