<commit_before>#!/usr/bin/env python3
import time
import bisect
import asyncio
from collections import UserList, defaultdict
from concurrent.futures import Executor, ThreadPoolExecutor
from phasortoolbox import Client
import logging

LOG=logging.getLogger('phasortoolbox.pdc')


class Synchrophasor(UserList):
    """docstring for Synchrophasor"""
    def __init__(self, list_, time_tag, arr_time, perf_counter):
        super(Synchrophasor, self).__init__(list_)
        self.time = time_tag
        self.arr_time = arr_time
        self.perf_counter = perf_counter


class PDC(object):
    """docstring for PDC


Example:
my_pdc = PDC()

This class aligns in coming synchrophasor messages using time tags.


    """
    def __init__(self, callback=lambda buf_sync: None, clients=[], time_out=0.1, steps=1, return_on_time_out=False, process_pool=False):
        self.callback = callback
        self.clients = clients
        self.receive_counter = 0
        self.time_out = time_out
        self.steps = steps
        self.return_on_time_out = return_on_time_out
        self.process_pool = process_pool

    def callback(self, buf_sync):
        """
        Implement your function
        """
        pass

    def run(self, c=0, loop=None, executor=None):
        self.set_loop(loop, executor)
        self.loop.create_task(self.coro_run(c))
        try:
            self.loop.run_forever()
        except KeyboardInterrupt:
            pass
        finally:
            self.loop.run_until_complete(self.coro_close())
            self.checkreceive_counter()
        self.loop.close()


    async def coro_run(self, c=0):
        self.receive_counter = 0
        self.c = c
        idcode_list = [client.idcode for client in self.clients]
        if len(idcode_list) > len(set(idcode_list)):
            raise Exception('Duplicate id_code found. C37.118.2 standard does not support duplicate id_code. idcode list:',idcode_list)
            
        self._buf = _Buffer(idcode_list, self._synchrophasors_created, self.time_out, self.steps, self.return_on_time_out)
        self._buf_task = asyncio.ensure_future(self._buf.coro_check_timeout())
        for client in self.clients:
            client._add_pdc(id(self), self._buf.add_msg, self.loop, self.executor)
            asyncio.ensure_future(client.coro_run())

    async def coro_close(self):
        if self._buf_task :
            self._buf_task.cancel()
        for client in self.clients:
            await client.coro_close()
            client._remove_pdc(id(self))

    def checkreceive_counter(self):
        LOG.warning(str(self.receive_counter)+' synchrophasors created.')
        for client in self.clients:
            client.checkreceive_counter()

    def _synchrophasors_created(self, synchrophasors):
        self.receive_counter += 1
        if self.process_pool:
            future = self.executor.submit(self.callback, synchrophasors)
            if future.exception():
                raise future.exception()
        else:
            self.callback(synchrophasors)
        del(synchrophasors)
        if self.c == 0:
            return
        elif self.c > 1:
            self.c -= 1
            return
        elif self.c == 1:
            self.loop.stop()

    def set_loop(self, loop=None, executor=None):
        self.loop = loop if loop is not None else asyncio.new_event_loop()
        self.executor = executor if executor is not None else ThreadPoolExecutor()



class _Buffer(object):
    def __init__(self, idcode_list, callback, time_out, steps, return_on_time_out):
        self.idcode_list = idcode_list
        self.callback = callback
        self.time_out = time_out
        self.steps = steps
        self.return_on_time_out = return_on_time_out
        self._buffer_time_out = 0.2 * (steps + 1)
        self._min_sleep = time_out/100
        self._data = defaultdict(lambda: defaultdict(lambda: None))
        self._arr_times = defaultdict(lambda: 0)
        self._perf_counter = defaultdict(lambda: 0)
        self._return_time_outs = {}
        self._buffer_time_outs = {}
        self._sorted_time_tags = []
        self._ready_to_send = []
        self._ready_to_send_s = set()
        self._last_sent_time_tag = None

    def add_msg(self, msg):
        if msg.time not in self._arr_times:
            bisect.insort(self._sorted_time_tags, msg.time)
        self._data[msg.time][msg.idcode] = msg
        self._arr_times[msg.time] = max(msg.arr_time,self._arr_times[msg.time])
        self._perf_counter[msg.time] = max(msg.perf_counter,self._perf_counter[msg.time])
        self._return_time_outs[msg.time] = self._arr_times[msg.time] + self.time_out
        self._buffer_time_outs[msg.time] = self._arr_times[msg.time] + self._buffer_time_out
        self._check_completeness(msg.time)

    def _check_completeness(self, time_tag):
        if len(self._data[time_tag]) == len(self.idcode_list) and time_tag not in self._ready_to_send_s:
            bisect.insort(self._ready_to_send, time_tag)
            self._ready_to_send_s.add(time_tag)
            self._send_synchrophasors_if_ready()

    def _send_synchrophasors_if_ready(self):
        if len(self._ready_to_send) >= self.steps:
            if self._ready_to_send[-1] != self._last_sent_time_tag:
                synchrophasors = []
                for i in reversed(range(1, self.steps+1)):
                    synchrophasors.append(Synchrophasor([self._data[self._ready_to_send[-i]][idcode] for idcode in self.idcode_list], self._ready_to_send[-i], self._arr_times[self._ready_to_send[-i]], self._perf_counter[self._ready_to_send[-i]]))
                self._last_sent_time_tag = self._ready_to_send[-1]
                self.callback(synchrophasors)
        
    def callback(self, synchrophasors):
        pass

    async def coro_check_timeout(self):
        while True:
            try:
                await asyncio.sleep(self._min_sleep)<commit_msg>Change variable name steps to history<commit_after>#!/usr/bin/env python3
import time
import bisect
import asyncio
from collections import UserList, defaultdict
from concurrent.futures import Executor, ThreadPoolExecutor
from phasortoolbox import Client
import logging

LOG=logging.getLogger('phasortoolbox.pdc')


class Synchrophasor(UserList):
    """
    time_tag is the time tag of the synchrophasor message
    arr_time is the unix time that the last data frame for the synchrophasor received 
    perf_counter is used the check the performace
    """
    def __init__(self, list_, time_tag, arr_time, perf_counter):
        super(Synchrophasor, self).__init__(list_)
        self.time = time_tag
        self.arr_time = arr_time
        self.perf_counter = perf_counter 


class PDC(object):
    """docstring for PDC


Example:
my_pdc = PDC()

This class aligns in coming synchrophasor messages using time tags.


    """
    def __init__(self, callback=lambda buf_sync: None, clients=[], time_out=0.1, history=1, return_on_time_out=False, process_pool=False):
        self.callback = callback
        self.clients = clients
        self.receive_counter = 0
        self.time_out = time_out
        self.history = history
        self.return_on_time_out = return_on_time_out
        self.process_pool = process_pool

    def callback(self, buf_sync):
        """
        Implement your function
        """
        pass

    def run(self, c=0, loop=None, executor=None):
        self.set_loop(loop, executor)
        self.loop.create_task(self.coro_run(c))
        try:
            self.loop.run_forever()
        except KeyboardInterrupt:
            pass
        finally:
            self.loop.run_until_complete(self.coro_close())
            self.checkreceive_counter()
        self.loop.close()


    async def coro_run(self, c=0):
        self.receive_counter = 0
        self.c = c
        idcode_list = [client.idcode for client in self.clients]
        if len(idcode_list) > len(set(idcode_list)):
            raise Exception('Duplicate id_code found. C37.118.2 standard does not support duplicate id_code. idcode list:',idcode_list)
            
        self._buf = _Buffer(idcode_list, self._synchrophasors_created, self.time_out, self.history, self.return_on_time_out)
        self._buf_task = asyncio.ensure_future(self._buf.coro_check_timeout())
        for client in self.clients:
            client._add_pdc(id(self), self._buf.add_msg, self.loop, self.executor)
            asyncio.ensure_future(client.coro_run())

    async def coro_close(self):
        if self._buf_task :
            self._buf_task.cancel()
        for client in self.clients:
            await client.coro_close()
            client._remove_pdc(id(self))

    def checkreceive_counter(self):
        LOG.warning(str(self.receive_counter)+' synchrophasors created.')
        for client in self.clients:
            client.checkreceive_counter()

    def _synchrophasors_created(self, synchrophasors):
        self.receive_counter += 1
        if self.process_pool:
            future = self.executor.submit(self.callback, synchrophasors)
            if future.exception():
                raise future.exception()
        else:
            self.callback(synchrophasors)
        del(synchrophasors)
        if self.c == 0:
            return
        elif self.c > 1:
            self.c -= 1
            return
        elif self.c == 1:
            self.loop.stop()

    def set_loop(self, loop=None, executor=None):
        self.loop = loop if loop is not None else asyncio.new_event_loop()
        self.executor = executor if executor is not None else ThreadPoolExecutor()



class _Buffer(object):
    def __init__(self, idcode_list, callback, time_out, history, return_on_time_out):
        self.idcode_list = idcode_list
        self.callback = callback
        self.time_out = time_out
        self.history = history
        self.return_on_time_out = return_on_time_out
        self._buffer_time_out = 0.2 * (history + 1)
        self._min_sleep = time_out/100
        self._data = defaultdict(lambda: defaultdict(lambda: None))
        self._arr_times = defaultdict(lambda: 0)
        self._perf_counter = defaultdict(lambda: 0)
        self._return_time_outs = {}
        self._buffer_time_outs = {}
        self._sorted_time_tags = []
        self._ready_to_send = []
        self._ready_to_send_s = set()
        self._last_sent_time_tag = None

    def add_msg(self, msg):
        if msg.time not in self._arr_times:
            bisect.insort(self._sorted_time_tags, msg.time)
        self._data[msg.time][msg.idcode] = msg
        self._arr_times[msg.time] = max(msg.arr_time,self._arr_times[msg.time])
        self._perf_counter[msg.time] = max(msg.perf_counter,self._perf_counter[msg.time])
        self._return_time_outs[msg.time] = self._arr_times[msg.time] + self.time_out
        self._buffer_time_outs[msg.time] = self._arr_times[msg.time] + self._buffer_time_out
        self._check_completeness(msg.time)

    def _check_completeness(self, time_tag):
        if len(self._data[time_tag]) == len(self.idcode_list) and time_tag not in self._ready_to_send_s:
            bisect.insort(self._ready_to_send, time_tag)
            self._ready_to_send_s.add(time_tag)
            self._send_synchrophasors_if_ready()

    def _send_synchrophasors_if_ready(self):
        if len(self._ready_to_send) >= self.history:
            if self._ready_to_send[-1] != self._last_sent_time_tag:
                synchrophasors = []
                for i in reversed(range(1, self.history+1)):
                    synchrophasors.append(Synchrophasor([self._data[self._ready_to_send[-i]][idcode] for idcode in self.idcode_list], self._ready_to_send[-i], self._arr_times[self._ready_to_send[-i]], self._perf_counter[self._ready_to_send[-i]]))
                self._last_sent_time_tag = self._ready_to_send[-1]
                self.callback(synchrophasors)
        
    def callback(self, synchrophasors):
        pass

    async def coro_check_timeout(self):
        while True:
            try:
                await asyncio.sleep(self._min_sleep)
{'fraction_non_alphanumeric': 0.09585406301824212, 'fraction_numerical': 0.005140961857379768, 'mean_word_length': 2.6035255452644157, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '5790955', 'n_tokens_mistral': 3896, 'n_tokens_neox': 3790, 'n_words': 733}