decoded/dr1p_pymux
decoded
/
dr1p_pymux
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added chat overlay

This commit is contained in:
.[d]. 2022-08-15 19:23:49 -05:00
commit 3275444755
37 changed files with 6060 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
README.md Normal file
View File

@ -0,0 +1,67 @@
## [MAPLE|G1MP]+[ML^N]+[NETSPANNING[i]]
---
```
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMh+MMMMMMMMMMMMMMhsMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMm/ oMMMMMMMMMMMMMMm +NMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMy` yMMMMMMMMMMMMMMM- -mMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMs+dMMMMMMMMMM+ sMMMMMMMMMMMMMMM- `dMMMMMMMMMMms/NMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMM+ .omMMMMMM: -MMMMMMMMMMMMMMo `yMMMMMMMy: `dMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMM- /dMMM+ sMMMMMMMMMMMMh `hMMMNo` sMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMd :dm `mMMMMMMMMMMN. .NNo` .MMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMM: - :MMMMMMMMMMs :` sMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMs ymNMMMMMNm. NMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMy `-/-` .MMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMo .NMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMNh+. :sdMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMhso+:. `-/+syMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMM- dMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMM` `.:+/. `/s+:. sMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMNo -oms. .//-` `:/:` `+md+` .hMMMMMMMMMMMMMMM
MMMMMMMMMMMMMNs` .odNdo. .ohmd+` :dMMMMMMMMMMMMM
MMMMMMMMMMMNo` .. .- :hMMMMMMMMMMM
MMMMMMMMMd+` -sNMMMMMMMM
MMMMMMNs- `.. `/-. `+dMMMMMM
MMMNy: ./sdNMMMh: `sNMMMNds/. .odMMM
MM+ :ymMMMMMMMMMMh. +NMMMMMMMMMMmo- /NM
MMMh: .sNMMMMMMMMMMMMMMN- `hMMMMMMMMMMMMMMMm+` :hMMM
MMMMMd:` ``-:+shmMMMMMMMMMMMMMMMMMMN. hMMMMMMMMMMMMMMMMMMMmhs+/-..``````./dMMMMM
MMMMMMMMMNNNNNNMMMMMMMMMMMMMMMMMMMMMMMMMMMo .MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMy .MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMN. /MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMN+` `+NMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMNs. -hMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
MMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMdyymMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM
```
---
## Summary
base: `https://github.com/prompt-toolkit/pymux`
```
tailoring a multiplexor for personal usage. using pymux for the base code, it's a tmux python port.
```
---
## Changelog - v0.1
- added own chat window, and wrote in an an arbitrary data poplulator until fd socks are networked.
---
## Prerequisites
###### [ substitute apt for the package manager of your choice ]
- `apt install python3` - `*note: python3.9 is ideal`
- `apt install python3-pip`
- `python3 -m pip install virtualenv`
---
## Instructions
###### - [ virtualenv could be substituted for pyenv/pipenv or no virtualenv et al ]
- `git clone --recursive https://git.tcp.direct/decoded/maple_netspan.git`
- `cd dr1p_pymux`
- `virtualenv -p python3.9 env`
- `source env/bin/activate`
- `pip install -r requirements.txt`
---
# Usage
## 1 - `cd dr1p_pymux`
## 2 - `source env/bin/activate`
## 3 - `python3.9 dr1p_pymux.py`
---

18
dr1p_pymux.py Normal file
View File

@ -0,0 +1,18 @@
###################################################################################### SOF
class DR1PP1NG():
color_depth="DEPTH_24_BIT"
def __init__(self,mode):
self.__mode__(mode)
def __mode__(self,mode):
if mode==1:
from pymux.main import Pymux
self.dr1pp1ng=Pymux()
self.dr1pp1ng.run_standalone(self.color_depth)
elif mode==2:
from pymux.entry_points.run_pymux import run
run()
##########################################################################################
if __name__=="__main__":
dr1pp1ng=DR1PP1NG(2)
#dr1pp1ng=DR1PP1NG(1)
###################################################################################### EOF

1
pymux/__init__.py Normal file
View File

@ -0,0 +1 @@
from __future__ import unicode_literals

8
pymux/__main__.py Normal file
View File

@ -0,0 +1,8 @@
"""
Make sure `python -m pymux` works.
"""
from __future__ import unicode_literals
from .entry_points.run_pymux import run
if __name__ == '__main__':
run()

737
pymux/arrangement.py Normal file
View File

@ -0,0 +1,737 @@
"""
Arrangement of panes.
Don't confuse with the prompt_toolkit VSplit/HSplit classes. This is a higher
level abstraction of the Pymux window layout.
An arrangement consists of a list of windows. And a window has a list of panes,
arranged by ordering them in HSplit/VSplit instances.
"""
from __future__ import unicode_literals
from ptterm import Terminal
from prompt_toolkit.application.current import get_app, set_app
from prompt_toolkit.buffer import Buffer
import math
import os
import weakref
import six
__all__ = (
'LayoutTypes',
'Pane',
'HSplit',
'VSplit',
'Window',
'Arrangement',
)
class LayoutTypes:
# The values are in lowercase with dashes, because that is what users can
# use at the command line.
EVEN_HORIZONTAL = 'even-horizontal'
EVEN_VERTICAL = 'even-vertical'
MAIN_HORIZONTAL = 'main-horizontal'
MAIN_VERTICAL = 'main-vertical'
TILED = 'tiled'
_ALL = [EVEN_HORIZONTAL, EVEN_VERTICAL, MAIN_HORIZONTAL, MAIN_VERTICAL, TILED]
class Pane(object):
"""
One pane, containing one process and a search buffer for going into copy
mode or displaying the help.
"""
_pane_counter = 1000 # Start at 1000, to be sure to not confuse this with pane indexes.
def __init__(self, terminal=None):
assert isinstance(terminal, Terminal)
self.terminal = terminal
self.chosen_name = None
# Displayed the clock instead of this pane content.
self.clock_mode = False
# Give unique ID.
Pane._pane_counter += 1
self.pane_id = Pane._pane_counter
# Prompt_toolkit buffer, for displaying scrollable text.
# (In copy mode, or help mode.)
# Note: Because the scroll_buffer can only contain text, we also use the
# get_tokens_for_line, that returns the token list with color
# information for each line.
self.scroll_buffer = Buffer(read_only=True)
self.copy_get_tokens_for_line = lambda lineno: []
self.display_scroll_buffer = False
self.scroll_buffer_title = ''
@property
def process(self):
return self.terminal.process
@property
def name(self):
"""
The name for the window as displayed in the title bar and status bar.
"""
# Name, explicitely set for the pane.
if self.chosen_name:
return self.chosen_name
else:
# Name from the process running inside the pane.
name = self.process.get_name()
if name:
return os.path.basename(name)
return ''
def enter_copy_mode(self):
"""
Suspend the process, and copy the screen content to the `scroll_buffer`.
That way the user can search through the history and copy/paste.
"""
self.terminal.enter_copy_mode()
def focus(self):
"""
Focus this pane.
"""
get_app().layout.focus(self.terminal)
class _WeightsDictionary(weakref.WeakKeyDictionary):
"""
Dictionary for the weights: weak keys, but defaults to 1.
(Weights are used to represent the proportion of pane sizes in
HSplit/VSplit lists.)
This dictionary maps the child (another HSplit/VSplit or Pane), to the
size. (Integer.)
"""
def __getitem__(self, key):
try:
# (Don't use 'super' here. This is a classobj in Python2.)
return weakref.WeakKeyDictionary.__getitem__(self, key)
except KeyError:
return 1
class _Split(list):
"""
Base class for horizontal and vertical splits. (This is a higher level
split than prompt_toolkit.layout.HSplit.)
"""
def __init__(self, *a, **kw):
list.__init__(self, *a, **kw)
# Mapping children to its weight.
self.weights = _WeightsDictionary()
def __hash__(self):
# Required in order to add HSplit/VSplit to the weights dict. "
return id(self)
def __repr__(self):
return '%s(%s)' % (self.__class__.__name__, list.__repr__(self))
class HSplit(_Split):
""" Horizontal split. """
class VSplit(_Split):
""" Horizontal split. """
class Window(object):
"""
Pymux window.
"""
_window_counter = 1000 # Start here, to avoid confusion with window index.
def __init__(self, index=0):
self.index = index
self.root = HSplit()
self._active_pane = None
self._prev_active_pane = None
self.chosen_name = None
self.previous_selected_layout = None
#: When true, the current pane is zoomed in.
self.zoom = False
#: When True, send input to all panes simultaniously.
self.synchronize_panes = False
# Give unique ID.
Window._window_counter += 1
self.window_id = Window._window_counter
def invalidation_hash(self):
"""
Return a hash (string) that can be used to determine when the layout
has to be rebuild.
"""
# if not self.root:
# return '<empty-window>'
def _hash_for_split(split):
result = []
for item in split:
if isinstance(item, (VSplit, HSplit)):
result.append(_hash_for_split(item))
elif isinstance(item, Pane):
result.append('p%s' % item.pane_id)
if isinstance(split, HSplit):
return 'HSplit(%s)' % (','.join(result))
else:
return 'VSplit(%s)' % (','.join(result))
return '<window_id=%s,zoom=%s,children=%s>' % (
self.window_id, self.zoom, _hash_for_split(self.root))
@property
def active_pane(self):
"""
The current active :class:`.Pane`.
"""
return self._active_pane
@active_pane.setter
def active_pane(self, value):
assert isinstance(value, Pane)
# Remember previous active pane.
if self._active_pane:
self._prev_active_pane = weakref.ref(self._active_pane)
self.zoom = False
self._active_pane = value
@property
def previous_active_pane(self):
"""
The previous active :class:`.Pane` or `None` if unknown.
"""
p = self._prev_active_pane and self._prev_active_pane()
# Only return when this pane actually still exists in the current
# window.
if p and p in self.panes:
return p
@property
def name(self):
"""
The name for this window as it should be displayed in the status bar.
"""
# Name, explicitely set for the window.
if self.chosen_name:
return self.chosen_name
else:
pane = self.active_pane
if pane:
return pane.name
return ''
def add_pane(self, pane, vsplit=False):
"""
Add another pane to this Window.
"""
assert isinstance(pane, Pane)
assert isinstance(vsplit, bool)
split_cls = VSplit if vsplit else HSplit
if self.active_pane is None:
self.root.append(pane)
else:
parent = self._get_parent(self.active_pane)
same_direction = isinstance(parent, split_cls)
index = parent.index(self.active_pane)
if same_direction:
parent.insert(index + 1, pane)
else:
new_split = split_cls([self.active_pane, pane])
parent[index] = new_split
# Give the newly created split the same weight as the original
# pane that was at this position.
parent.weights[new_split] = parent.weights[self.active_pane]
self.active_pane = pane
self.zoom = False
def remove_pane(self, pane):
"""
Remove pane from this Window.
"""
assert isinstance(pane, Pane)
if pane in self.panes:
# When this pane was focused, switch to previous active or next in order.
if pane == self.active_pane:
if self.previous_active_pane:
self.active_pane = self.previous_active_pane
else:
self.focus_next()
# Remove from the parent. When the parent becomes empty, remove the
# parent itself recursively.
p = self._get_parent(pane)
p.remove(pane)
while len(p) == 0 and p != self.root:
p2 = self._get_parent(p)
p2.remove(p)
p = p2
# When the parent has only one item left, collapse into its parent.
while len(p) == 1 and p != self.root:
p2 = self._get_parent(p)
p2.weights[p[0]] = p2.weights[p] # Keep dimensions.
i = p2.index(p)
p2[i] = p[0]
p = p2
@property
def panes(self):
" List with all panes from this Window. "
result = []
for s in self.splits:
for item in s:
if isinstance(item, Pane):
result.append(item)
return result
@property
def splits(self):
" Return a list with all HSplit/VSplit instances. "
result = []
def collect(split):
result.append(split)
for item in split:
if isinstance(item, (HSplit, VSplit)):
collect(item)
collect(self.root)
return result
def _get_parent(self, item):
" The HSplit/VSplit that contains the active pane. "
for s in self.splits:
if item in s:
return s
@property
def has_panes(self):
" True when this window contains at least one pane. "
return len(self.panes) > 0
@property
def active_process(self):
" Return `Process` that should receive user input. "
p = self.active_pane
if p is not None:
return p.process
def focus_next(self, count=1):
" Focus the next pane. "
panes = self.panes
if panes:
self.active_pane = panes[(panes.index(self.active_pane) + count) % len(panes)]
else:
self.active_pane = None # No panes left.
def focus_previous(self):
" Focus the previous pane. "
self.focus_next(count=-1)
def rotate(self, count=1, with_pane_before_only=False, with_pane_after_only=False):
"""
Rotate panes.
When `with_pane_before_only` or `with_pane_after_only` is True, only rotate
with the pane before/after the active pane.
"""
# Create (split, index, pane, weight) tuples.
items = []
current_pane_index = None
for s in self.splits:
for index, item in enumerate(s):
if isinstance(item, Pane):
items.append((s, index, item, s.weights[item]))
if item == self.active_pane:
current_pane_index = len(items) - 1
# Only before after? Reduce list of panes.
if with_pane_before_only:
items = items[current_pane_index - 1:current_pane_index + 1]
elif with_pane_after_only:
items = items[current_pane_index:current_pane_index + 2]
# Rotate positions.
for i, triple in enumerate(items):
split, index, pane, weight = triple
new_item = items[(i + count) % len(items)][2]
split[index] = new_item
split.weights[new_item] = weight
def select_layout(self, layout_type):
"""
Select one of the predefined layouts.
"""
assert layout_type in LayoutTypes._ALL
# When there is only one pane, always choose EVEN_HORIZONTAL,
# Otherwise, we create VSplit/HSplit instances with an empty list of
# children.
if len(self.panes) == 1:
layout_type = LayoutTypes.EVEN_HORIZONTAL
# even-horizontal.
if layout_type == LayoutTypes.EVEN_HORIZONTAL:
self.root = HSplit(self.panes)
# even-vertical.
elif layout_type == LayoutTypes.EVEN_VERTICAL:
self.root = VSplit(self.panes)
# main-horizontal.
elif layout_type == LayoutTypes.MAIN_HORIZONTAL:
self.root = HSplit([
self.active_pane,
VSplit([p for p in self.panes if p != self.active_pane])
])
# main-vertical.
elif layout_type == LayoutTypes.MAIN_VERTICAL:
self.root = VSplit([
self.active_pane,
HSplit([p for p in self.panes if p != self.active_pane])
])
# tiled.
elif layout_type == LayoutTypes.TILED:
panes = self.panes
column_count = math.ceil(len(panes) ** .5)
rows = HSplit()
current_row = VSplit()
for p in panes:
current_row.append(p)
if len(current_row) >= column_count:
rows.append(current_row)
current_row = VSplit()
if current_row:
rows.append(current_row)
self.root = rows
self.previous_selected_layout = layout_type
def select_next_layout(self, count=1):
"""
Select next layout. (Cycle through predefined layouts.)
"""
# List of all layouts. (When we have just two panes, only toggle
# between horizontal/vertical.)
if len(self.panes) == 2:
all_layouts = [LayoutTypes.EVEN_HORIZONTAL, LayoutTypes.EVEN_VERTICAL]
else:
all_layouts = LayoutTypes._ALL
# Get index of current layout.
layout = self.previous_selected_layout or LayoutTypes._ALL[-1]
try:
index = all_layouts.index(layout)
except ValueError:
index = 0
# Switch to new layout.
new_layout = all_layouts[(index + count) % len(all_layouts)]
self.select_layout(new_layout)
def select_previous_layout(self):
self.select_next_layout(count=-1)
def change_size_for_active_pane(self, up=0, right=0, down=0, left=0):
"""
Increase the size of the current pane in any of the four directions.
"""
child = self.active_pane
self.change_size_for_pane(child, up=up, right=right, down=down, left=left)
def change_size_for_pane(self, pane, up=0, right=0, down=0, left=0):
"""
Increase the size of the current pane in any of the four directions.
Positive values indicate an increase, negative values a decrease.
"""
assert isinstance(pane, Pane)
def find_split_and_child(split_cls, is_before):
" Find the split for which we will have to update the weights. "
child = pane
split = self._get_parent(child)
def found():
return isinstance(split, split_cls) and (
not is_before or split.index(child) > 0) and (
is_before or split.index(child) < len(split) - 1)
while split and not found():
child = split
split = self._get_parent(child)
return split, child # split can be None!
def handle_side(split_cls, is_before, amount, trying_other_side=False):
" Increase weights on one side. (top/left/right/bottom). "
if amount:
split, child = find_split_and_child(split_cls, is_before)
if split:
# Find neighbour.
neighbour_index = split.index(child) + (-1 if is_before else 1)
neighbour_child = split[neighbour_index]
# Increase/decrease weights.
split.weights[child] += amount
split.weights[neighbour_child] -= amount
# Ensure that all weights are at least one.
for k, value in split.weights.items():
if value < 1:
split.weights[k] = 1
else:
# When no split has been found where we can move in this
# direction, try to move the other side instead using a
# negative amount. This happens when we run "resize-pane -R 4"
# inside the pane that is completely on the right. In that
# case it's logical to move the left border to the right
# instead.
if not trying_other_side:
handle_side(split_cls, not is_before, -amount,
trying_other_side=True)
handle_side(VSplit, True, left)
handle_side(VSplit, False, right)
handle_side(HSplit, True, up)
handle_side(HSplit, False, down)
def get_pane_index(self, pane):
" Return the index of the given pane. ValueError if not found. "
assert isinstance(pane, Pane)
return self.panes.index(pane)
class Arrangement(object):
"""
Arrangement class for one Pymux session.
This contains the list of windows and the layout of the panes for each
window. All the clients share the same Arrangement instance, but they can
have different windows active.
"""
def __init__(self):
self.windows = []
self.base_index = 0
self._active_window_for_cli = weakref.WeakKeyDictionary()
self._prev_active_window_for_cli = weakref.WeakKeyDictionary()
# The active window of the last CLI. Used as default when a new session
# is attached.
self._last_active_window = None
def invalidation_hash(self):
"""
When this changes, the layout needs to be rebuild.
"""
if not self.windows:
return '<no-windows>'
w = self.get_active_window()
return w.invalidation_hash()
def get_active_window(self):
"""
The current active :class:`.Window`.
"""
app = get_app()
try:
return self._active_window_for_cli[app]
except KeyError:
self._active_window_for_cli[app] = self._last_active_window or self.windows[0]
return self.windows[0]
def set_active_window(self, window):
assert isinstance(window, Window)
app = get_app()
previous = self.get_active_window()
self._prev_active_window_for_cli[app] = previous
self._active_window_for_cli[app] = window
self._last_active_window = window
def set_active_window_from_pane_id(self, pane_id):
"""
Make the window with this pane ID the active Window.
"""
assert isinstance(pane_id, int)
for w in self.windows:
for p in w.panes:
if p.pane_id == pane_id:
self.set_active_window(w)
def get_previous_active_window(self):
" The previous active Window or None if unknown. "
app = get_app()
try:
return self._prev_active_window_for_cli[app]
except KeyError:
return None
def get_window_by_index(self, index):
" Return the Window with this index or None if not found. "
for w in self.windows:
if w.index == index:
return w
def create_window(self, pane, name=None, set_active=True):
"""
Create a new window that contains just this pane.
:param pane: The :class:`.Pane` instance to put in the new window.
:param name: If given, name for the new window.
:param set_active: When True, focus the new window.
"""
assert isinstance(pane, Pane)
assert name is None or isinstance(name, six.text_type)
# Take the first available index.
taken_indexes = [w.index for w in self.windows]
index = self.base_index
while index in taken_indexes:
index += 1
# Create new window and add it.
w = Window(index)
w.add_pane(pane)
self.windows.append(w)
# Sort windows by index.
self.windows = sorted(self.windows, key=lambda w: w.index)
app = get_app(return_none=True)
if app is not None and set_active:
self.set_active_window(w)
if name is not None:
w.chosen_name = name
assert w.active_pane == pane
assert w._get_parent(pane)
def move_window(self, window, new_index):
"""
Move window to a new index.
"""
assert isinstance(window, Window)
assert isinstance(new_index, int)
window.index = new_index
# Sort windows by index.
self.windows = sorted(self.windows, key=lambda w: w.index)
def get_active_pane(self):
"""
The current :class:`.Pane` from the current window.
"""
w = self.get_active_window()
if w is not None:
return w.active_pane
def remove_pane(self, pane):
"""
Remove a :class:`.Pane`. (Look in all windows.)
"""
assert isinstance(pane, Pane)
for w in self.windows:
w.remove_pane(pane)
# No panes left in this window?
if not w.has_panes:
# Focus next.
for app, active_w in self._active_window_for_cli.items():
if w == active_w:
with set_app(app):
self.focus_next_window()
self.windows.remove(w)
def focus_previous_window(self):
w = self.get_active_window()
self.set_active_window(self.windows[
(self.windows.index(w) - 1) % len(self.windows)])
def focus_next_window(self):
w = self.get_active_window()
self.set_active_window(self.windows[
(self.windows.index(w) + 1) % len(self.windows)])
def break_pane(self, set_active=True):
"""
When the current window has multiple panes, remove the pane from this
window and put it in a new window.
:param set_active: When True, focus the new window.
"""
w = self.get_active_window()
if len(w.panes) > 1:
pane = w.active_pane
self.get_active_window().remove_pane(pane)
self.create_window(pane, set_active=set_active)
def rotate_window(self, count=1):
" Rotate the panes in the active window. "
w = self.get_active_window()
w.rotate(count=count)
@property
def has_panes(self):
" True when any of the windows has a :class:`.Pane`. "
for w in self.windows:
if w.has_panes:
return True
return False

3
pymux/client/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from __future__ import unicode_literals
from .base import Client
from .defaults import create_client, list_clients

22
pymux/client/base.py Normal file
View File

@ -0,0 +1,22 @@
from __future__ import unicode_literals
from prompt_toolkit.output import ColorDepth
from abc import ABCMeta
from six import with_metaclass
__all__ = [
'Client',
]
class Client(with_metaclass(ABCMeta, object)):
def run_command(self, command, pane_id=None):
"""
Ask the server to run this command.
"""
def attach(self, detach_other_clients=False, color_depth=ColorDepth.DEPTH_8_BIT):
"""
Attach client user interface.
"""

24
pymux/client/defaults.py Normal file
View File

@ -0,0 +1,24 @@
from __future__ import unicode_literals
from prompt_toolkit.utils import is_windows
__all__ = [
'create_client',
'list_clients',
]
def create_client(socket_name):
if is_windows():
from .windows import WindowsClient
return WindowsClient(socket_name)
else:
from .posix import PosixClient
return PosixClient(socket_name)
def list_clients():
if is_windows():
from .windows import list_clients
return list_clients()
else:
from .posix import list_clients
return list_clients()

207
pymux/client/posix.py Normal file
View File

@ -0,0 +1,207 @@
from __future__ import unicode_literals
from prompt_toolkit.eventloop.select import select_fds
from prompt_toolkit.input.posix_utils import PosixStdinReader
from prompt_toolkit.input.vt100 import raw_mode, cooked_mode
from prompt_toolkit.output.vt100 import _get_size, Vt100_Output
from prompt_toolkit.output import ColorDepth
from pymux.utils import nonblocking
import getpass
import glob
import json
import os
import signal
import socket
import sys
import tempfile
from .base import Client
INPUT_TIMEOUT = .5
__all__ = (
'PosixClient',
'list_clients',
)
class PosixClient(Client):
def __init__(self, socket_name):
self.socket_name = socket_name
self._mode_context_managers = []
# Connect to socket.
self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.socket.connect(socket_name)
self.socket.setblocking(1)
# Input reader.
# Some terminals, like lxterminal send non UTF-8 input sequences,
# even when the input encoding is supposed to be UTF-8. This
# happens in the case of mouse clicks in the right area of a wide
# terminal. Apparently, these are some binary blobs in between the
# UTF-8 input.)
# We should not replace these, because this would break the
# decoding otherwise. (Also don't pass errors='ignore', because
# that doesn't work for parsing mouse input escape sequences, which
# consist of a fixed number of bytes.)
self._stdin_reader = PosixStdinReader(sys.stdin.fileno(), errors='replace')
def run_command(self, command, pane_id=None):
"""
Ask the server to run this command.
:param pane_id: Optional identifier of the current pane.
"""
self._send_packet({
'cmd': 'run-command',
'data': command,
'pane_id': pane_id
})
def attach(self, detach_other_clients=False, color_depth=ColorDepth.DEPTH_8_BIT):
"""
Attach client user interface.
"""
assert isinstance(detach_other_clients, bool)
self._send_size()
self._send_packet({
'cmd': 'start-gui',
'detach-others': detach_other_clients,
'color-depth': color_depth,
'term': os.environ.get('TERM', ''),
'data': ''
})
with raw_mode(sys.stdin.fileno()):
data_buffer = b''
stdin_fd = sys.stdin.fileno()
socket_fd = self.socket.fileno()
current_timeout = INPUT_TIMEOUT # Timeout, used to flush escape sequences.
try:
def winch_handler(signum, frame):
self._send_size()
signal.signal(signal.SIGWINCH, winch_handler)
while True:
r = select_fds([stdin_fd, socket_fd], current_timeout)
if socket_fd in r:
# Received packet from server.
#################################################################################### DECODED
try:
data = self.socket.recv(1024)
except ConnectionResetError:
pass
#################################################################################### DECODED
if data == b'':
# End of file. Connection closed.
# Reset terminal
o = Vt100_Output.from_pty(sys.stdout)
o.quit_alternate_screen()
o.disable_mouse_support()
o.disable_bracketed_paste()
o.reset_attributes()
o.flush()
return
else:
data_buffer += data
while b'\0' in data_buffer:
pos = data_buffer.index(b'\0')
self._process(data_buffer[:pos])
data_buffer = data_buffer[pos + 1:]
elif stdin_fd in r:
# Got user input.
self._process_stdin()
current_timeout = INPUT_TIMEOUT
else:
# Timeout. (Tell the server to flush the vt100 Escape.)
self._send_packet({'cmd': 'flush-input'})
current_timeout = None
finally:
signal.signal(signal.SIGWINCH, signal.SIG_IGN)
def _process(self, data_buffer):
"""
Handle incoming packet from server.
"""
packet = json.loads(data_buffer.decode('utf-8'))
if packet['cmd'] == 'out':
# Call os.write manually. In Python2.6, sys.stdout.write doesn't use UTF-8.
os.write(sys.stdout.fileno(), packet['data'].encode('utf-8'))
elif packet['cmd'] == 'suspend':
# Suspend client process to background.
if hasattr(signal, 'SIGTSTP'):
os.kill(os.getpid(), signal.SIGTSTP)
elif packet['cmd'] == 'mode':
# Set terminal to raw/cooked.
action = packet['data']
if action == 'raw':
cm = raw_mode(sys.stdin.fileno())
cm.__enter__()
self._mode_context_managers.append(cm)
elif action == 'cooked':
cm = cooked_mode(sys.stdin.fileno())
cm.__enter__()
self._mode_context_managers.append(cm)
elif action == 'restore' and self._mode_context_managers:
cm = self._mode_context_managers.pop()
cm.__exit__()
def _process_stdin(self):
"""
Received data on stdin. Read and send to server.
"""
with nonblocking(sys.stdin.fileno()):
data = self._stdin_reader.read()
# Send input in chunks of 4k.
step = 4056
for i in range(0, len(data), step):
self._send_packet({
'cmd': 'in',
'data': data[i:i + step],
})
def _send_packet(self, data):
" Send to server. "
data = json.dumps(data).encode('utf-8')
# Be sure that our socket is blocking, otherwise, the send() call could
# raise `BlockingIOError` if the buffer is full.
self.socket.setblocking(1)
self.socket.send(data + b'\0')
def _send_size(self):
" Report terminal size to server. "
rows, cols = _get_size(sys.stdout.fileno())
self._send_packet({
'cmd': 'size',
'data': [rows, cols]
})
def list_clients():
"""
List all the servers that are running.
"""
p = '%s/pymux.sock.%s.*' % (tempfile.gettempdir(), getpass.getuser())
for path in glob.glob(p):
try:
yield PosixClient(path)
except socket.error:
pass

128
pymux/client/windows.py Normal file
View File

@ -0,0 +1,128 @@
from __future__ import unicode_literals
from ctypes import byref, windll
from ctypes.wintypes import DWORD
from prompt_toolkit.eventloop import ensure_future, From
from prompt_toolkit.eventloop import get_event_loop
from prompt_toolkit.input.win32 import Win32Input
from prompt_toolkit.output import ColorDepth
from prompt_toolkit.output.win32 import Win32Output
from prompt_toolkit.win32_types import STD_OUTPUT_HANDLE
import json
import os
import sys
from ..pipes.win32_client import PipeClient
from .base import Client
__all__ = [
'WindowsClient',
'list_clients',
]
# See: https://msdn.microsoft.com/pl-pl/library/windows/desktop/ms686033(v=vs.85).aspx
ENABLE_PROCESSED_INPUT = 0x0001
ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004
class WindowsClient(Client):
def __init__(self, pipe_name):
self._input = Win32Input()
self._hconsole = windll.kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
self._data_buffer = b''
self.pipe = PipeClient(pipe_name)
def attach(self, detach_other_clients=False, color_depth=ColorDepth.DEPTH_8_BIT):
assert isinstance(detach_other_clients, bool)
self._send_size()
self._send_packet({
'cmd': 'start-gui',
'detach-others': detach_other_clients,
'color-depth': color_depth,
'term': os.environ.get('TERM', ''),
'data': ''
})
f = ensure_future(self._start_reader())
with self._input.attach(self._input_ready):
# Run as long as we have a connection with the server.
get_event_loop().run_until_complete(f) # Run forever.
def _start_reader(self):
"""
Read messages from the Win32 pipe server and handle them.
"""
while True:
message = yield From(self.pipe.read_message())
self._process(message)
def _process(self, data_buffer):
"""
Handle incoming packet from server.
"""
packet = json.loads(data_buffer)
if packet['cmd'] == 'out':
# Call os.write manually. In Python2.6, sys.stdout.write doesn't use UTF-8.
original_mode = DWORD(0)
windll.kernel32.GetConsoleMode(self._hconsole, byref(original_mode))
windll.kernel32.SetConsoleMode(self._hconsole, DWORD(
ENABLE_PROCESSED_INPUT | ENABLE_VIRTUAL_TERMINAL_PROCESSING))
try:
os.write(sys.stdout.fileno(), packet['data'].encode('utf-8'))
finally:
windll.kernel32.SetConsoleMode(self._hconsole, original_mode)
elif packet['cmd'] == 'suspend':
# Suspend client process to background.
pass
elif packet['cmd'] == 'mode':
pass
# # Set terminal to raw/cooked.
# action = packet['data']
# if action == 'raw':
# cm = raw_mode(sys.stdin.fileno())
# cm.__enter__()
# self._mode_context_managers.append(cm)
# elif action == 'cooked':
# cm = cooked_mode(sys.stdin.fileno())
# cm.__enter__()
# self._mode_context_managers.append(cm)
# elif action == 'restore' and self._mode_context_managers:
# cm = self._mode_context_managers.pop()
# cm.__exit__()
def _input_ready(self):
keys = self._input.read_keys()
if keys:
self._send_packet({
'cmd': 'in',
'data': ''.join(key_press.data for key_press in keys),
})
def _send_packet(self, data):
" Send to server. "
data = json.dumps(data)
ensure_future(self.pipe.write_message(data))
def _send_size(self):
" Report terminal size to server. "
output = Win32Output(sys.stdout)
rows, cols = output.get_size()
self._send_packet({
'cmd': 'size',
'data': [rows, cols]
})
def list_clients():
return []

0
pymux/commands/__init__.py Normal file
View File

51
pymux/commands/aliases.py Normal file
View File

@ -0,0 +1,51 @@
"""
Aliases for all commands.
(On purpose kept compatible with tmux.)
"""
from __future__ import unicode_literals
__all__ = (
'ALIASES',
)
ALIASES = {
'bind': 'bind-key',
'breakp': 'break-pane',
'clearhist': 'clear-history',
'confirm': 'confirm-before',
'detach': 'detach-client',
'display': 'display-message',
######################################################################################################## DECODED
'decoded': 'display-decoded',
######################################################################################################## DECODED
'displayp': 'display-panes',
'killp': 'kill-pane',
'killw': 'kill-window',
'last': 'last-window',
'lastp': 'last-pane',
'lextl': 'next-layout',
'lsk': 'list-keys',
'lsp': 'list-panes',
'movew': 'move-window',
'neww': 'new-window',
'next': 'next-window',
'pasteb': 'paste-buffer',
'prev': 'previous-window',
'prevl': 'previous-layout',
'rename': 'rename-session',
'renamew': 'rename-window',
'resizep': 'resize-pane',
'rotatew': 'rotate-window',
'selectl': 'select-layout',
'selectp': 'select-pane',
'selectw': 'select-window',
'send': 'send-keys',
'set': 'set-option',
'setw': 'set-window-option',
'source': 'source-file',
'splitw': 'split-window',
'suspendc': 'suspend-client',
'swapp': 'swap-pane',
'unbind': 'unbind-key',
}

676
pymux/commands/commands.py Normal file
View File

@ -0,0 +1,676 @@
from __future__ import unicode_literals
import docopt
import os
import re
import shlex
import six
from prompt_toolkit.application.current import get_app
from prompt_toolkit.document import Document
from prompt_toolkit.key_binding.vi_state import InputMode
from pymux.arrangement import LayoutTypes
from pymux.commands.aliases import ALIASES
from pymux.commands.utils import wrap_argument
from pymux.format import format_pymux_string
from pymux.key_mappings import pymux_key_to_prompt_toolkit_key_sequence, prompt_toolkit_key_to_vt100_key
from pymux.layout import focus_right, focus_left, focus_up, focus_down
from pymux.log import logger
from pymux.options import SetOptionError
__all__ = (
'call_command_handler',
'get_documentation_for_command',
'get_option_flags_for_command',
'handle_command',
'has_command_handler',
)
COMMANDS_TO_HANDLERS = {} # Global mapping of pymux commands to their handlers.
COMMANDS_TO_HELP = {}
COMMANDS_TO_OPTION_FLAGS = {}
def has_command_handler(command):
return command in COMMANDS_TO_HANDLERS
def get_documentation_for_command(command):
""" Return the help text for this command, or None if the command is not
known. """
if command in COMMANDS_TO_HELP:
return 'Usage: %s %s' % (command, COMMANDS_TO_HELP.get(command, ''))
def get_option_flags_for_command(command):
" Return a list of options (-x flags) for this command. "
return COMMANDS_TO_OPTION_FLAGS.get(command, [])
def handle_command(pymux, input_string):
"""
Handle command.
"""
assert isinstance(input_string, six.text_type)
input_string = input_string.strip()
logger.info('handle command: %s %s.', input_string, type(input_string))
if input_string and not input_string.startswith('#'): # Ignore comments.
try:
if six.PY2:
# In Python2.6, shlex doesn't work with unicode input at all.
# In Python2.7, shlex tries to encode using ASCII.
parts = shlex.split(input_string.encode('utf-8'))
parts = [p.decode('utf-8') for p in parts]
else:
parts = shlex.split(input_string)
except ValueError as e:
# E.g. missing closing quote.
pymux.show_message('Invalid command %s: %s' % (input_string, e))
else:
call_command_handler(parts[0], pymux, parts[1:])
def call_command_handler(command, pymux, arguments):
"""
Execute command.
:param arguments: List of options.
"""
assert isinstance(arguments, list)
# Resolve aliases.
command = ALIASES.get(command, command)
try:
handler = COMMANDS_TO_HANDLERS[command]
except KeyError:
pymux.show_message('Invalid command: %s' % (command,))
else:
try:
handler(pymux, arguments)
except CommandException as e:
pymux.show_message(e.message)
def cmd(name, options=''):
"""
Decorator for all commands.
Commands will receive (pymux, variables) as input.
Commands can raise CommandException.
"""
# Validate options.
if options:
try:
docopt.docopt('Usage:\n %s %s' % (name, options, ), [])
except SystemExit:
pass
def decorator(func):
def command_wrapper(pymux, arguments):
# Hack to make the 'bind-key' option work.
# (bind-key expects a variable number of arguments.)
if name == 'bind-key' and '--' not in arguments:
# Insert a double dash after the first non-option.
for i, p in enumerate(arguments):
if not p.startswith('-'):
arguments.insert(i + 1, '--')
break
# Parse options.
try:
# Python 2 workaround: pass bytes to docopt.
# From the following, only the bytes version returns the right
# output in Python 2:
# docopt.docopt('Usage:\n app <params>...', [b'a', b'b'])
# docopt.docopt('Usage:\n app <params>...', [u'a', u'b'])
# https://github.com/docopt/docopt/issues/30
# (Not sure how reliable this is...)
if six.PY2:
arguments = [a.encode('utf-8') for a in arguments]
received_options = docopt.docopt(
'Usage:\n %s %s' % (name, options),
arguments,
help=False) # Don't interpret the '-h' option as help.
# Make sure that all the received options from docopt are
# unicode objects. (Docopt returns 'str' for Python2.)
for k, v in received_options.items():
if isinstance(v, six.binary_type):
received_options[k] = v.decode('utf-8')
except SystemExit:
raise CommandException('Usage: %s %s' % (name, options))
# Call handler.
func(pymux, received_options)
# Invalidate all clients, not just the current CLI.
pymux.invalidate()
COMMANDS_TO_HANDLERS[name] = command_wrapper
COMMANDS_TO_HELP[name] = options
# Get list of option flags.
flags = re.findall(r'-[a-zA-Z0-9]\b', options)
COMMANDS_TO_OPTION_FLAGS[name] = flags
return func
return decorator
class CommandException(Exception):
" When raised from a command handler, this message will be shown. "
def __init__(self, message):
self.message = message
#
# The actual commands.
#
@cmd('break-pane', options='[-d]')
def break_pane(pymux, variables):
dont_focus_window = variables['-d']
pymux.arrangement.break_pane(set_active=not dont_focus_window)
pymux.invalidate()
@cmd('select-pane', options='(-L|-R|-U|-D|-t <pane-id>)')
def select_pane(pymux, variables):
if variables['-t']:
pane_id = variables['<pane-id>']
w = pymux.arrangement.get_active_window()
if pane_id == ':.+':
w.focus_next()
elif pane_id == ':.-':
w.focus_previous()
else:
# Select pane by index.
try:
pane_id = int(pane_id[1:])
w.active_pane = w.panes[pane_id]
except (IndexError, ValueError):
raise CommandException('Invalid pane.')
else:
if variables['-L']: h = focus_left
if variables['-U']: h = focus_up
if variables['-D']: h = focus_down
if variables['-R']: h = focus_right
h(pymux)
@cmd('select-window', options='(-t <target-window>)')
def select_window(pymux, variables):
"""
Select a window. E.g: select-window -t :3
"""
window_id = variables['<target-window>']
def invalid_window():
raise CommandException('Invalid window: %s' % window_id)
if window_id.startswith(':'):
try:
number = int(window_id[1:])
except ValueError:
invalid_window()
else:
w = pymux.arrangement.get_window_by_index(number)
if w:
pymux.arrangement.set_active_window(w)
else:
invalid_window()