@@ -0,0 +1,3478 @@
+# The MIT License (MIT) # Copyright (c) 2014-2017 University of Bristol%0A#%0A#  Permission is hereby granted, free of charge, to any person obtaining a copy%0A#  of this software and associated documentation files (the %22Software%22), to deal%0A#  in the Software without restriction, including without limitation the rights%0A#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell%0A#  copies of the Software, and to permit persons to whom the Software is%0A#  furnished to do so, subject to the following conditions:%0A#%0A#  The above copyright notice and this permission notice shall be included in all%0A#  copies or substantial portions of the Software.%0A#%0A#  THE SOFTWARE IS PROVIDED %22AS IS%22, WITHOUT WARRANTY OF ANY KIND,%0A#  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF%0A#  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.%0A#  IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,%0A#  DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR%0A#  OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE%0A#  OR OTHER DEALINGS IN THE SOFTWARE.%0A%0Afrom hyperstream import TimeInterval, RelativeTimeInterval%0Afrom hyperstream.tool import Tool, check_input_stream_count%0Afrom datetime import timedelta%0Aimport logging%0A%0A%0Aclass Aggregate(Tool):%0A    def __init__(self, func):%0A        super(Aggregate, self).__init__(func=func)%0A        self.func = func%0A%0A    # noinspection PyCompatibility%0A    @check_input_stream_count(2)%0A    def _execute(self, input_streams, interval, writer):%0A%0A        timer = input_streams%5B0%5D%0A        # TODO: Check that this is a timer stream%0A%0A        data_stream = input_streams%5B1%5D%0A%0A        rel_start = timedelta(0)%0A        rel_end = timedelta(0)%0A        if isinstance(data_stream.time_interval, RelativeTimeInterval):%0A            rel_start = data_stream.time_interval.start%0A            rel_end = data_stream.time_interval.end%0A        elif isinstance(data_stream.time_interval, TimeInterval):%0A            if interval != data_stream.time_interval:%0A                # TODO: What if stream and input stream have different absolute intervals?%0A                logging.error(%22interval %7B%7D != input_streams%5B0%5D.time_interval %7B%7D%22.format(%0A                    interval, data_stream.time_interval))%0A                raise NotImplementedError%0A%0A        data = data_stream.window(interval).iteritems()%0A%0A        window = %5B%5D%0A        future = %5B%5D%0A        for (t, _) in timer.window(interval).iteritems():%0A            while (len(window) %3E 0) and (window%5B0%5D%5B0%5D %3C= t + rel_start):%0A                window = window%5B1:%5D%0A            while (len(future) %3E 0) and (future%5B0%5D%5B0%5D %3C= t + rel_end):%0A                doc = future%5B0%5D%0A                future = future%5B1:%5D%0A                if t + rel_start %3C doc%5B0%5D %3C= t + rel_end:%0A                    window.append(doc)%0A            while True:%0A                try:%0A                    doc = next(data)%0A                    if t + rel_start %3C doc%5B0%5D %3C= t + rel_end:%0A                        window.append(doc)%0A                    elif doc%5B0%5D %3E t + rel_end:%0A                        future.append(doc)%0A                        break%0A                    else:%0A                        pass%0A                except StopIteration:%0A                    break%0A            # single-document case:%0A            writer(%5B(t, self.func(iter(window)))%5D)%0A            # multi-document case:%0A            # for x in func( (doc for doc in window) ):%0A            #        writer(%5B(t,x)%5D)%0A

Update to aggregate tool (removed timer as parameter)