<commit_before>#!/usr/bin/env python
# coding=utf-8
from __future__ import division, print_function, unicode_literals
import hashlib
import json
import os

import sacred.optional as opt
from sacred.commandline_options import CommandLineOption
from sacred.dependencies import get_digest
from sacred.observers.base import RunObserver
from sacred.serializer import flatten, restore

DEFAULT_SQL_PRIORITY = 40


# ############################# Observer #################################### #

class SqlObserver(RunObserver):
    @classmethod
    def create(cls, url, echo=False, priority=DEFAULT_SQL_PRIORITY):
        engine = sa.create_engine(url, echo=echo)
        return cls(engine, sessionmaker(bind=engine)(), priority)

    def __init__(self, engine, session, priority=DEFAULT_SQL_PRIORITY):
        self.engine = engine
        self.session = session
        self.priority = priority
        self.run = None

    def started_event(self, ex_info, command, host_info, start_time, config,
                      meta_info, _id):
        Base.metadata.create_all(self.engine)
        sql_exp = Experiment.get_or_create(ex_info, self.session)
        sql_host = Host.get_or_create(host_info, self.session)
        if _id is None:
            i = self.session.query(Run).order_by(Run.id.desc()).first()
            _id = 0 if i is None else i.id + 1

        self.run = Run(run_id=str(_id),
                       start_time=start_time,
                       config=json.dumps(flatten(config)),
                       command=command,
                       priority=meta_info.get('priority', 0),
                       comment=meta_info.get('comment', ''),
                       experiment=sql_exp,
                       host=sql_host,
                       status='RUNNING')
        self.session.add(self.run)
        self.session.commit()
        return _id or self.run.run_id

    def queued_event(self, ex_info, command, queue_time, config, meta_info,
                     _id):

        Base.metadata.create_all(self.engine)
        sql_exp = Experiment.get_or_create(ex_info, self.session)
        if _id is None:
            i = self.session.query(Run).order_by(Run.id.desc()).first()
            _id = 0 if i is None else i.id + 1

        self.run = Run(run_id=str(_id),
                       config=json.dumps(flatten(config)),
                       command=command,
                       priority=meta_info.get('priority', 0),
                       comment=meta_info.get('comment', ''),
                       experiment=sql_exp,
                       status='QUEUED')
        self.session.add(self.run)
        self.session.commit()
        return _id or self.run.run_id

    def heartbeat_event(self, info, captured_out, beat_time, result):
        self.run.info = json.dumps(flatten(info))
        self.run.captured_out = captured_out
        self.run.heartbeat = beat_time
        self.run.result = result
        self.session.commit()

    def completed_event(self, stop_time, result):
        self.run.stop_time = stop_time
        self.run.result = result
        self.run.status = 'COMPLETED'
        self.session.commit()

    def interrupted_event(self, interrupt_time, status):
        self.run.stop_time = interrupt_time
        self.run.status = status
        self.session.commit()

    def failed_event(self, fail_time, fail_trace):
        self.run.stop_time = fail_time
        self.run.fail_trace = '\n'.join(fail_trace)
        self.run.status = 'FAILED'
        self.session.commit()

    def resource_event(self, filename):
        res = Resource.get_or_create(filename, self.session)
        self.run.resources.append(res)
        self.session.commit()

    def artifact_event(self, name, filename):
        a = Artifact.create(name, filename)
        self.run.artifacts.append(a)
        self.session.commit()

    def query(self, _id):
        run = self.session.query(Run).filter_by(id=_id).first()
        return run.to_json()

    def __eq__(self, other):
        if isinstance(other, SqlObserver):
            # fixme: this will probably fail to detect two equivalent engines
            return (self.engine == other.engine and
                    self.session == other.session)
        return False

    def __ne__(self, other):
        return not self.__eq__(other)
<commit_msg>added a lock to the SQLObserver.<commit_after>#!/usr/bin/env python
# coding=utf-8
from __future__ import division, print_function, unicode_literals
import hashlib
import json
import os
from threading import Lock

import sacred.optional as opt
from sacred.commandline_options import CommandLineOption
from sacred.dependencies import get_digest
from sacred.observers.base import RunObserver
from sacred.serializer import flatten, restore

DEFAULT_SQL_PRIORITY = 40


# ############################# Observer #################################### #

class SqlObserver(RunObserver):
    @classmethod
    def create(cls, url, echo=False, priority=DEFAULT_SQL_PRIORITY):
        engine = sa.create_engine(url, echo=echo)
        return cls(engine, sessionmaker(bind=engine)(), priority)

    def __init__(self, engine, session, priority=DEFAULT_SQL_PRIORITY):
        self.engine = engine
        self.session = session
        self.priority = priority
        self.run = None
        self.lock = Lock()

    def started_event(self, ex_info, command, host_info, start_time, config,
                      meta_info, _id):
        Base.metadata.create_all(self.engine)
        sql_exp = Experiment.get_or_create(ex_info, self.session)
        sql_host = Host.get_or_create(host_info, self.session)
        if _id is None:
            i = self.session.query(Run).order_by(Run.id.desc()).first()
            _id = 0 if i is None else i.id + 1

        self.run = Run(run_id=str(_id),
                       start_time=start_time,
                       config=json.dumps(flatten(config)),
                       command=command,
                       priority=meta_info.get('priority', 0),
                       comment=meta_info.get('comment', ''),
                       experiment=sql_exp,
                       host=sql_host,
                       status='RUNNING')
        self.session.add(self.run)
        self.save()
        return _id or self.run.run_id

    def queued_event(self, ex_info, command, queue_time, config, meta_info,
                     _id):

        Base.metadata.create_all(self.engine)
        sql_exp = Experiment.get_or_create(ex_info, self.session)
        if _id is None:
            i = self.session.query(Run).order_by(Run.id.desc()).first()
            _id = 0 if i is None else i.id + 1

        self.run = Run(run_id=str(_id),
                       config=json.dumps(flatten(config)),
                       command=command,
                       priority=meta_info.get('priority', 0),
                       comment=meta_info.get('comment', ''),
                       experiment=sql_exp,
                       status='QUEUED')
        self.session.add(self.run)
        self.save()
        return _id or self.run.run_id

    def heartbeat_event(self, info, captured_out, beat_time, result):
        self.run.info = json.dumps(flatten(info))
        self.run.captured_out = captured_out
        self.run.heartbeat = beat_time
        self.run.result = result
        self.save()

    def completed_event(self, stop_time, result):
        self.run.stop_time = stop_time
        self.run.result = result
        self.run.status = 'COMPLETED'
        self.save()

    def interrupted_event(self, interrupt_time, status):
        self.run.stop_time = interrupt_time
        self.run.status = status
        self.save()

    def failed_event(self, fail_time, fail_trace):
        self.run.stop_time = fail_time
        self.run.fail_trace = '\n'.join(fail_trace)
        self.run.status = 'FAILED'
        self.save()

    def resource_event(self, filename):
        res = Resource.get_or_create(filename, self.session)
        self.run.resources.append(res)
        self.save()

    def artifact_event(self, name, filename):
        a = Artifact.create(name, filename)
        self.run.artifacts.append(a)
        self.save()

    def save(self):
        with self.lock:
            self.session.commit()

    def query(self, _id):
        run = self.session.query(Run).filter_by(id=_id).first()
        return run.to_json()

    def __eq__(self, other):
        if isinstance(other, SqlObserver):
            # fixme: this will probably fail to detect two equivalent engines
            return (self.engine == other.engine and
                    self.session == other.session)
        return False

    def __ne__(self, other):
        return not self.__eq__(other)

{'fraction_non_alphanumeric': 0.11391244556497823, 'fraction_numerical': 0.00206280082512033, 'mean_word_length': 2.2060984570168993, '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': '4893622', 'n_tokens_mistral': 2558, 'n_tokens_neox': 2468, 'n_words': 530}