<commit_before>from mirte.core import Module

from sarah.event import Event

import os
import base64
import logging
import threading

class JoyceError(Exception):
	pass
class HijackedChannel(JoyceError):
	pass
class UnsupportedProtocol(JoyceError):
	pass

class JoyceChannel(object):
	def __init__(self, relay, token, logger):
		self.relay = relay
		self.l = logger
		self.token = token
		self.on_message = Event()
	def send_message(self, d):
		self.relay.send_message(self.token, d)
	def handle_message(self, d):
		self.on_message(self, d)
	def close(self):
		self.relay.close_channel(self.token)

class JoyceRelay(object):
	def __init__(self, hub, logger):
		self.l = logger
		self.hub = hub
	def send_message(self, token, d):
		raise NotImplementedError
	def handle_message(self, token, d):
		self.hub.handle_message(token, d, self)
	def close_channel(self, token):
		self.hub.remove_channel(token)

class JoyceHub(Module):
	def __init__(self, channel_class=None, *args, **kwargs):
		super(JoyceHub, self).__init__(*args, **kwargs)
		self.channel_class = (JoyceChannel
				if channel_class is None else channel_class)
		self.lock = threading.Lock()
		self.channel_to_relay = dict()
		self.relay_to_channels = dict()
		self.channels = dict()
		self.on_new_channel = Event()
	def _get_channel_for_relay(self, token, relay):
		new_channel = False
		with self.lock:
			if not token in self.channels:
				c = self._create_channel(token, relay)
				new_channel = True
			else:
				if relay != self.channel_to_relay[token]:
					raise HijackedChannel
				c = self.channels[token]
		if new_channel:
			self.on_new_channel(c)
		return c
	def handle_message(self, token, d, relay):
		c = self._get_channel_for_relay(token, relay)
		c.handle_message(d)
	def broadcast_message(self, d):
		with self.lock:
			channels = self.channels.values()
		for channel in channels:
			channel.send_message(d)
	
	def _create_channel(self, token, relay, channel_class=None):
		if token is None:<commit_msg>base: add stream API<commit_after>from mirte.core import Module

from sarah.event import Event

import os
import base64
import logging
import threading

class JoyceError(Exception):
	pass
class HijackedChannel(JoyceError):
	pass
class UnsupportedProtocol(JoyceError):
	pass

class JoyceChannel(object):
	def __init__(self, relay, token, logger):
		self.relay = relay
		self.l = logger
		self.token = token
		self.on_message = Event()
		self.on_stream = Event()
	def send_stream(self, stream, blocking=True):
		self.relay.send_stream(self.token, stream, blocking)
	def send_message(self, d):
		self.relay.send_message(self.token, d)
	def handle_message(self, d):
		self.on_message(self, d)
	def handle_stream(self, stream):
		self.on_stream(self, stream)
	def close(self):
		self.relay.close_channel(self.token)

class JoyceRelay(object):
	def __init__(self, hub, logger):
		self.l = logger
		self.hub = hub
	def send_message(self, token, d):
		raise NotImplementedError
	def send_stream(self, token, stream, blocking=True):
		raise NotImplementedError
	def handle_message(self, token, d):
		self.hub.handle_message(token, d, self)
	def handle_stream(self, token, stream):
		self.hub.handle_stream(token, d, self)
	def close_channel(self, token):
		self.hub.remove_channel(token)

class JoyceHub(Module):
	def __init__(self, channel_class=None, *args, **kwargs):
		super(JoyceHub, self).__init__(*args, **kwargs)
		self.channel_class = (JoyceChannel
				if channel_class is None else channel_class)
		self.lock = threading.Lock()
		self.channel_to_relay = dict()
		self.relay_to_channels = dict()
		self.channels = dict()
		self.on_new_channel = Event()
	def _get_channel_for_relay(self, token, relay):
		new_channel = False
		with self.lock:
			if not token in self.channels:
				c = self._create_channel(token, relay)
				new_channel = True
			else:
				if relay != self.channel_to_relay[token]:
					raise HijackedChannel
				c = self.channels[token]
		if new_channel:
			self.on_new_channel(c)
		return c
	def handle_stream(self, token, stream, relay):
		c = self._get_channel_for_relay(token, relay)
		c.handle_stream(stream)
	def handle_message(self, token, d, relay):
		c = self._get_channel_for_relay(token, relay)
		c.handle_message(d)
	def broadcast_message(self, d):
		with self.lock:
			channels = self.channels.values()
		for channel in channels:
			channel.send_message(d)
	
	def _create_channel(self, token, relay, channel_class=None):
		if token is None:
{'fraction_non_alphanumeric': 0.10784093462143339, 'fraction_numerical': 0.0008986744551786115, 'mean_word_length': 4.834862385321101, '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': '559133', 'n_tokens_mistral': 1668, 'n_tokens_neox': 1528, 'n_words': 377}