MCPcopy Create free account
hub / github.com/dot-agent/nextpy / _update_display

Method _update_display

nextpy/ai/engine/_program.py:510–576  ·  view source on GitHub ↗

Updates the display with the current marked text after debouncing. Parameters ---------- last : bool If True, this is the last update and we should clear the send queue and prepare the UI for saving etc. force : bool If True, we wi

(self, last=False)

Source from the content-addressed store, hash-verified

508 yield self
509
510 def _update_display(self, last=False):
511 """Updates the display with the current marked text after debouncing.
512
513 Parameters
514 ----------
515 last : bool
516 If True, this is the last update and we should clear the send queue and prepare the
517 UI for saving etc.
518 force : bool
519 If True, we will update the display even if it would otherwise be throttled.
520 """
521
522 log.debug(
523 f"Updating display (last={last}, self._displaying={self._displaying}, self._comm={self._comm})"
524 )
525
526 if self.stream:
527 if self.async_mode:
528 # if we are streaming in async mode then we set the event to let the generator know it can yield
529 self._emit_stream_event.set()
530
531 else:
532 # if we are streaming not in async mode then we pause the event loop to let the generator
533 # that is controlling execution return (it will restart the event loop when it is ready)
534 if self._executor is not None:
535 asyncio.get_event_loop().stop()
536
537 # this is always called during execution, and we only want to update the display if we are displaying
538 if not self._displaying:
539 return
540
541 # debounce the display updates
542 # now = time.time()
543 # log.debug(now - self._last_display_update)
544 # debounce_delay = self.display_throttle_limit if self._comm and self._comm.is_open else self.display_throttle_limit_low
545 # if last or (now - self._last_display_update > debounce_delay):
546 if self._displaying_html:
547 out = self._build_html(self.marked_text)
548
549 # clear the send queue if this is the last update
550 if last and self._comm:
551 self._comm.clear_send_queue()
552
553 # send an update to the front end client if we have one...
554 # TODO: we would like to call `display` for the last update so NB saving works, but see https://github.com/microsoft/vscode-jupyter/issues/13243
555 if (
556 self._displayed and self._comm and self._comm.is_open
557 ): # (not last or self._comm.is_open):
558 log.debug(f"Updating display send message to front end")
559 # log.debug(out)
560 self._comm.send({"replace": out})
561 if last:
562 self._comm.send({"event": "complete"})
563
564 # ...otherwise dump the client to the front end
565 else:
566 log.debug(f"Updating display dump to front end")
567 from IPython.display import clear_output

Callers

nothing calls this directly

Calls 6

_build_htmlMethod · 0.95
_display_htmlMethod · 0.95
clear_send_queueMethod · 0.80
setMethod · 0.45
stopMethod · 0.45
sendMethod · 0.45

Tested by

no test coverage detected