<commit_before>import sys
import os
import tempfile
import shutil
# import zlib
import json
import traceback
from time import sleep

from redis import StrictRedis
from redis.connection import ConnectionError
from figment.entity import Entity
from figment.logger import log

def fatal(message):
    log.error('Fatal: %s' % message)
    sys.exit(1)


class Zone(object):
    def __init__(self):
        self.id = None
        self.entities = {}
        self.ticking_entities = set()
        self.tick_interval = 1
        self.running = False
        self.redis = None

    @classmethod
    def from_config(cls, id, config_path='config.json'):
        self = cls()

        self.id = id
        self.working_dir = '.'

        self.load_config(config_path)

        return self

    @property
    def tick_key(self):
        return 'zone:%s:tick' % self.id

    @property
    def incoming_key(self):
        return 'zone:%s:incoming' % self.id

    @property
    def import_key(self):
        return 'zone:%s:imports' % self.id

    def load_config(self, path):
        full_path = os.path.abspath(os.path.expanduser(path))

        try:
            with open(full_path) as f:
                config = json.loads(f.read())
        except EnvironmentError:
            fatal("couldn't read configuration file %s" % path)
        except ValueError as e:
            fatal("error in configuration file: %s" % e.message)

        if not self.id in config['zones']:
            fatal("undefined zone '%s'" % self.id)

        tick_interval = config['zones'][self.id].get('tick', 1)
        self.tick_interval = tick_interval

        # TODO: per-zone persistence settings

        persistence = config.get('persistence')
        if not persistence:
            fatal('unspecified persistence settings')

        # TODO: alternate persistence modes

        if not persistence['mode'] == 'snapshot':
            fatal("unrecognized persistence mode '%s'" % persistence['mode'])

        self.config = config
        self.working_dir = os.path.dirname(full_path)

        # TODO: Read redis connection params from config
        self.redis = StrictRedis()

    @property
    def snapshot_path(self):
        snapshot_path = self.config['persistence']['file']
        try:
            snapshot_path = snapshot_path % self.id
        except TypeError:
            pass
        return os.path.expanduser(snapshot_path)

    def load_snapshot(self):
        if not os.path.exists(self.snapshot_path):
            return False

        log.info('Loading entities from snapshot.')
        with open(self.snapshot_path, 'r') as f:
            snapshot = f.read()
            # if self.config['persistence'].get('compressed'):
            #     snapshot = zlib.decompress(snapshot)
            snapshot = json.loads(snapshot)
            for entity_dict in snapshot['entities']:
                entity = Entity.from_dict(entity_dict, zone=self)

        return True

    def save_snapshot(self):
        log.info('Saving snapshot: %s' % self.snapshot_path)
        child_pid = os.fork()

        if not child_pid:
            f = tempfile.NamedTemporaryFile(delete=False)
            snapshot = json.dumps({
                'entities': [e.to_dict() for e in self.all()]
            })
            # if self.config['persistence'].get('compressed'):
            #     snapshot = zlib.compress(snapshot)
            f.write(snapshot)
            f.close()
            shutil.move(f.name, self.snapshot_path)
            os._exit(os.EX_OK)

    def load_components(self):
        # HACK: add basedir of the config file to the import path
        sys.path.append(self.working_dir)
        # As a side effect, Component.ALL gets populated with Component subclasses
        __import__('components')

    def start(self):
        try:
            self.redis.ping()
        except ConnectionError as e:
            fatal("Redis error: %s" % e.message)

        self.running = True
        log.info('Listening.')

        # Clear any existing tick events
        self.redis.ltrim(self.tick_key, 0, 0)
        try:
            while self.running:
                self.process_one_event()
        except Exception as e:
            log.error(traceback.format_exc())
        except BaseException as e:
            pass

        self.save_snapshot()

    def restart(self):
        # TODO: Make sure this actually works
        log.info('Restarting.')
        python = sys.executable
        os.execl(python, python, *sys.argv)

    def stop(self):
        self.running = False

    def start_ticker(self):
        log.info('Ticking every %ss.' % self.tick_interval)
        while True:
            log.debug('Tick.')
            # TODO: timestamp here instead of True, for debugging?
            self.redis.rpush(self.tick_key, True)
            sleep(self.tick_interval)

    def listen(self, entity_id):
        pubsub = self.redis.pubsub()
        pubsub.subscribe(Entity.messages_key_from_id(entity_id))<commit_msg>Add silly clock noises to ticker debug output<commit_after>import sys
import os
import tempfile
import shutil
# import zlib
import json
import traceback
from time import sleep

from redis import StrictRedis
from redis.connection import ConnectionError
from figment.entity import Entity
from figment.logger import log

def fatal(message):
    log.error('Fatal: %s' % message)
    sys.exit(1)


class Zone(object):
    def __init__(self):
        self.id = None
        self.entities = {}
        self.ticking_entities = set()
        self.tick_interval = 1
        self.running = False
        self.redis = None

    @classmethod
    def from_config(cls, id, config_path='config.json'):
        self = cls()

        self.id = id
        self.working_dir = '.'

        self.load_config(config_path)

        return self

    @property
    def tick_key(self):
        return 'zone:%s:tick' % self.id

    @property
    def incoming_key(self):
        return 'zone:%s:incoming' % self.id

    @property
    def import_key(self):
        return 'zone:%s:imports' % self.id

    def load_config(self, path):
        full_path = os.path.abspath(os.path.expanduser(path))

        try:
            with open(full_path) as f:
                config = json.loads(f.read())
        except EnvironmentError:
            fatal("couldn't read configuration file %s" % path)
        except ValueError as e:
            fatal("error in configuration file: %s" % e.message)

        if not self.id in config['zones']:
            fatal("undefined zone '%s'" % self.id)

        tick_interval = config['zones'][self.id].get('tick', 1)
        self.tick_interval = tick_interval

        # TODO: per-zone persistence settings

        persistence = config.get('persistence')
        if not persistence:
            fatal('unspecified persistence settings')

        # TODO: alternate persistence modes

        if not persistence['mode'] == 'snapshot':
            fatal("unrecognized persistence mode '%s'" % persistence['mode'])

        self.config = config
        self.working_dir = os.path.dirname(full_path)

        # TODO: Read redis connection params from config
        self.redis = StrictRedis()

    @property
    def snapshot_path(self):
        snapshot_path = self.config['persistence']['file']
        try:
            snapshot_path = snapshot_path % self.id
        except TypeError:
            pass
        return os.path.expanduser(snapshot_path)

    def load_snapshot(self):
        if not os.path.exists(self.snapshot_path):
            return False

        log.info('Loading entities from snapshot.')
        with open(self.snapshot_path, 'r') as f:
            snapshot = f.read()
            # if self.config['persistence'].get('compressed'):
            #     snapshot = zlib.decompress(snapshot)
            snapshot = json.loads(snapshot)
            for entity_dict in snapshot['entities']:
                entity = Entity.from_dict(entity_dict, zone=self)

        return True

    def save_snapshot(self):
        log.info('Saving snapshot: %s' % self.snapshot_path)
        child_pid = os.fork()

        if not child_pid:
            f = tempfile.NamedTemporaryFile(delete=False)
            snapshot = json.dumps({
                'entities': [e.to_dict() for e in self.all()]
            })
            # if self.config['persistence'].get('compressed'):
            #     snapshot = zlib.compress(snapshot)
            f.write(snapshot)
            f.close()
            shutil.move(f.name, self.snapshot_path)
            os._exit(os.EX_OK)

    def load_components(self):
        # HACK: add basedir of the config file to the import path
        sys.path.append(self.working_dir)
        # As a side effect, Component.ALL gets populated with Component subclasses
        __import__('components')

    def start(self):
        try:
            self.redis.ping()
        except ConnectionError as e:
            fatal("Redis error: %s" % e.message)

        self.running = True
        log.info('Listening.')

        # Clear any existing tick events
        self.redis.ltrim(self.tick_key, 0, 0)
        try:
            while self.running:
                self.process_one_event()
        except Exception as e:
            log.error(traceback.format_exc())
        except BaseException as e:
            pass

        self.save_snapshot()

    def restart(self):
        # TODO: Make sure this actually works
        log.info('Restarting.')
        python = sys.executable
        os.execl(python, python, *sys.argv)

    def stop(self):
        self.running = False

    def start_ticker(self):
        log.info('Ticking every %ss.' % self.tick_interval)
        tock = False
        while True:
            log.debug('Tock.' if tock else 'Tick.')
            # TODO: timestamp here instead of True, for debugging?
            self.redis.rpush(self.tick_key, True)
            sleep(self.tick_interval)
            tock = not tock

    def listen(self, entity_id):
        pubsub = self.redis.pubsub()
        pubsub.subscribe(Entity.messages_key_from_id(entity_id))
{'fraction_non_alphanumeric': 0.10248015873015873, 'fraction_numerical': 0.000992063492063492, 'mean_word_length': 2.090435315757204, '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': '1303218', 'n_tokens_mistral': 2908, 'n_tokens_neox': 2862, 'n_words': 719}