Spaces:
Sleeping
Sleeping
from abc import ABC, abstractmethod | |
import json | |
import os | |
class BaseTask(ABC): | |
def __init__(self, config_path, input_structure_path, output_structure_path): | |
self.config = self.load_config(config_path) | |
self.input_structure = self.load_json(input_structure_path) | |
self.output_structure = self.load_json(output_structure_path) | |
def load_json(self, file_path): | |
"""Load a JSON file.""" | |
if not os.path.exists(file_path): | |
raise FileNotFoundError(f"{file_path} not found.") | |
with open(file_path, 'r') as file: | |
return json.load(file) | |
def load_config(self, config_path): | |
"""Load task configuration from the config file.""" | |
if not os.path.exists(config_path): | |
raise FileNotFoundError(f"{config_path} not found.") | |
# Implement logic to load and parse the config file | |
with open(config_path, 'r') as file: | |
return json.load(file) | |
def validate_input(self, input_data): | |
"""Validate the input data against the input structure.""" | |
# Implement validation logic comparing input_data with self.input_structure | |
for key, value_type in self.input_structure.items(): | |
if key not in input_data or not isinstance(input_data[key], value_type): | |
raise ValueError(f"Invalid input for {key}: Expected {value_type}, got {type(input_data.get(key))}") | |
def validate_output(self, output_data): | |
"""Validate the output data against the output structure.""" | |
for key, value_type in self.output_structure.items(): | |
if key not in output_data or not isinstance(output_data[key], value_type): | |
raise ValueError(f"Invalid output for {key}: Expected {value_type}, got {type(output_data.get(key))}") | |
def load_input(self, input_data): | |
"""Load and validate input data.""" | |
self.validate_input(input_data) | |
pass | |
def process(self): | |
"""Process the input and perform the task's specific action.""" | |
pass | |
def save_output(self, result): | |
"""Format, validate, and return the output data.""" | |
self.validate_output(result) | |
pass | |
def execute(self, input_data): | |
"""The main method to run the task end-to-end.""" | |
self.load_input(input_data) | |
result = self.process() | |
return self.save_output(result) | |