# Path: antevents/base.py
# class DefaultSubscriber:
# class CallableAsSubscriber:
# class FatalError(Exception):
# class InvalidTopicError(FatalError):
# class UnknownTopicError(FatalError):
# class TopicAlreadyClosed(FatalError):
# class ExcInDispatch(FatalError):
# class Publisher:
# class Filter(Publisher, DefaultSubscriber):
# class _ThunkBuilder:
# class DirectPublisherMixin:
# class EventLoopPublisherMixin:
# class IterableAsPublisher(Publisher, DirectPublisherMixin):
# class FunctionIteratorAsPublisher(Publisher, DirectPublisherMixin):
# class SensorPub(Publisher, DirectPublisherMixin):
# class BlockingSubscriber:
# class _ThreadForBlockingSubscriber(threading.Thread):
# class _ThreadForBlockingPublisher(threading.Thread):
# class ScheduleError(FatalError):
# class Scheduler:
#     def on_next(self, x):
#     def on_error(self, e):
#     def on_completed(self):
# def _on_next_name(topic):
# def _on_error_name(topic):
# def _on_completed_name(topic):
#     def __init__(self, on_next=None, on_error=None, on_completed=None,
#                  topic=None):
#             def default_error(err):
#     def __str__(self):
#     def __init__(self, topics=None):
#     def subscribe(self, subscriber, topic_mapping=None):
#         def dispose():
#     def _schedule(self, enqueue_fn):
#     def _close_topic(self, topic):
#     def _dispatch_next(self, x, topic=None):
#     def _dispatch_completed(self, topic=None):
#     def _dispatch_error(self, e, topic=None):
#     def print_downstream(self):
#         def has_subscribers(step):
#         def print_from(current_seq, step):
#     def pp_subscribers(self):
#     def __init__(self, previous_in_chain,
#                  on_next=None, on_completed=None,
#                  on_error=None, name=None):
#     def on_next(self, x):
#     def on_error(self, e):
#     def on_completed(self):
#     def __str__(self):
# def _is_thunk(t):
# def _make_thunk(t):
#     def __init__(self, func):
#     def __call__(self, *args, **kwargs):
#         def apply(this):
#     def __repr__(self):
# def _subscribe_thunk(prev, thunk):
# def filtermethod(base, alias=None):
#     def inner(func):
#     def _observe(self):
#     def _observe_event_loop(self):
#     def _stop_loop(self):
#     def __init__(self, iterable, name=None):
#     def _observe(self):
#     def _close(self):
#     def __str__(self):
# def from_iterable(i):
# def from_list(l):
#     def __init__(self, initial_state, condition, iterate, result_selector):
#     def _observe(self):
# def from_func(init, cond, iter, selector):
#     def __str__(self):
# def make_sensor_event(sensor, sample):
#     def __init__(self, sensor, make_event_fn=make_sensor_event):
#     def _observe(self):
#     def __repr__(self):
#     def __init__(self, scheduler, topics=None):
#         def start():
#     def request_stop(self):
#     def _wait_and_dispatch(self):
#     def _on_next(self, topic, x):
#     def _on_completed(self, topic):
#     def _on_error(self, topic, e):
#     def _close(self):
#     def __init__(self, subscriber, scheduler):
#     def run(self):
#             def die(): # need to stop the scheduler in the main loop
#             def done():
#     def __init__(self, publisher, interval, scheduler):
#     def _stop_loop(self):
#     def run(self):
#         def enqueue_fn(fn, *args):
#             def die(): # need to stop the scheduler in the main loop
#             def done():
#     def __init__(self, event_loop):
#         def exception_handler(loop, context):
#     def _remove_from_active_schedules(self, publisher):
#     def schedule_periodic(self, publisher, interval):
#         def cancel():
#         def run():
#     def schedule_sensor(self, sensor, interval, *subscriber_sequence,
#                         make_event_fn=make_sensor_event,
#                         print_downstream=False):
#     def schedule_recurring(self, publisher):
#         def cancel():
#         def run():
#     def schedule_on_main_event_loop(self, publisher):
#         def stop():
#     def schedule_on_private_event_loop(self, publisher):
#         def enqueue_fn(fn, *args):
#         def thread_main():
#                 def die(): # need to stop the scheduler in the main loop
#                 def loop_done():
#     def schedule_periodic_on_separate_thread(self, publisher, interval):
#     def schedule_sensor_on_separate_thread(self, sensor, interval, *subscriber_sequence,
#                                            make_event_fn=make_sensor_event):
#     def schedule_later_one_time(self, publisher, interval):
#         def cancel():
#         def run():
#     def run_forever(self):
#     def _schedule_coroutine(self, coro, done_callback):
#         def cb(f):
#     def stop(self):
#                 def recheck_stop(f):


from statistics import median
from antevents.base import SensorEvent
import json
import asyncio
import random
import time
import hbmqtt.client
        self.client = hbmqtt.client.MQTTClient(loop=event_loop)
        self.connected = False

    def _to_message(self, msg):
        return bytes(json.dumps((msg.sensor_id, msg.ts, msg.val),), encoding='utf-8')
    
    async def send(self, msg):
        if not self.connected:
            print("attempting connection")
            await self.client.connect(self.url)
            self.connected = True
            print("connected")
        print("sending %s" % self._to_message(msg))
        await self.client.publish(self.topic, self._to_message(msg))

    async def disconnect(self):
        if self.connected:
            await self.client.disconnect()


async def sample_and_process(sensor, mqtt_writer, xducer):
    try:
        sample = sensor.sample()
    except StopIteration:
        final_event = xducer.complete()
        if final_event:
            await mqtt_writer.send(final_event)
        print("disconnecting")
        await mqtt_writer.disconnect()
        return False

# Path: antevents/base.py
# class DefaultSubscriber:
# class CallableAsSubscriber:
# class FatalError(Exception):
# class InvalidTopicError(FatalError):
# class UnknownTopicError(FatalError):
# class TopicAlreadyClosed(FatalError):
# class ExcInDispatch(FatalError):
# class Publisher:
# class Filter(Publisher, DefaultSubscriber):
# class _ThunkBuilder:
# class DirectPublisherMixin:
# class EventLoopPublisherMixin:
# class IterableAsPublisher(Publisher, DirectPublisherMixin):
# class FunctionIteratorAsPublisher(Publisher, DirectPublisherMixin):
# class SensorPub(Publisher, DirectPublisherMixin):
# class BlockingSubscriber:
# class _ThreadForBlockingSubscriber(threading.Thread):
# class _ThreadForBlockingPublisher(threading.Thread):
# class ScheduleError(FatalError):
# class Scheduler:
#     def on_next(self, x):
#     def on_error(self, e):
#     def on_completed(self):
# def _on_next_name(topic):
# def _on_error_name(topic):
# def _on_completed_name(topic):
#     def __init__(self, on_next=None, on_error=None, on_completed=None,
#                  topic=None):
#             def default_error(err):
#     def __str__(self):
#     def __init__(self, topics=None):
#     def subscribe(self, subscriber, topic_mapping=None):
#         def dispose():
#     def _schedule(self, enqueue_fn):
#     def _close_topic(self, topic):
#     def _dispatch_next(self, x, topic=None):
#     def _dispatch_completed(self, topic=None):
#     def _dispatch_error(self, e, topic=None):
#     def print_downstream(self):
#         def has_subscribers(step):
#         def print_from(current_seq, step):
#     def pp_subscribers(self):
#     def __init__(self, previous_in_chain,
#                  on_next=None, on_completed=None,
#                  on_error=None, name=None):
#     def on_next(self, x):
#     def on_error(self, e):
#     def on_completed(self):
#     def __str__(self):
# def _is_thunk(t):
# def _make_thunk(t):
#     def __init__(self, func):
#     def __call__(self, *args, **kwargs):
#         def apply(this):
#     def __repr__(self):
# def _subscribe_thunk(prev, thunk):
# def filtermethod(base, alias=None):
#     def inner(func):
#     def _observe(self):
#     def _observe_event_loop(self):
#     def _stop_loop(self):
#     def __init__(self, iterable, name=None):
#     def _observe(self):
#     def _close(self):
#     def __str__(self):
# def from_iterable(i):
# def from_list(l):
#     def __init__(self, initial_state, condition, iterate, result_selector):
#     def _observe(self):
# def from_func(init, cond, iter, selector):
#     def __str__(self):
# def make_sensor_event(sensor, sample):
#     def __init__(self, sensor, make_event_fn=make_sensor_event):
#     def _observe(self):
#     def __repr__(self):
#     def __init__(self, scheduler, topics=None):
#         def start():
#     def request_stop(self):
#     def _wait_and_dispatch(self):
#     def _on_next(self, topic, x):
#     def _on_completed(self, topic):
#     def _on_error(self, topic, e):
#     def _close(self):
#     def __init__(self, subscriber, scheduler):
#     def run(self):
#             def die(): # need to stop the scheduler in the main loop
#             def done():
#     def __init__(self, publisher, interval, scheduler):
#     def _stop_loop(self):
#     def run(self):
#         def enqueue_fn(fn, *args):
#             def die(): # need to stop the scheduler in the main loop
#             def done():
#     def __init__(self, event_loop):
#         def exception_handler(loop, context):
#     def _remove_from_active_schedules(self, publisher):
#     def schedule_periodic(self, publisher, interval):
#         def cancel():
#         def run():
#     def schedule_sensor(self, sensor, interval, *subscriber_sequence,
#                         make_event_fn=make_sensor_event,
#                         print_downstream=False):
#     def schedule_recurring(self, publisher):
#         def cancel():
#         def run():
#     def schedule_on_main_event_loop(self, publisher):
#         def stop():
#     def schedule_on_private_event_loop(self, publisher):
#         def enqueue_fn(fn, *args):
#         def thread_main():
#                 def die(): # need to stop the scheduler in the main loop
#                 def loop_done():
#     def schedule_periodic_on_separate_thread(self, publisher, interval):
#     def schedule_sensor_on_separate_thread(self, sensor, interval, *subscriber_sequence,
#                                            make_event_fn=make_sensor_event):
#     def schedule_later_one_time(self, publisher, interval):
#         def cancel():
#         def run():
#     def run_forever(self):
#     def _schedule_coroutine(self, coro, done_callback):
#         def cb(f):
#     def stop(self):
#                 def recheck_stop(f):


# Path: examples/event_library_comparison/asyncawait.py
from statistics import median
from antevents.base import SensorEvent
import json
import asyncio
import random
import time
import hbmqtt.client

        self.client = hbmqtt.client.MQTTClient(loop=event_loop)
        self.connected = False

    def _to_message(self, msg):
        return bytes(json.dumps((msg.sensor_id, msg.ts, msg.val),), encoding='utf-8')
    
    async def send(self, msg):
        if not self.connected:
            print("attempting connection")
            await self.client.connect(self.url)
            self.connected = True
            print("connected")
        print("sending %s" % self._to_message(msg))
        await self.client.publish(self.topic, self._to_message(msg))

    async def disconnect(self):
        if self.connected:
            await self.client.disconnect()


async def sample_and_process(sensor, mqtt_writer, xducer):
    try:
        sample = sensor.sample()
    except StopIteration:
        final_event = xducer.complete()
        if final_event:
            await mqtt_writer.send(final_event)
        print("disconnecting")
        await mqtt_writer.disconnect()
        return False

    event = SensorEvent(sensor_id=sensor.sensor_id, ts=time.time(), val=sample)