Created
April 26, 2014 10:58
-
-
Save apieum/11317165 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
if [[ -z "$TMUXP_STARTED" ]]; then | |
pydev | |
fi |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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)