cengal.parallel_execution.coroutines.integrations.pytermgui.versions.v_0.pytermgui

View Source

  1#!/usr/bin/env python
  2# coding=utf-8
  3
  4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
  5# 
  6# Licensed under the Apache License, Version 2.0 (the "License");
  7# you may not use this file except in compliance with the License.
  8# You may obtain a copy of the License at
  9# 
 10#     http://www.apache.org/licenses/LICENSE-2.0
 11# 
 12# Unless required by applicable law or agreed to in writing, software
 13# distributed under the License is distributed on an "AS IS" BASIS,
 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15# See the License for the specific language governing permissions and
 16# limitations under the License.
 17
 18
 19__all__ = ['WindowManagerCS', 'CompositorCS', 'TerminalApplication', 'TermApp']
 20
 21
 22"""
 23Module Docstring
 24Docstrings: http://www.python.org/dev/peps/pep-0257/
 25"""
 26
 27__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 28__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 29__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 30__license__ = "Apache License, Version 2.0"
 31__version__ = "4.4.1"
 32__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 33__email__ = "gtalk@butenkoms.space"
 34# __status__ = "Prototype"
 35__status__ = "Development"
 36# __status__ = "Production"
 37
 38
 39from cengal.parallel_execution.coroutines.coro_scheduler import *
 40from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import put_coro_to
 41from cengal.parallel_execution.coroutines.coro_standard_services.shutdown_loop import ShutdownLoop
 42from cengal.parallel_execution.coroutines.coro_tools.prepare_loop import prepare_loop
 43from cengal.code_flow_control.smart_values import ValueExistence
 44from cengal.data_manipulation.conversion.reinterpret_cast import reinterpret_cast
 45
 46from pytermgui import WindowManager as PTGWindowManager
 47from pytermgui.window_manager import Compositor as PTGCompositor
 48
 49from typing import Any, Dict, Tuple, Optional, Callable
 50
 51
 52class CompositorCS(PTGCompositor):
 53    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
 54        self.setup_coro_worker: AnyWorker = setup_coro_worker
 55        self.setup_coro_worker_args: Tuple = args
 56        self.setup_coro_worker_kwargs: Dict = kwargs
 57        self.cs: CoroSchedulerType = None
 58        self.setup_coro_worker_value_holder: ValueExistence = None
 59
 60    def ensure_cs(self):
 61        if self.cs is None:
 62            self.cs, self.setup_coro_worker_value_holder = prepare_loop(self.setup_coro_worker, *self.setup_coro_worker_args, **self.setup_coro_worker_kwargs)
 63
 64    def iter_cs(self):
 65        in_work = self.cs.iteration()
 66
 67    def draw(self, force: bool = False) -> None:
 68        self.ensure_cs()
 69        self.iter_cs()
 70        super().draw(force)
 71    
 72    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
 73        put_coro_to((self.cs, None, False), coro_worker, *args, **kwargs)
 74
 75    def stop(self) -> None:
 76        async def coro(i: Interface, self):
 77            def shutdown_handler():
 78                self._is_running = False
 79            
 80            i._loop.on_destroyed_handlers.add(shutdown_handler)
 81            await i(ShutdownLoop)
 82        
 83        self.put_coro(coro, self)
 84
 85
 86class WindowManagerCS(PTGWindowManager):
 87    @staticmethod
 88    def replace(window_manager: PTGWindowManager, model, view_setup: Callable, controller_coro_setup_worker: Optional[AnyWorker] = None, *args, **kwargs) -> 'WindowManagerCS':
 89        reinterpret_cast(WindowManagerCS, window_manager)
 90        view_setup(window_manager, model)
 91        window_manager.init_cs(controller_coro_setup_worker, model, *args, **kwargs)
 92        return window_manager
 93
 94    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
 95        reinterpret_cast(CompositorCS, self.compositor)
 96        self.cs: CoroSchedulerType = None
 97        self.compositor.init_cs(setup_coro_worker, *args, **kwargs)
 98    
 99    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
100        self.compositor.put_coro(coro_worker, *args, **kwargs)
101
102
103class TerminalApplication:
104    def __init__(self):
105        self.model = self.model_setup()
106        self.manager: WindowManagerCS = None
107    
108    def run(self):
109        with PTGWindowManager() as manager:
110            self.manager = manager
111            WindowManagerCS.replace(manager, self.model, self.ui_setup, self.controller_loop_setup)
112    
113    def __call__(self):
114        self.run()
115    
116    def model_setup(self):
117        raise NotImplementedError()
118    
119    def ui_setup(self, manager: WindowManagerCS, model):
120        raise NotImplementedError()
121    
122    def controller_loop_setup(self, i: Interface, model):
123        raise NotImplementedError()
124
125
126TermApp = TerminalApplication

class WindowManagerCS(pytermgui.window_manager.manager.WindowManager): View Source

 87class WindowManagerCS(PTGWindowManager):
 88    @staticmethod
 89    def replace(window_manager: PTGWindowManager, model, view_setup: Callable, controller_coro_setup_worker: Optional[AnyWorker] = None, *args, **kwargs) -> 'WindowManagerCS':
 90        reinterpret_cast(WindowManagerCS, window_manager)
 91        view_setup(window_manager, model)
 92        window_manager.init_cs(controller_coro_setup_worker, model, *args, **kwargs)
 93        return window_manager
 94
 95    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
 96        reinterpret_cast(CompositorCS, self.compositor)
 97        self.cs: CoroSchedulerType = None
 98        self.compositor.init_cs(setup_coro_worker, *args, **kwargs)
 99    
100    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
101        self.compositor.put_coro(coro_worker, *args, **kwargs)

The manager of windows.

This class can be used, or even subclassed in order to create full-screen applications, using the pytermgui.window_manager.window.Window class and the general Widget API.

@staticmethod

def replace( window_manager: pytermgui.window_manager.manager.WindowManager, model, view_setup: typing.Callable, controller_coro_setup_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]], NoneType] = None, *args, **kwargs) -> WindowManagerCS: View Source

88    @staticmethod
89    def replace(window_manager: PTGWindowManager, model, view_setup: Callable, controller_coro_setup_worker: Optional[AnyWorker] = None, *args, **kwargs) -> 'WindowManagerCS':
90        reinterpret_cast(WindowManagerCS, window_manager)
91        view_setup(window_manager, model)
92        window_manager.init_cs(controller_coro_setup_worker, model, *args, **kwargs)
93        return window_manager

def init_cs( self, setup_coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs): View Source

95    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
96        reinterpret_cast(CompositorCS, self.compositor)
97        self.cs: CoroSchedulerType = None
98        self.compositor.init_cs(setup_coro_worker, *args, **kwargs)

def put_coro( self, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs): View Source

100    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
101        self.compositor.put_coro(coro_worker, *args, **kwargs)

Inherited Members

pytermgui.window_manager.manager.WindowManager: WindowManager; focusing_actions; autorun; focused; layout; compositor; mouse_translator; restrict_within_bounds; get_lines; clear_cache; on_resize; run; stop; add; remove; focus; focus_next; handle_key; process_mouse; screenshot; show_positions; alert; toast
pytermgui.widgets.base.Widget: set_style; set_char; styles; chars; keys; serialized; size_policy; parent_align; from_data; box; width; height; pos; depth; parent; selected_index; positioned_line_buffer; bindings; id; selectables_length; selectables; is_selectable; static_width; relative_width; terminal; get_change; contains; handle_mouse; serialize; copy; bind; unbind; execute_binding; select; print; debug

class CompositorCS(pytermgui.window_manager.compositor.Compositor): View Source

53class CompositorCS(PTGCompositor):
54    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
55        self.setup_coro_worker: AnyWorker = setup_coro_worker
56        self.setup_coro_worker_args: Tuple = args
57        self.setup_coro_worker_kwargs: Dict = kwargs
58        self.cs: CoroSchedulerType = None
59        self.setup_coro_worker_value_holder: ValueExistence = None
60
61    def ensure_cs(self):
62        if self.cs is None:
63            self.cs, self.setup_coro_worker_value_holder = prepare_loop(self.setup_coro_worker, *self.setup_coro_worker_args, **self.setup_coro_worker_kwargs)
64
65    def iter_cs(self):
66        in_work = self.cs.iteration()
67
68    def draw(self, force: bool = False) -> None:
69        self.ensure_cs()
70        self.iter_cs()
71        super().draw(force)
72    
73    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
74        put_coro_to((self.cs, None, False), coro_worker, *args, **kwargs)
75
76    def stop(self) -> None:
77        async def coro(i: Interface, self):
78            def shutdown_handler():
79                self._is_running = False
80            
81            i._loop.on_destroyed_handlers.add(shutdown_handler)
82            await i(ShutdownLoop)
83        
84        self.put_coro(coro, self)

The class used to draw pytermgui.window_managers.manager.WindowManager state.

This class handles turning a list of windows into a drawable buffer (composite), and then drawing it onto the screen.

Calling its run method will start the drawing thread, which will draw the current window states onto the screen. This routine targets framerate, though will likely not match it perfectly.

def init_cs( self, setup_coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs): View Source

54    def init_cs(self, setup_coro_worker: AnyWorker, *args, **kwargs):
55        self.setup_coro_worker: AnyWorker = setup_coro_worker
56        self.setup_coro_worker_args: Tuple = args
57        self.setup_coro_worker_kwargs: Dict = kwargs
58        self.cs: CoroSchedulerType = None
59        self.setup_coro_worker_value_holder: ValueExistence = None

def ensure_cs(self): View Source

61    def ensure_cs(self):
62        if self.cs is None:
63            self.cs, self.setup_coro_worker_value_holder = prepare_loop(self.setup_coro_worker, *self.setup_coro_worker_args, **self.setup_coro_worker_kwargs)

def iter_cs(self): View Source

65    def iter_cs(self):
66        in_work = self.cs.iteration()

def draw(self, force: bool = False) -> None: View Source

68    def draw(self, force: bool = False) -> None:
69        self.ensure_cs()
70        self.iter_cs()
71        super().draw(force)

Writes composited screen to the terminal.

At the moment this uses full-screen rewrites. There is a compositing implementation in composite, but it is currently not performant enough to use.

Args: force: When set, new composited lines will not be checked against the previous ones, and everything will be redrawn.

def put_coro( self, coro_worker: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.ExplicitWorker, collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Any], collections.abc.Callable[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, typing.Awaitable[typing.Any]]], *args, **kwargs): View Source

73    def put_coro(self, coro_worker: AnyWorker, *args, **kwargs):
74        put_coro_to((self.cs, None, False), coro_worker, *args, **kwargs)

def stop(self) -> None: View Source

76    def stop(self) -> None:
77        async def coro(i: Interface, self):
78            def shutdown_handler():
79                self._is_running = False
80            
81            i._loop.on_destroyed_handlers.add(shutdown_handler)
82            await i(ShutdownLoop)
83        
84        self.put_coro(coro, self)

Stops the compositor.

Inherited Members

pytermgui.window_manager.compositor.Compositor: Compositor; fps; framerate; terminal; clear_cache; run; composite; set_redraw; redraw; capture

class TerminalApplication: View Source

104class TerminalApplication:
105    def __init__(self):
106        self.model = self.model_setup()
107        self.manager: WindowManagerCS = None
108    
109    def run(self):
110        with PTGWindowManager() as manager:
111            self.manager = manager
112            WindowManagerCS.replace(manager, self.model, self.ui_setup, self.controller_loop_setup)
113    
114    def __call__(self):
115        self.run()
116    
117    def model_setup(self):
118        raise NotImplementedError()
119    
120    def ui_setup(self, manager: WindowManagerCS, model):
121        raise NotImplementedError()
122    
123    def controller_loop_setup(self, i: Interface, model):
124        raise NotImplementedError()

model

manager: WindowManagerCS

def run(self): View Source

109    def run(self):
110        with PTGWindowManager() as manager:
111            self.manager = manager
112            WindowManagerCS.replace(manager, self.model, self.ui_setup, self.controller_loop_setup)

def model_setup(self): View Source

117    def model_setup(self):
118        raise NotImplementedError()

def ui_setup( self, manager: WindowManagerCS, model): View Source

120    def ui_setup(self, manager: WindowManagerCS, model):
121        raise NotImplementedError()

def controller_loop_setup( self, i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, model): View Source

123    def controller_loop_setup(self, i: Interface, model):
124        raise NotImplementedError()

TermApp = <class 'TerminalApplication'>