logger.setLevel(settings["worker"]["loglevel"])
def class_from_args(class_name, arg_dict: dict):
fieldSet = {f.name for f in fields(class_name) if f.init}
filteredArgDict = {k: v for k, v in arg_dict.items() if k in fieldSet}
return class_name(**filteredArgDict)
@dataclass(frozen=True, init=True)
class TaskJob:
settings: SettingsManager
project: ProjectMetadata
source: SourceMetadata
output_file_path: str
output_file_name: str
output_directory: str
input_level: str
def __post_init__(self):
# TODO: Custom exceptions for task job validation
assert os.path.exists(self.source.file_path) # SOURCE ACCESSIBLE
assert os.path.exists(self.output_directory) # CLEAR EXPORT PATH
assert not os.path.exists(self.output_file_path) # NO OVERWRITE
assert self.input_level in ["in_range=full", "in_range=video"]
def ensure_logs_output_path(job: TaskJob):
"""
Ensure log folder is writable, get path.
Custom exceptions for task job validation
assert os.path.exists(self.output_directory) # CLEAR EXPORT PATH
assert not os.path.exists(self.output_file_path) # NO OVERWRITE
https://github.com/in03/proxima/blob/9b1569618a33712f15cc307d5d2650fbe96f5ab5/proxima/worker/tasks.py#L45
b1031cb58d36c4c23e081f0303ff0614567946e4