<s> from __future__ import division , print_function , unicode_literals <EOL> import os . path <EOL> import pickle <EOL> import re <EOL> import sys <EOL> import time <EOL> import bson <EOL> import gridfs <EOL> import pymongo <EOL> import sacred . optional as opt <EOL> from pymongo . errors import AutoReconnect , InvalidDocument <EOL> from pymongo . son_manipulator import SONManipulator <EOL> from sacred . commandline_options import CommandLineOption <EOL> from sacred . dependencies import get_digest <EOL> from sacred . observers . base import RunObserver <EOL> from sacred . utils import ObserverError <EOL> SON_MANIPULATORS = [ ] <EOL> if opt . has_numpy : <EOL> class NumpyArraysToList ( SONManipulator ) : <EOL> """<STR_LIT>""" <EOL> def transform_incoming ( self , son , collection ) : <EOL> for ( key , value ) in son . items ( ) : <EOL> if isinstance ( value , opt . np . ndarray ) : <EOL> son [ key ] = value . tolist ( ) <EOL> elif isinstance ( value , dict ) : <EOL> son [ key ] = self . transform_incoming ( value , collection ) <EOL> return son <EOL> SON_MANIPULATORS . append ( NumpyArraysToList ( ) ) <EOL> if opt . has_pandas : <EOL> pd = opt . pandas <EOL> import json <EOL> class PandasToJson ( SONManipulator ) : <EOL> """<STR_LIT>""" <EOL> def transform_incoming ( self , son , collection ) : <EOL> for ( key , value ) in son . items ( ) : <EOL> if isinstance ( value , ( pd . Series , pd . DataFrame , pd . Panel ) ) : <EOL> son [ key ] = json . loads ( value . to_json ( ) ) <EOL> elif isinstance ( value , dict ) : <EOL> son [ key ] = self . transform_incoming ( value , collection ) <EOL> return son <EOL> SON_MANIPULATORS . append ( PandasToJson ( ) ) <EOL> def force_valid_bson_key ( key ) : <EOL> key = str ( key ) <EOL> if key . startswith ( '<STR_LIT:$>' ) : <EOL> key = '<STR_LIT:@>' + key [ <NUM_LIT:1> : ] <EOL> key = key . replace ( '<STR_LIT:.>' , '<STR_LIT:U+002C>' ) <EOL> return key <EOL> def force_bson_encodeable ( obj ) : <EOL> if isinstance ( obj , dict ) : <EOL> try : <EOL> bson . BSON . encode ( obj , check_keys = True ) <EOL> return obj <EOL> except bson . InvalidDocument : <EOL> return { force_valid_bson_key ( k ) : force_bson_encodeable ( v ) <EOL> for k , v in obj . items ( ) } <EOL> elif opt . has_numpy and isinstance ( obj , opt . np . ndarray ) : <EOL> return obj <EOL> else : <EOL> try : <EOL> bson . BSON . encode ( { '<STR_LIT>' : obj } ) <EOL> return obj <EOL> except bson . InvalidDocument : <EOL> return str ( obj ) <EOL> class MongoObserver ( RunObserver ) : <EOL> @ staticmethod <EOL> def create ( url = '<STR_LIT:localhost>' , db_name = '<STR_LIT>' , prefix = '<STR_LIT:default>' , ** kwargs ) : <EOL> client = pymongo . MongoClient ( url , ** kwargs ) <EOL> database = client [ db_name ] <EOL> for manipulator in SON_MANIPULATORS : <EOL> database . add_son_manipulator ( manipulator ) <EOL> runs_collection = database [ prefix + '<STR_LIT>' ] <EOL> fs = gridfs . GridFS ( database , collection = prefix ) <EOL> return MongoObserver ( runs_collection , fs ) <EOL> def __init__ ( self , runs_collection , fs ) : <EOL> self . runs = runs_collection <EOL> self . fs = fs <EOL> self . run_entry = None <EOL> def save ( self ) : <EOL> try : <EOL> self . runs . save ( self . run_entry ) <EOL> except AutoReconnect : <EOL> pass <EOL> except InvalidDocument : <EOL> raise ObserverError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def final_save ( self , attempts = <NUM_LIT:10> ) : <EOL> for i in range ( attempts ) : <EOL> try : <EOL> self . runs . save ( self . run_entry ) <EOL> return <EOL> except AutoReconnect : <EOL> if i < attempts - <NUM_LIT:1> : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> except InvalidDocument : <EOL> self . run_entry = force_bson_encodeable ( self . run_entry ) <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> file = sys . stderr ) <EOL> from tempfile import NamedTemporaryFile <EOL> with NamedTemporaryFile ( suffix = '<STR_LIT>' , delete = False , <EOL> prefix = '<STR_LIT>' ) as f : <EOL> pickle . dump ( self . run_entry , f ) <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( f . name ) , <EOL> file = sys . stderr ) <EOL> def started_event ( self , ex_info , host_info , start_time , config , comment ) : <EOL> self . run_entry = { <EOL> '<STR_LIT>' : dict ( ex_info ) , <EOL> '<STR_LIT:host>' : dict ( host_info ) , <EOL> '<STR_LIT>' : start_time , <EOL> '<STR_LIT>' : config , <EOL> '<STR_LIT>' : comment , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:info>' : { } , <EOL> '<STR_LIT>' : None <EOL> } <EOL> self . save ( ) <EOL> for source_name , md5 in ex_info [ '<STR_LIT>' ] : <EOL> if not self . fs . exists ( filename = source_name , md5 = md5 ) : <EOL> with open ( source_name , '<STR_LIT:rb>' ) as f : <EOL> self . fs . put ( f , filename = source_name ) <EOL> def heartbeat_event ( self , info , captured_out , beat_time ) : <EOL> self . run_entry [ '<STR_LIT:info>' ] = info <EOL> self . run_entry [ '<STR_LIT>' ] = captured_out <EOL> self . run_entry [ '<STR_LIT>' ] = beat_time <EOL> self . save ( ) <EOL> def completed_event ( self , stop_time , result ) : <EOL> self . run_entry [ '<STR_LIT>' ] = stop_time <EOL> self . run_entry [ '<STR_LIT:result>' ] = result <EOL> self . run_entry [ '<STR_LIT:status>' ] = '<STR_LIT>' <EOL> self . final_save ( attempts = <NUM_LIT:10> ) <EOL> def interrupted_event ( self , interrupt_time ) : <EOL> self . run_entry [ '<STR_LIT>' ] = interrupt_time <EOL> self . run_entry [ '<STR_LIT:status>' ] = '<STR_LIT>' <EOL> self . final_save ( attempts = <NUM_LIT:3> ) <EOL> def failed_event ( self , fail_time , fail_trace ) : <EOL> self . run_entry [ '<STR_LIT>' ] = fail_time <EOL> self . run_entry [ '<STR_LIT:status>' ] = '<STR_LIT>' <EOL> self . run_entry [ '<STR_LIT>' ] = fail_trace <EOL> self . final_save ( attempts = <NUM_LIT:1> ) <EOL> def resource_event ( self , filename ) : <EOL> if self . fs . exists ( filename = filename ) : <EOL> md5hash = get_digest ( filename ) <EOL> if self . fs . exists ( filename = filename , md5 = md5hash ) : <EOL> resource = ( filename , md5hash ) <EOL> if resource not in self . run_entry [ '<STR_LIT>' ] : <EOL> self . run_entry [ '<STR_LIT>' ] . append ( resource ) <EOL> self . save ( ) <EOL> return <EOL> with open ( filename , '<STR_LIT:rb>' ) as f : <EOL> file_id = self . fs . put ( f , filename = filename ) <EOL> md5hash = self . fs . get ( file_id ) . md5 <EOL> self . run_entry [ '<STR_LIT>' ] . append ( ( filename , md5hash ) ) <EOL> self . save ( ) <EOL> def artifact_event ( self , filename ) : <EOL> with open ( filename , '<STR_LIT:rb>' ) as f : <EOL> _ , tail = os . path . split ( filename ) <EOL> run_id = self . run_entry [ '<STR_LIT>' ] <EOL> db_filename = '<STR_LIT>' . format ( <EOL> self . run_entry [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , run_id , tail ) <EOL> file_id = self . fs . put ( f , filename = db_filename ) <EOL> self . run_entry [ '<STR_LIT>' ] . append ( file_id ) <EOL> self . save ( ) <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , MongoObserver ) : <EOL> return self . runs == other . runs <EOL> return False <EOL> def __ne__ ( self , other ) : <EOL> return not self . __eq__ ( other ) <EOL> class MongoDbOption ( CommandLineOption ) : <EOL> """<STR_LIT>""" <EOL> arg = '<STR_LIT>' <EOL> arg_description = "<STR_LIT>" "<STR_LIT>" <EOL> DB_NAME_PATTERN = r"<STR_LIT>" <EOL> HOSTNAME_PATTERN = r"<STR_LIT>" r"<STR_LIT>" r"<STR_LIT>" r"<STR_LIT>" <EOL> URL_PATTERN = "<STR_LIT>" + HOSTNAME_PATTERN + "<STR_LIT:)>" + "<STR_LIT::>" + "<STR_LIT>" <EOL> DB_NAME = re . compile ( "<STR_LIT>" + DB_NAME_PATTERN + "<STR_LIT:$>" ) <EOL> URL = re . compile ( "<STR_LIT>" + URL_PATTERN + "<STR_LIT:$>" ) <EOL> URL_DB_NAME = re . compile ( "<STR_LIT>" + URL_PATTERN + "<STR_LIT:)>" + "<STR_LIT::>" + <EOL> "<STR_LIT>" + DB_NAME_PATTERN + "<STR_LIT>" ) <EOL> @ classmethod <EOL> def apply ( cls , args , run ) : <EOL> url , db_name , prefix = cls . parse_mongo_db_arg ( args ) <EOL> if prefix : <EOL> mongo = MongoObserver . create ( db_name = db_name , url = url , <EOL> prefix = prefix ) <EOL> else : <EOL> mongo = MongoObserver . create ( db_name = db_name , url = url ) <EOL> run . observers . append ( mongo ) <EOL> @ classmethod <EOL> def parse_mongo_db_arg ( cls , mongo_db ) : <EOL> if cls . DB_NAME . match ( mongo_db ) : <EOL> db_name , _ , prefix = mongo_db . partition ( '<STR_LIT:.>' ) <EOL> return '<STR_LIT>' , db_name , prefix <EOL> elif cls . URL . match ( mongo_db ) : <EOL> return mongo_db , '<STR_LIT>' , '<STR_LIT>' <EOL> elif cls . URL_DB_NAME . match ( mongo_db ) : <EOL> match = cls . URL_DB_NAME . match ( mongo_db ) <EOL> db_name , _ , prefix = match . group ( '<STR_LIT>' ) . partition ( '<STR_LIT:.>' ) <EOL> return match . group ( '<STR_LIT:url>' ) , db_name , prefix <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( mongo_db ) ) </s>
