Source code for pisak.text_tools

Text operations-related tools.
import threading

from gi.repository import GObject, Clutter

from pisak import properties, configurator

[docs]class Predictor(Clutter.Actor, properties.PropertyAdapter, configurator.Configurable): """ Base class for objects that follow changes in the given target text and supply suggestions based on the text context. Searching through a predictor database happens in another thread. Properties: * :attr:`target` """ __gsignals__ = { "content_update": ( GObject.SIGNAL_RUN_FIRST, None, ()), "processing_on": ( GObject.SIGNAL_RUN_FIRST, None, ()) } __gproperties__ = { "target": ( Clutter.Actor.__gtype__, "target to follow", "id of text box to follow", GObject.PARAM_READWRITE) } def __init__(self): super().__init__() = None self.content = []
[docs] def get_suggestion(self, accuracy_level): """ Get single suggestion from the list of all generated by the predictor. :param accuracy_level: what level of prediction accuracy is required for the requested suggestion item. :return: single item from the predictor content list. """ if accuracy_level < len(self.content): return self.content[accuracy_level]
[docs] def notify_content_update(self): """ Notify that the predictor content has been updated. """ Clutter.threads_add_idle(0, self.emit, "content-update")
[docs] def do_prediction(self, text, position): """ Method that performs the proper action of prediction, based on the given text feed. When all the processing is done and content has been updated successfully, then it should call the 'notify_content_update' method. Method to be overwritten by child. :param text: text being the base for prediction. :param position: current position in the text. """ raise NotImplementedError
def _update_content(self, *args): self.emit("processing-on") text = position = worker = threading.Thread( target=self.do_prediction, args=(text, position), daemon=True) worker.start() def _follow_target(self): if is not None: text_field = text_field.connect("text-changed", self._update_content) def _stop_following_target(self): try: if is not None: text_field = text_field.disconnect_by_func("text-changed", self._update_content) except AttributeError: return None @property def target(self): """ Target object containing text to be watched by the predictor. """ return self._target @target.setter def target(self, value): self._stop_following_target() self._target = value self._follow_target()