IIabm/computation.py

53 lines
1.5 KiB
Python

import os
import datetime
from model import Model
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from controller_db import ControllerDB
class Computation:
def __init__(self, c_db: 'ControllerDB'):
self.c_db = c_db
self.pid = os.getpid()
def run(self, str_code='0', s_id=None):
sample_random = self.c_db.fetch_a_sample(s_id)
if sample_random is None:
return True
# lock this row by update is_done_flag to 0
self.c_db.lock_the_sample(sample_random)
print(f"Pid {self.pid} ({str_code}) is running sample {sample_random.id} at {datetime.datetime.now()}")
dct_exp = {column: getattr(sample_random.experiment, column)
for column in sample_random.experiment.__table__.c.keys()}
del dct_exp['id']
dct_sample_para = {'sample': sample_random,
'seed': sample_random.seed,
**dct_exp}
model = Model(dct_sample_para)
results = model.run(display=False)
return False
if __name__ == '__main__':
str_exp = 'test'
from controller_db import ControllerDB
controller_db = ControllerDB(str_exp)
controller_db.reset_db()
# print(controller_db.dct_parameter)
exp = Computation(controller_db)
is_all_done = exp.run('999', 87)
# while 1:
# # time.sleep(random.uniform(0, 10))
# is_all_done = exp.run(str_exp)
# if is_all_done:
# break