Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_yaml(self, cfg='./goodluck/test/default.yaml', name=None, exit=True, wait=False, v=True, vv=True):
self.v = v
self.vv = vv
if self.v or self.vv:
self.logger.vinfo()
assert os.path.exists(cfg), "The configuration doesn't exist"
with open(cfg) as f:
exp_dict = yaml.load(f)
session_name = name if name else cfg.split('/')[-1].replace('.', '_')
server = libtmux.Server()
# import pdb;pdb.set_trace()
session = get_session(server, session_name)
for i, (exp_name, kwargs) in enumerate(exp_dict.items()):
if i==0:
session.attached_window.rename_window(exp_name) #Rename the name of window 0
else:
window = session.new_window(attach=True, window_name=exp_name)
pane = session.attached_pane
ssh_command = self.get_command(**kwargs)
pane.send_keys(ssh_command)
if exit:
print("The opened session will be closed.")
time.sleep(5)
server.kill_session(session_name)
def kill_tmux_session(session_name: str):
"""Find and kill this session"""
server = libtmux.Server()
if server and server.has_session(session_name):
logger.info(
'%s Killing the running tmux session "%s" ...'
% (settings.TERM_LBL, session_name)
)
call(["tmux", "kill-session", "-t", session_name])
def get_slash_tmux_session(session_name):
try:
tmux_server = libtmux.Server()
return tmux_server.find_where({"session_name":session_name})
except libtmux.exc.LibTmuxException:
_logger.debug('No tmux server is running')
return
import libtmux
import time
server = libtmux.Server()
session = server.new_session(session_name="atari")
window = session.new_window()
pane = window.panes[0]
pane.split_window(vertical=False)
pane.send_keys("cd /data/lyj/rlpack")
time.sleep(0.3)
pane.send_keys("conda activate gpu")
time.sleep(0.3)
pane.send_keys(f"python examples/distributed_ramatari/run_dqn.py --gpu 0")
pane1 = window.panes[1]
pane1.send_keys("1")
def dev(dev_target, stage):
# # first check dynamodb install
# if dev_target in {'api', 'all'}:
# # we need to ensure dynamodb is installed for API dev (note: currently not _actually_ required for anything... -- 9/9/2018)
# must_run("cd packages/api && node_modules/.bin/sls dynamodb install")
# `--stage` has special meaning with dev cmd (emulates a stage via api points (UI))
os.environ["STAGE"] = "dev"
import libtmux
api_port = 52700
sam_port = 52701
ui_port = 52710
TMP_SESSION = 'tmp-session'
sess_name = f"dev-{int(time.time())}"
server = libtmux.Server(socket_name='flux-app-tmux-session')
# server.cmd('set -g destroy-unattached off')
server.cmd('set-option -g default-shell /bin/bash')
session = server.new_session(session_name=sess_name, start_directory='./', window_command="sleep 1")
session.set_option('mouse', True)
# session.set_option('destroy-unattached', 'off')
# session.set_option('remain-on-exit', 'on')
def kill_sessions():
with suppress(Exception):
try:
for s in server.list_sessions():
print(s)
print(s.show_option('default-shell'))
s.kill_session()
except Exception as e:
def fetch_pane(session_name, window_name):
try:
server = libtmux.Server()
local_session = server.new_session(session_name)
window = local_session.attached_window
window.rename_window(window_name)
except TmuxSessionExists:
local_session = server.find_where({'session_name': session_name})
if local_session.find_where({'window_name': window_name}):
local_session.kill_session()
message = 'Window named {0} exists in tmux session named {1}'.format(
window_name, session_name)
six.raise_from(TmuxWindowExists(message), None)
else:
window = local_session.new_window(window_name)
return window.attached_pane
def __init__(self,
start_dir=None,
preamble_cmd=None,
verbose=True,
dry_run=False):
"""
Args:
start_dir:
preamble_cmd: execute before every run(), can be overriden in run arg
can be either a string or a list of strings to execute in order
verbose:
dry_run:
"""
self._server = libtmux.Server()
self._start_dir = os.path.expanduser(start_dir)
self._verbose = verbose
self._dry_run = dry_run
self._preamble_cmd = preamble_cmd
# two-level dict of session:window:cmd
self.records = {}
if self._dry_run:
print('TmuxExecutor: dry run.')
def start_gdb_in_tmux_pane(command):
import libtmux
server = libtmux.Server()
if server is None:
raise Exception("Tmux server not found")
sessions = server.list_sessions()
if len(sessions) != 1:
raise Exception("There should be only one tmux session running")
session = server.list_sessions()[0]
window = session.attached_window # type: libtmux.Window
pane = window.attached_pane
# Note: multiply by two since most monospace fonts are taller than wide
vertical = int(pane.height) * 2 > int(pane.width)
self.verbose_print("Current window h =", window.height, "w =", window.width)
self.verbose_print("Current pane h =", window.attached_pane.height, "w =", window.attached_pane.width)
if self.config.pretend:
self.info("Would have split current tmux pane", "vertically." if vertical else "horizontally.")
self.info("Would have run", coloured(AnsiColour.yellow, command), "in new pane.")
else: