• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

SwissDataScienceCenter / renku-python / 5529030370

pending completion
5529030370

push

github-actions

Ralf Grubenmann
fix docker build, pin versions

24252 of 28479 relevant lines covered (85.16%)

2.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

16.15
/renku/ui/cli/utils/curses.py
1
# -*- coding: utf-8 -*-
2
#
3
# Copyright 2017-2022 - Swiss Data Science Center (SDSC)
4
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
5
# Eidgenössische Technische Hochschule Zürich (ETHZ).
6
#
7
# Licensed under the Apache License, Version 2.0 (the "License");
8
# you may not use this file except in compliance with the License.
9
# You may obtain a copy of the License at
10
#
11
#     http://www.apache.org/licenses/LICENSE-2.0
12
#
13
# Unless required by applicable law or agreed to in writing, software
14
# distributed under the License is distributed on an "AS IS" BASIS,
15
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16
# See the License for the specific language governing permissions and
17
# limitations under the License.
18
"""Curses utilities."""
2✔
19
import curses
2✔
20
import curses.panel
2✔
21
from typing import Dict, List, Optional, Tuple
2✔
22

23
from renku.command.view_model.text_canvas import Point
2✔
24
from renku.core import errors
2✔
25
from renku.domain_model.provenance.activity import Activity
2✔
26

27

28
class CursesActivityGraphViewer:
2✔
29
    """An interactive viewer for activity graphs."""
30

31
    COLOR_MAPPING = {
2✔
32
        "[31": curses.COLOR_RED,
33
        "[32": curses.COLOR_GREEN,
34
        "[33": curses.COLOR_YELLOW,
35
        "[34": curses.COLOR_BLUE,
36
        "[35": curses.COLOR_MAGENTA,
37
        "[36": curses.COLOR_CYAN,
38
        "[37": curses.COLOR_WHITE,
39
        "[0": -1,
40
    }
41

42
    ACTIVITY_OVERLAY_WIDTH = 60
2✔
43
    ACTIVITY_OVERLAY_HEIGHT = 40
2✔
44
    HELP_OVERLAY_WIDTH = 60
2✔
45
    HELP_OVERLAY_HEIGHT = 6
2✔
46
    DATE_FORMAT = "%Y-%m-%d %H:%M:S"
2✔
47

48
    def __init__(
2✔
49
        self,
50
        text_data: str,
51
        navigation_data: List[List[Tuple[Point, Point, Activity]]],
52
        vertical_space: int,
53
        use_color: bool,
54
    ):
55
        self.text_data = text_data
×
56
        self.navigation_data = navigation_data
×
57
        self.vertical_space = vertical_space
×
58

59
        self.current_layer = 0
×
60
        self.layer_position = 0
×
61
        self.max_layer = len(navigation_data) - 1
×
62
        self.y_pos = 0
×
63
        self.x_pos = 0
×
64
        self.color_cache: Dict[str, int] = {}
×
65
        self.activity_overlay: Optional[curses._CursesWindow] = None
×
66
        self.help_overlay: Optional[curses._CursesWindow] = None
×
67
        self._select_activity()
×
68
        self.use_color = use_color
×
69

70
    def _init_curses(self, screen):
2✔
71
        """Initialize curses screen for interactive mode."""
72
        screen.keypad(True)
×
73
        curses.use_default_colors()
×
74
        curses.start_color()
×
75
        curses.curs_set(0)
×
76
        curses.noecho()
×
77
        screen.refresh()
×
78
        self._init_console_colors()
×
79
        self.rows, self.cols = screen.getmaxyx()
×
80

81
        min_width = max(self.ACTIVITY_OVERLAY_WIDTH, self.HELP_OVERLAY_WIDTH) + 1
×
82
        min_height = max(self.ACTIVITY_OVERLAY_HEIGHT, self.HELP_OVERLAY_HEIGHT) + 1
×
83
        if self.cols < min_width or self.rows < min_height:
×
84
            raise errors.TerminalSizeError(
×
85
                f"Terminal is too small for interactive mode, size should be at least {min_width}x{min_height}."
86
            )
87

88
        text_data_lines = self.text_data.splitlines()
×
89

90
        self.content_max_x = max(len(line) for line in text_data_lines)
×
91
        self.content_max_y = len(self.text_data)
×
92
        self.content_pad = curses.newpad(self.content_max_y, self.content_max_x)
×
93
        for i, l in enumerate(text_data_lines):
×
94
            self._addstr_with_color_codes(self.content_pad, i, 0, l)
×
95

96
        self._blink_text(self.content_pad, self.activity_start, self.activity_end, bold=True)
×
97

98
        self._update_help_overlay(screen)
×
99

100
        self._refresh(screen)
×
101

102
    def _init_console_colors(self):
2✔
103
        """Setup curses color mapping."""
104
        if not self.use_color:
×
105
            curses.init_pair(100, -1, -1)
×
106
            for color_symbol, _ in self.COLOR_MAPPING.items():
×
107
                self.color_cache[color_symbol] = 100
×
108
        else:
109
            for i, (color_symbol, color) in enumerate(self.COLOR_MAPPING.items(), start=100):
×
110
                curses.init_pair(i, color, -1)
×
111
                self.color_cache[color_symbol] = i
×
112

113
    def _select_activity(self):
2✔
114
        """Set the currently selected activity."""
115
        start, end, self.selected_activity = self.navigation_data[self.current_layer][self.layer_position]
×
116

117
        # Ignore borders, we only care about text
118
        self.activity_start = Point(start.x + 1, start.y + 1)
×
119
        self.activity_end = Point(end.x - 1, end.y - 1)
×
120

121
    def _addstr_with_color_codes(self, window, y: int, x: int, text: str):
2✔
122
        """Replace ANSI color codes with curses colors."""
123
        if not curses.has_colors():
×
124
            window.addstr(y, x, text)
×
125
            return
×
126

127
        split_text = text.split("\033")
×
128

129
        # add first part without color
130
        window.addstr(y, x, split_text[0], curses.color_pair(self.color_cache["[0"]))
×
131

132
        x += len(split_text[0])
×
133

134
        for substring in split_text[1:]:
×
135
            color, snippet = substring.split("m", 1)
×
136

137
            if len(snippet) == 0:
×
138
                continue
×
139

140
            if color == "[1":
×
141
                curses_color = curses.A_BOLD
×
142
            else:
143
                curses_color = curses.color_pair(self.color_cache[color])
×
144

145
            window.addstr(y, x, snippet, curses_color)
×
146

147
            x += len(snippet)
×
148

149
    def _blink_text(self, window, start: Point, end: Point, bold: bool = True):
2✔
150
        """Change text between start and end to blinking."""
151
        style = curses.A_BLINK
×
152

153
        if bold:
×
154
            style |= curses.A_BOLD
×
155

156
        for y in range(start.y, end.y + 1):
×
157
            text = window.instr(y, start.x, end.x - start.x + 1)
×
158
            window.addstr(y, start.x, text, style)
×
159

160
    def _unblink_text(self, window, start: Point, end: Point, bold: bool = True):
2✔
161
        """Change text between start and end to not-blinking."""
162

163
        for y in range(start.y, end.y + 1):
×
164
            text = window.instr(y, start.x, end.x - start.x + 1)
×
165
            if bold:
×
166
                window.addstr(y, start.x, text, curses.A_BOLD)
×
167
            else:
168
                window.addstr(y, start.x, text)
×
169

170
    def _add_multiline_text_with_wrapping(self, window, text: str):
2✔
171
        """Add a multiline text to a window, wrapping text to fit."""
172
        height, width = window.getmaxyx()
×
173

174
        # don't write to window borders
175
        width -= 2
×
176
        i = 1
×
177

178
        for line in text.splitlines():
×
179
            chunks = [line[p : p + width] for p in range(0, len(line), width)]
×
180
            if not chunks:
×
181
                # Lines containing only \n
182
                i += 1
×
183
                continue
×
184

185
            for chunk in chunks:
×
186
                if i >= height:
×
187
                    # TODO: Add scrolling using a pad?
188
                    return
×
189
                window.addstr(i, 1, chunk)
×
190
                i += 1
×
191

192
    def _refresh(self, screen):
2✔
193
        """Refresh curses screens/pads/windows/panels."""
194
        self.content_pad.refresh(self.y_pos, self.x_pos, 0, 0, self.rows - 1, self.cols - 1)
×
195

196
        if self.activity_overlay:
×
197
            self.activity_overlay.overlay(screen)
×
198
            self.activity_overlay.refresh()
×
199

200
        if self.help_overlay:
×
201
            self.help_overlay.overlay(screen)
×
202
            self.help_overlay.refresh()
×
203

204
        if self.activity_overlay or self.help_overlay:
×
205
            curses.panel.update_panels()
×
206

207
    def _change_layer(self, step: int):
2✔
208
        """Change the currently active layer."""
209
        self.current_layer = max(min(self.current_layer + step, self.max_layer), 0)
×
210
        self.layer_position = max(min(self.layer_position, len(self.navigation_data[self.current_layer]) - 1), 0)
×
211
        del self.activity_overlay
×
212
        self.activity_overlay = None
×
213

214
    def _change_layer_position(self, step: int):
2✔
215
        """Change position inside current layer."""
216
        self.layer_position = max(min(self.layer_position + step, len(self.navigation_data[self.current_layer]) - 1), 0)
×
217
        del self.activity_overlay
×
218
        self.activity_overlay = None
×
219

220
    def _update_activity_overlay(self, screen):
2✔
221
        """Show/Hide the activity overlay."""
222
        if not self.activity_overlay:
×
223
            del self.help_overlay
×
224
            self.help_overlay = None
×
225

226
            self.activity_overlay = curses.newwin(
×
227
                self.ACTIVITY_OVERLAY_HEIGHT,
228
                self.ACTIVITY_OVERLAY_WIDTH,
229
                self.rows - self.ACTIVITY_OVERLAY_HEIGHT,
230
                self.cols - self.ACTIVITY_OVERLAY_WIDTH,
231
            )
232
            curses.panel.new_panel(self.activity_overlay)
×
233
            self.activity_overlay.border()
×
234

235
            started_date = self.selected_activity.started_at_time.strftime(self.DATE_FORMAT)
×
236
            ended_date = self.selected_activity.ended_at_time.strftime(self.DATE_FORMAT)
×
237
            usages = "\n".join(u.entity.path for u in self.selected_activity.usages)
×
238
            generations = "\n".join(g.entity.path for g in self.selected_activity.generations)
×
239
            agents = ", ".join(getattr(a, "full_name", a.name) for a in self.selected_activity.agents)
×
240
            full_command = " ".join(self.selected_activity.plan_with_values.to_argv(with_streams=True))
×
241

242
            content = (
×
243
                "Id:\n"
244
                f"{self.selected_activity.id}\n\n"
245
                "Started:\n"
246
                f"{started_date}\n\n"
247
                "Ended:\n"
248
                f"{ended_date}\n\n"
249
                "Agents:\n"
250
                f"{agents}\n\n"
251
                "Plan Id:\n"
252
                f"{self.selected_activity.association.plan.id}\n\n"
253
                "Plan Name:\n"
254
                f"{self.selected_activity.association.plan.name}\n\n"
255
                "Inputs:\n"
256
                f"{usages}\n\n"
257
                "Outputs:\n"
258
                f"{generations}\n\n"
259
                "Full Command:\n"
260
                f"{full_command}\n\n"
261
            )
262

263
            self._add_multiline_text_with_wrapping(self.activity_overlay, content)
×
264
            self.activity_overlay.overlay(screen)
×
265
        else:
266
            del self.activity_overlay
×
267
            self.activity_overlay = None
×
268

269
    def _update_help_overlay(self, screen):
2✔
270
        """Show/hide the help overlay."""
271
        if not self.help_overlay:
×
272
            del self.activity_overlay
×
273
            self.activity_overlay = None
×
274

275
            self.help_overlay = curses.newwin(
×
276
                self.HELP_OVERLAY_HEIGHT, self.HELP_OVERLAY_WIDTH, 0, self.cols - self.HELP_OVERLAY_WIDTH
277
            )
278
            curses.panel.new_panel(self.help_overlay)
×
279
            self.help_overlay.border()
×
280

281
            content = (
×
282
                "Navigate using arrow keys\n"
283
                "Press <enter> to show activity details\n"
284
                "Press <h> to show/hide this help\n"
285
                "Press <q> to exit\n"
286
            )
287
            self._add_multiline_text_with_wrapping(self.help_overlay, content)
×
288
            self.help_overlay.overlay(screen)
×
289
        else:
290
            del self.help_overlay
×
291
            self.help_overlay = None
×
292

293
    def _move_viewscreen(self):
2✔
294
        """Move viewscreen to include selected activity."""
295
        if self.activity_start.x - 1 < self.x_pos:
×
296
            self.x_pos = max(self.activity_start.x - 1, 0)
×
297
        elif self.activity_end.x + 1 > self.x_pos + self.cols:
×
298
            self.x_pos = min(self.activity_end.x + 1, self.content_max_x) - self.cols - 1
×
299

300
        if self.activity_start.y - 1 < self.y_pos:
×
301
            self.y_pos = max(self.activity_start.y - self.vertical_space - 3, 0)
×
302
        elif self.activity_end.y + 1 > self.y_pos + self.rows:
×
303
            self.y_pos = min(self.activity_end.y + self.vertical_space + 3, self.content_max_y) - self.rows - 1
×
304

305
    def _loop(self, screen):
2✔
306
        """The interaction loop."""
307
        running = True
×
308

309
        while running:
×
310
            input_char = screen.getch()
×
311

312
            # update screen size to deal with resizes
313
            self.rows, self.cols = screen.getmaxyx()
×
314

315
            # handle keypress
316
            if input_char == curses.KEY_DOWN or chr(input_char) == "k":
×
317
                self._change_layer(1)
×
318
            elif input_char == curses.KEY_UP or chr(input_char) == "i":
×
319
                self._change_layer(-1)
×
320
            elif input_char == curses.KEY_RIGHT or chr(input_char) == "l":
×
321
                self._change_layer_position(1)
×
322
            elif input_char == curses.KEY_LEFT or chr(input_char) == "j":
×
323
                self._change_layer_position(-1)
×
324
            elif input_char == curses.KEY_ENTER or input_char == 10 or input_char == 13:
×
325
                self._update_activity_overlay(screen)
×
326
            elif chr(input_char) == "h":
×
327
                self._update_help_overlay(screen)
×
328
            elif input_char < 256 and chr(input_char) == "q":
×
329
                running = False
×
330

331
            self._unblink_text(self.content_pad, self.activity_start, self.activity_end, bold=True)
×
332
            self._select_activity()
×
333
            self._blink_text(self.content_pad, self.activity_start, self.activity_end, bold=True)
×
334

335
            self._move_viewscreen()
×
336

337
            self._refresh(screen)
×
338

339
    def _main(self, screen):
2✔
340
        """The main execution method for wrapped curses execution."""
341
        self._init_curses(screen)
×
342
        self._loop(screen)
×
343

344
    def run(self):
2✔
345
        """Run interactive curses mode."""
346
        curses.wrapper(self._main)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc