<commit_before>""" Module containing the main node types of the data flow graph
"""
import asyncio

class BaseNode:
    """ Base class for all node types
    """

    def __init__(self):
        self._watchers = set()
        self._dependants = set()
        self._dirty = False
        self._value = None 

    def set_dirty(self):
        self._dirty = True
        for node in self._dependants:
            node.set_dirty()
        for watcher in self._watchers:
            # Create tasks that evaluates asynchronosly
            task = asyncio.ensure_future(watcher.evaluate())
    
    def add_dependant(self, other):
        self._dependants.add(other)


class FuncNode(BaseNode):
    """ Node that calls a function
    """
    def __init__(self, func, *args):
        super().__init__()
        self._func = func
        self._args = args
        for arg in args:
            arg.add_dependant(self)

    async def value(self):
        if not self._dirty:
            return self._value
        args = []
        for arg in self._args:
            value = await arg.value()
            args.append(value)
        self._value = self._func(*args)
        self._dirty = False
        return self._value

    def add_watcher(self, watcher):
        self._watchers.add(watcher)

class SourceNode(BaseNode):
    """ A node that is a source of values. The source object is queue like and should
    inplement an async get method. The source should be monitored by a coroutine
    that will also set the SourceNode to dirty.
    """
    def __init__(self, source):                
        super().__init__()
        self._source = source
    
    async def value(self):
        return self._value

async def source_node_monitor(node, source):
    try:
        print(source)
        while True:
            message = await source.get()
            if message:
                result = message.json()['price']
                message.ack()
                node._value = result
                node.set_dirty()
    except asyncio.CancelledError:
        # Finish silently on cancel
        pass

class PrintWatcher:
    def __init__(self, node):
        self.watched_node = node
        node.add_watcher(self)

    async def evaluate(self):
        value = await self.watched_node.value()
        print('Value:', value)



<commit_msg>some cleanup@<commit_after>""" Module containing the main node types of the data flow graph
"""
import asyncio

class BaseNode:
    """ Base class for all node types
    """

    def __init__(self):
        self._dependants = set()
        self._update_event = None
        self._cached_value = None
        self._dirty = True

    def notify_update(self):
        self._dirty = True
        for node in self._dependants:
            node.notify_update()
        if self.calculate_event:
            self.calculate_event.set()
    
    def add_dependant(self, other):
        self._dependants.add(other)

    def subscribe(self):
        if self._update_event is None:
            self._update_event = asyncio.Event()
        return self._update_event

class FuncNode(BaseNode):
    """ Node that calls a function
    """
    def __init__(self, func, *args):
        super().__init__()
        self._func = func
        self._args = args
        for arg in args:
            arg.add_dependant(self)

    def value(self):
        if not self._dirty:
            return self._cached_value
        args = []
        for arg in self._args:
            value = arg.value()
            args.append(value)
        self._cached_value = self._func(*args)
        self._dirty = False
        return self._cached_value

class SourceNode(BaseNode):
    """ A node that is a source of values. 
    The source object is queue like and should implement get
    """
    def __init__(self, source):                
        super().__init__()
        self.source = source
    
    def value(self):
        return self._cached_value

    def set_value(self, value):
        self._cached_value = value
        self.notify_update()

    async def monitor(self):
        try:
            print(source)
            while True:
                value = await self.source.get()
                self.set_value(value)
        except asyncio.CancelledError:
            # Finish silently on cancel
            pass

async def print_watcher(node):
    update_event = node.subscribe()
    while True:
        await update_event
        value = node.value()
        print(value)


class PrintWatcher:
    def __init__(self, node):
        self.watched_node = node
        node.add_watcher(self)

    async def evaluate(self):
        value = await self.watched_node.value()
        print('Value:', value)




{'fraction_non_alphanumeric': 0.08814783347493628, 'fraction_numerical': 0.0, 'mean_word_length': 1.90679012345679, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6395642', 'n_tokens_mistral': 1315, 'n_tokens_neox': 1283, 'n_words': 379}