#!/bin/env python #----------------------------------------------------------------------------- # simulation.py #----------------------------------------------------------------------------- # # Copyright (c) 2009 Stanford University and the Authors. # Authors: Christopher Bruns # Contributors: # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), # to deal in the Software without restriction, including without limitation # the rights to use, copy, modify, merge, publish, distribute, sublicense, # and/or sell copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL # THE AUTHORS, CONTRIBUTORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, # DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR # OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE # USE OR OTHER DEALINGS IN THE SOFTWARE. # ---------------------------------------------------------------------------- from datetime import datetime import time import sys import os import zephyr.zephyrrc from zephyr.observable import Observable, ObservableValue class Simulation(object): def __init__(self): self.zephyr_lib_dir = os.path.abspath(os.path.dirname(sys.argv[0])) self.bin_dir = os.path.join(self.zephyr_lib_dir, "bin", "win32") self.test_data_dir = os.path.join(self.zephyr_lib_dir, "testData") # Progress and log observables for GUI self.progress_status = Simulation.ProgressStatus() self.log_messages = Simulation.LogMessages() # initialize rc param keys before loading .zephyrrc file self.rc_params = zephyr.zephyrrc.ZephyrRC() self.pdb_file = None self.simulation_path = None self.short_status = ObservableValue("(Idle)") self.status_string = ObservableValue("Waiting for you to press 'Simulate'") ProgressToStatus(self); # Link progress bar to short status def create_mol_dir(self): """ Create simulation subdirectory name for current molecule. """ self.file_root = os.path.splitext( os.path.basename(self.get_pdb_file()) )[0] self.mol_dir = os.path.join(self.get_simulation_path(), self.file_root) # Leave it to the GUI to create the directory... # Remember to load .zephyrrc params *after* GUI is created, # so data widgets will be updated properly def load_rc_params(self): self.rc_params.load_params() # If pdb_file is not set, load a suitable default default_pdb = os.path.join(self.test_data_dir, "alanylalanine_capped.pdb") if None == self.pdb_file: self.set_pdb_file(default_pdb) elif not os.path.exists(self.pdb_file): self.set_pdb_file(default_pdb) # Fallback default for simulation directory self.set_simulation_path(self.rc_params.get_simulation_path()) def log(self, message): self.log_messages.add_message(message) def clean_up(self): self.rc_params.write_params() def set_rc_param(self, param_name, value): if param_name not in self.rc_params: self.rc_params[param_name] = ObservableValue(value) else: self.rc_params[param_name].set(value) def get_rc_param(self, param_name): return self.rc_params[param_name] def complete_progress(self): p = self.progress_status p.set_sim_time(p.total_sim_time) # Directory where simulation files will be created def set_simulation_path(self, path): self.set_rc_param('simulationPath', path) def get_simulation_path(self): return self.get_rc_param('simulationPath').get() simulation_path = property(get_simulation_path, set_simulation_path) # Name of input PDB file def set_pdb_file(self, file): self.set_rc_param('pdbFile', str(file)) def get_pdb_file(self): return self.get_rc_param('pdbFile').get() pdb_file = property(get_pdb_file, set_pdb_file) class ProgressStatus(Observable): def __init__(self): Observable.__init__(self) self.start( total_sim_time=None ) def start(self, total_sim_time, start_sim_time = 0.0, modifier=None): self.start_lab_time = datetime.now() self.start_sim_time = start_sim_time self.percent_complete = 0.0 self.total_sim_time = total_sim_time self.elapsed_lab_time = 0.0 self.elapsed_sim_time = 0.0 self.remaining_lab_time = None; # unknowable now self.notify(modifier) def set_sim_time(self, sim_time, modifier=None): self.elapsed_sim_time = sim_time now = datetime.now() self.elapsed_lab_time = now - self.start_lab_time dt = sim_time - self.start_sim_time if dt > 0: p = self.percent_complete = dt/self.total_sim_time e = self.elapsed_lab_time s = e.days*86400.0 + e.seconds + e.microseconds/1e6 self.remaining_lab_time = (1.0/p - 1.0) * s self.notify(modifier) class LogMessages(Observable): def add_message(self, message, modifier=None): self.message = message self.notify(modifier) # Toy Simulation for testing class TestSimulation(Simulation): def run_simulation(self): self.total_sim_time = 20 self.progress_status.start(self.total_sim_time) self.exiting = False for t in range(self.total_sim_time + 1): time.sleep(0.3) if self.exiting: break self.progress_status.set_sim_time(float(t)) self.log_messages.add_message("Progress %f\n" % t) self.exiting = True def kill(self): self.exiting = True def is_running(self): return not self.exiting # Instead of a whole GUI, write progress numbers to terminal class PrintProgressObserver(object): def __init__(self, simulation): simulation.progress_status.attach(self) simulation.run_simulation() def update(self, progress): print "Progress %f" % progress.elapsed_sim_time if progress.remaining_lab_time: print " %s remaining" % progress.remaining_lab_time else: print " (calculating time remaining...)" sys.stdout.flush() class ProgressToStatus(object): "Update short status when progress is updated" def __init__(self, simulation): self.simulation = simulation simulation.progress_status.attach(self) def update(self, progress): self.simulation.short_status.set( "%d%%" % int(progress.percent_complete * 100.0) ) if __name__ == "__main__": PrintProgressObserver( TestSimulation() )