jshbrmn/irt-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into feature/QUANT-3102

Browse Source
This commit is contained in:
brmnjsh 2023-11-29 22:40:24 +00:00
commit d58954ec1d
6 changed files with 65 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/main.py
View File

@ -43,7 +43,7 @@ class ServiceListener(Consumer):
logging.error(f'action of type {action} does not exist.')
def main():
logging.info('Starting IRT Service: The Enemies Within (v1.7.0)...')
logging.info('Starting IRT Service: Tokyo Drift 2: Driftocolypse (v1.8.0)...')
# ToDo: Figure out a much better way of doing this.
# LocalStack wants 'endpoint_url', while prod doesnt :(

3
app/models/item.py
View File

@ -1,6 +1,5 @@
import logging
from pydantic import BaseModel, validator
from typing import List, Optional, Tuple
from typing import List, Optional
from models.attribute import Attribute

56
app/models/objective_function.py
View File

@ -1,13 +1,17 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Literal, Union
from pydantic import BaseModel
from typing import Dict, List, AnyStr
from pulp import lpSum
from pydantic import BaseModel, validator
from typing import Dict, List
from pulp import lpSum, LpMinimize, LpMaximize
from models.targets.tif_target import TifTarget
from models.targets.tcc_target import TccTarget
from models.problem import Problem
if TYPE_CHECKING:
from models.solver_run import SolverRun
class ObjectiveFunction(BaseModel):
# minimizing tif/tcc target value is only option currently
# as we add more we can build this out to be more dynamic
@ -15,17 +19,45 @@ class ObjectiveFunction(BaseModel):
tif_targets: List[TifTarget]
tcc_targets: List[TccTarget]
target_variance_percentage: int = 10
objective: AnyStr = "minimize"
objective: Literal[1,-1] = 1
functions: List[Literal['tcc', 'tif']] = ['tcc', 'tif']
weight: Dict = {'tif': 1, 'tcc': 1}
def for_problem(self, problem_handler: Problem) -> None:
problem_handler.problem += lpSum([
bundle.count * problem_handler.solver_bundles_var[bundle.id]
for bundle in problem_handler.bundles
] + [
problem_handler.solver_items_var[item.id]
for item in problem_handler.items
])
@validator("objective", pre=True)
def set_objective(cls, v) -> List[int]:
if v == 'minimize':
return 1
elif v == 'maximize':
return -1
else:
return None
def for_problem(self, problem_handler: Problem, solver_run: SolverRun) -> List[lpSum]:
functions = []
for function in self.functions:
if function == 'tcc':
functions.append(lpSum([
bundle.trf(solver_run.irt_model, solver_run.theta_cut_score)
* problem_handler.solver_bundles_var[bundle.id]
for bundle in problem_handler.bundles
] + [
item.irf(solver_run.irt_model, solver_run.theta_cut_score) *
problem_handler.solver_items_var[item.id]
for item in problem_handler.items
]))
elif function == 'tif':
problem_handler.problem += lpSum([
bundle.tif(solver_run.irt_model, solver_run.theta_cut_score)
* problem_handler.solver_bundles_var[bundle.id]
for bundle in problem_handler.bundles
] + [
item.iif(solver_run.irt_model, solver_run.theta_cut_score) *
problem_handler.solver_items_var[item.id]
for item in problem_handler.items
])
return functions
def increment_targets_drift(self,
limit: float or bool,

10
app/models/problem.py
View File

@ -43,7 +43,12 @@ class Problem(BaseModel):
def solve(self, solver_run: SolverRun, enemy_ids: List[int] = []) -> LpProblem:
logging.info('solving problem...')
self.problem.solve()
# creating problem multi-objective functions
objective_functions = solver_run.objective_function.for_problem(self, solver_run)
self.problem.sequentialSolve(objective_functions)
return self.problem
# NOTICE: Legacy enemies implementation
# leaving this in, just in case the current impl fails to function
@ -106,9 +111,6 @@ class Problem(BaseModel):
def generate(self, solution: Solution, solver_run: SolverRun) -> None:
try:
# creating problem objective function
solver_run.objective_function.for_problem(self)
logging.info('Creating Constraints...')
# generic constraints
for constraint in solver_run.constraints:

6
app/models/solver_run.py
View File

@ -18,10 +18,6 @@ from models.bundle import Bundle
from models.objective_function import ObjectiveFunction
from models.advanced_options import AdvancedOptions
if TYPE_CHECKING:
from models.solution import Solution
from models.problem import Problem
ConstraintType = TypeVar('ConstraintType', bound=GenericConstraint)
class SolverRun(BaseModel):
@ -36,6 +32,8 @@ class SolverRun(BaseModel):
theta_cut_score: float = 0.00
drift_style: Literal['constant', 'variable'] = 'constant'
allow_enemies: bool = False
max_attempts: int
max_drift: int = 10
advanced_options: Optional[AdvancedOptions]
engine: str

16
app/services/form_generation_service.py
View File

@ -1,6 +1,6 @@
import json, random, io, logging
from pulp import LpProblem, LpMinimize, LpStatus
from pulp import LpProblem, LpStatus
from lib.application_configs import ApplicationConfigs
from helpers import aws_helper, tar_helper, csv_helper, service_helper
@ -71,17 +71,22 @@ class FormGenerationService(Base):
# iterate for number of forms that require creation
for form_count in range(self.solver_run.total_forms):
form_number = form_count + 1
drift_increment = self.solver_run.max_drift / (self.solver_run.max_attempts - 1)
current_drift = 0 # FF Tokyo Drift
current_attempt = 0
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():
# respect max attempts
# this will likely be more built out when we add increment rate & drif limit
while current_attempt <= self.solver_run.max_attempts:
drift_percent = current_drift / 100
self.solver_run.objective_function.update_targets_drift(
drift_percent)
# create problem
problem_handler = Problem(items = self.solver_run.unbundled_items(), bundles = self.solver_run.bundles, problem = LpProblem('ata-form-generate', LpMinimize))
problem = LpProblem('ata-form-generate', self.solver_run.objective_function.objective)
problem_handler = Problem(items = self.solver_run.unbundled_items(), bundles = self.solver_run.bundles, problem = problem)
problem_handler.generate(solution, self.solver_run)
problem = problem_handler.solve(self.solver_run)
@ -90,7 +95,7 @@ class FormGenerationService(Base):
f'attempt infeasible for drift of {current_drift}%')
if current_drift >= Target.max_drift(
): # this is the last attempt, so lets finalize the solution
) or current_attempt >= self.solver_run.max_attempts: # this is the last attempt, so lets finalize the solution
if ApplicationConfigs.local_dev_env:
service_helper.print_problem_variables(problem)
@ -103,7 +108,8 @@ class FormGenerationService(Base):
break
current_drift += Target.max_drift_increment()
current_drift += drift_increment
current_attempt += 1
else:
if ApplicationConfigs.local_dev_env:
service_helper.print_problem_variables(problem)