import os, 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.item import Item
from models.target import Target
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...')
solution = Solution(response_id=random.randint(100, 5000), forms=[]) # unsolved solution
# setup common Solver variables
items = LpVariable.dicts("Item", [item.id for item in self.solver_run.items], lowBound=0, upBound=1, cat='Binary')
bundles = LpVariable.dicts("Bundle", [bundle.id for bundle in self.solver_run.bundles], lowBound=0, upBound=1, cat='Binary')
# 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
logging.info(f'Generating Solution for Form {form_number}')
while current_drift <= Target.max_drift():
drift_percent = current_drift / 100
# create problem
problem = LpProblem('ata-form-generate', LpMinimize)
# objective function
problem += lpSum(
[
bundle.count * bundles[bundle.id] for bundle in self.solver_run.bundles
] +
[
items[item.id] for item in self.solver_run.unbundled_items()
]
)
# Form Constraints
problem += lpSum(
[
bundle.count * bundles[bundle.id] for bundle in self.solver_run.bundles
] +
[
1 * items[item.id] for item in self.solver_run.unbundled_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)
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 self.solver_run.bundles
] +
[
item.iif(self.solver_run, tif_target.theta) * items[item.id]
for item in self.solver_run.unbundled_items()
]
) >= tif_target.minimum(drift_percent), 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 self.solver_run.bundles
] +
[
item.iif(self.solver_run, tif_target.theta) * items[item.id]
for item in self.solver_run.unbundled_items()
]
) <= tif_target.maximum(drift_percent), 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 self.solver_run.bundles
] +
[
item.irf(self.solver_run, tcc_target.theta) * items[item.id]
for item in self.solver_run.unbundled_items()
]
) >= tcc_target.minimum(drift_percent), 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 self.solver_run.bundles
] +
[
item.irf(self.solver_run, tcc_target.theta) * items[item.id]
for item in self.solver_run.unbundled_items()
]
) <= tcc_target.maximum(drift_percent), 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
for v in problem.variables(): print(v.name, "=", v.varValue);
logging.info(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:
for v in problem.variables(): print(v.name, "=", v.varValue);
logging.info(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, solution):
# add return items and create as a form
form_items = service_helper.solution_items(problem.variables(), self.solver_run)
form = Form.create(form_items, self.solver_run, LpStatus[problem.status])
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)