Spaces:
Paused
Paused
File size: 2,626 Bytes
1c72248 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
import importlib
from collections import OrderedDict
from typing import List
from jobs.process import BaseProcess
class BaseJob:
def __init__(self, config: OrderedDict):
if not config:
raise ValueError('config is required')
self.process: List[BaseProcess]
self.config = config['config']
self.raw_config = config
self.job = config['job']
self.torch_profiler = self.get_conf('torch_profiler', False)
self.name = self.get_conf('name', required=True)
if 'meta' in config:
self.meta = config['meta']
else:
self.meta = OrderedDict()
def get_conf(self, key, default=None, required=False):
if key in self.config:
return self.config[key]
elif required:
raise ValueError(f'config file error. Missing "config.{key}" key')
else:
return default
def run(self):
print("")
print(f"#############################################")
print(f"# Running job: {self.name}")
print(f"#############################################")
print("")
# implement in child class
# be sure to call super().run() first
pass
def load_processes(self, process_dict: dict):
# only call if you have processes in this job type
if 'process' not in self.config:
raise ValueError('config file is invalid. Missing "config.process" key')
if len(self.config['process']) == 0:
raise ValueError('config file is invalid. "config.process" must be a list of processes')
module = importlib.import_module('jobs.process')
# add the processes
self.process = []
for i, process in enumerate(self.config['process']):
if 'type' not in process:
raise ValueError(f'config file is invalid. Missing "config.process[{i}].type" key')
# check if dict key is process type
if process['type'] in process_dict:
if isinstance(process_dict[process['type']], str):
ProcessClass = getattr(module, process_dict[process['type']])
else:
# it is the class
ProcessClass = process_dict[process['type']]
self.process.append(ProcessClass(i, self, process))
else:
raise ValueError(f'config file is invalid. Unknown process type: {process["type"]}')
def cleanup(self):
# if you implement this in child clas,
# be sure to call super().cleanup() LAST
del self
|