import os, json, random, io, logging from pulp import LpProblem, LpVariable, LpMinimize, LpStatus, lpSum 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.item import Item from models.bundle import Bundle from services.base import Base class LoftService(Base): 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 for item in service_helper.items_csv_to_dict(items_csv_reader, solver_run): solver_run.items.append(Item.parse_obj(item)) logging.info('Processed Attributes...') return solver_run def generate_solution(self) -> Solution: logging.info('Generating Solution...') # counter for number of forms f = 0 # setup vars items = LpVariable.dicts( "Item", [item.id for item in self.solver_run.items], lowBound=1, upBound=1, cat='Binary') # check if problem request has bundles bundle_constraint = self.solver_run.get_constraint_by_type('bundle') # iterate for number of forms that require creation # currently creates distinc forms with no item overlap while f < self.solver_run.total_forms: # unsolved solution solution = Solution( response_id=random.randint(100, 5000), forms=[] ) # initiate problem problem = None if bundle_constraint: bundles_amount = random.randint(int(bundle_constraint.minimum), int(bundle_constraint.maximum)) problem = self.recursive_solve(items, bundles_amount) else: # no bundles problem = self.solve(items) # successfull form, increment and exit out of loop f += 1 # add return items and create as a form form_items = service_helper.solution_items(problem.variables(), self.solver_run) # add form to solution solution.forms.append(Form.create(form_items, self.solver_run, LpStatus[problem.status])) return solution def recursive_solve(self, items, bundles_amount, attempts = 1000) -> LpProblem: selected_bundles = solver_helper.get_random_bundles( self.solver_run.total_form_items, bundles_amount, self.solver_run.bundles) problem = self.solve(items, selected_bundles) # if optimal solution found, end recursion if LpStatus[problem.status] == 'Optimal' or attempts == 0: return problem else: return self.recursive_solve(items, max, attempts - 1) def solve(self, items: list[Item], bundles: list[Bundle] or None = None) -> LpProblem: # create problem problem = LpProblem("ata-form-generate", LpMinimize) # dummy objective function, because it just makes things easier™ problem += lpSum([items[item.id] for item in self.solver_run.items]) # constraints problem += lpSum([items[item.id] for item in self.solver_run.items]) == self.solver_run.total_form_items, 'Total form items' # dynamic constraints problem = solver_helper.build_constraints(self.solver_run, problem, items, bundles) # multi-objective constraints for target in self.solver_run.objective_function.tif_targets: problem += lpSum([item.iif(self.solver_run, target.theta)*items[item.id] for item in self.solver_run.items]) >= target.value - 8, f'max tif theta ({target.theta}) target value {target.value}' problem += lpSum([item.iif(self.solver_run, target.theta)*items[item.id] for item in self.solver_run.items]) <= target.value + 8, f'min tif theta ({target.theta}) target value {target.value}' for target in self.solver_run.objective_function.tcc_targets: problem += lpSum([item.irf(self.solver_run, target.theta)*items[item.id] for item in self.solver_run.items]) >= target.value - 20, f'max tcc theta ({target.theta}) target value {target.value}' problem += lpSum([item.irf(self.solver_run, target.theta)*items[item.id] for item in self.solver_run.items]) <= target.value + 20, f'min tcc theta ({target.theta}) target value {target.value}' # solve problem problem.solve() return problem 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, os.environ['S3_PROCESSED_BUCKET']) solution_file = service_helper.error_to_file(buffer, error) else: logging.info('Streaming %s to s3 bucket - %s', self.file_name, os.environ['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, os.environ['S3_PROCESSED_BUCKET'])