import json, random, io, logging from pulp import LpProblem, LpVariable, LpMinimize, LpMaximize, LpAffineExpression, LpConstraint, LpStatus, lpSum from lib.application_configs import ApplicationConfigs from helpers import aws_helper, tar_helper, csv_helper, service_helper, solver_helper from lib.errors.item_generation_error import ItemGenerationError from models.solver_run import SolverRun from models.solution import Solution from models.form import Form from models.target import Target from services.base import Base class FormGenerationService(Base): ACTION = 'formGeneration' def process(self): try: self.solver_run = self.create_solver_run_from_attributes() self.solver_run.generate_bundles() self.solution = self.generate_solution() self.result = self.stream_to_s3_bucket() except ItemGenerationError as error: self.result = self.stream_to_s3_bucket(error) except TypeError as error: logging.error(error) self.result = self.stream_to_s3_bucket( ItemGenerationError( "Provided params causing error in calculation results")) def create_solver_run_from_attributes(self) -> SolverRun: logging.info('Retrieving attributes from message...') # get s3 object self.key = aws_helper.get_key_from_message(self.source) s3_object = aws_helper.get_object( self.key, aws_helper.get_bucket_from_message(self.source)) # convert to tar self.tar = tar_helper.raw_to_tar(s3_object) # get attributes file and convert to dict attributes = json.loads( tar_helper.extract_file_from_tar( self.tar, 'solver_run_attributes.json').read()) # create solver run solver_run = SolverRun.parse_obj(attributes) # get items file and convert to dict items_csv = tar_helper.extract_file_from_tar(self.tar, 'items.csv') items_csv_reader = csv_helper.file_stream_reader(items_csv) # add items to solver run solver_run.items = service_helper.csv_to_item(items_csv_reader, solver_run) logging.info('Processed Attributes...') return solver_run def generate_solution(self) -> Solution: logging.info('Generating Solution...') solution = Solution(response_id=random.randint(100, 5000), forms=[]) # unsolved solution # iterate for number of forms that require creation for form_count in range(self.solver_run.total_forms): form_number = form_count + 1 current_drift = 0 # FF Tokyo Drift # adding an element of randomness to the items and bundles used # may need to change impl based on limit of items available selected_items = self.solver_run.select_items_by_percent(30) selected_bundles = self.solver_run.select_bundles_by_percent( 30) # setup common Solver variables items = LpVariable.dicts("Item", [item.id for item in selected_items], lowBound=0, upBound=1, cat='Binary') bundles = LpVariable.dicts("Bundle", [bundle.id for bundle in selected_bundles], lowBound=0, upBound=1, cat='Binary') logging.info(f'Generating Solution for Form {form_number} using the {self.solver_run.irt_model.model} IRT model') while current_drift <= Target.max_drift(): drift_percent = current_drift / 100 self.solver_run.objective_function.update_targets_drift( drift_percent) # create problem problem = LpProblem('ata-form-generate', LpMinimize) # objective function problem += lpSum([ bundle.count * bundles[bundle.id] for bundle in selected_bundles ] + [ items[item.id] for item in selected_items ]) # Form Constraints problem += lpSum( [ bundle.count * bundles[bundle.id] for bundle in selected_bundles ] + [ 1 * items[item.id] for item in selected_items ] ) == self.solver_run.total_form_items, f'Total bundle form items for form {form_number}' # Dynamic constraints.. currently we only support Metadata and Bundles(Cases/Passages) problem = solver_helper.build_constraints( self.solver_run, problem, items, bundles, selected_items, selected_bundles) # form uniqueness constraints # for form in solution.forms: # form_item_options = [ # bundles[bundle.id] # for bundle in selected_bundles # ] + [ # items[item.id] # for item in selected_items # ] # problem += len( # set(form.solver_variables) # & set(form_item_options)) / float( # len( # set(form.solver_variables) # | set(form_item_options))) * 100 >= 10 logging.info('Creating TIF and TCC Elastic constraints') # Behold our very own Elastic constraints! for tif_target in self.solver_run.objective_function.tif_targets: problem += lpSum([ bundle.tif(self.solver_run.irt_model, tif_target.theta) * bundles[bundle.id] for bundle in selected_bundles ] + [ item.iif(self.solver_run, tif_target.theta) * items[item.id] for item in selected_items ]) >= tif_target.minimum( ), f'Min TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%' problem += lpSum([ bundle.tif(self.solver_run.irt_model, tif_target.theta) * bundles[bundle.id] for bundle in selected_bundles ] + [ item.iif(self.solver_run, tif_target.theta) * items[item.id] for item in selected_items ]) <= tif_target.maximum( ), f'Max TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%' for tcc_target in self.solver_run.objective_function.tcc_targets: problem += lpSum([ bundle.trf(self.solver_run.irt_model, tcc_target.theta) * bundles[bundle.id] for bundle in selected_bundles ] + [ item.irf(self.solver_run, tcc_target.theta) * items[item.id] for item in selected_items ]) >= tcc_target.minimum( ), f'Min TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%' problem += lpSum([ bundle.trf(self.solver_run.irt_model, tcc_target.theta) * bundles[bundle.id] for bundle in selected_bundles ] + [ item.irf(self.solver_run, tcc_target.theta) * items[item.id] for item in selected_items ]) <= tcc_target.maximum( ), f'Max TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%' logging.info( f'Solving for Form {form_number} with a drift of {current_drift}%' ) problem.solve() if LpStatus[problem.status] == 'Infeasible': logging.info( f'attempt infeasible for drift of {current_drift}%') if current_drift >= Target.max_drift( ): # this is the last attempt, so lets finalize the solution if ApplicationConfigs.local_dev_env: service_helper.print_problem_variables(problem) logging.info( f'No feasible solution found for Form {form_number}!' ) self.add_form_to_solution(problem, solution) break current_drift += Target.max_drift_increment() else: if ApplicationConfigs.local_dev_env: service_helper.print_problem_variables(problem) logging.info( f'Optimal solution found with drift of {current_drift}%!' ) self.add_form_to_solution(problem, solution) break logging.info('Solution Generated.') return solution def add_form_to_solution(self, problem: LpProblem, solution: Solution): # add return items and create as a form form_items, solver_variables = service_helper.solution_items( problem.variables(), self.solver_run) form = Form.create(form_items, self.solver_run, LpStatus[problem.status], solver_variables) solution.forms.append(form) logging.info('Form generated and added to solution...') def stream_to_s3_bucket(self, error=None): self.file_name = f'{service_helper.key_to_uuid(self.key)}.csv' solution_file = None # setup writer buffer and write processed forms to file buffer = io.StringIO() if error: logging.info('Streaming %s error response to s3 bucket - %s', self.file_name, ApplicationConfigs.s3_processed_bucket) solution_file = service_helper.error_to_file(buffer, error) else: logging.info('Streaming %s to s3 bucket - %s', self.file_name, ApplicationConfigs.s3_processed_bucket) solution_file = service_helper.solution_to_file( buffer, self.solver_run.total_form_items, self.solution.forms) # upload generated file to s3 and return result return aws_helper.file_stream_upload( solution_file, self.file_name, ApplicationConfigs.s3_processed_bucket, self.ACTION)