working TCC (Test response funciton)
This commit is contained in:
parent
6b1bbda169
commit
258915b08f
@ -40,7 +40,7 @@ def solution_to_file(buffer, total_form_items, forms):
|
||||
# add each form as row to processed csv
|
||||
for form in forms:
|
||||
# provide generated items and cut score
|
||||
row = form.items + [form.cut_score]
|
||||
row = [item.id for item in form.items] + [form.cut_score]
|
||||
wr.writerow(row)
|
||||
|
||||
buff2 = io.BytesIO(buffer.getvalue().encode())
|
||||
|
@ -1,4 +1,4 @@
|
||||
from models.three_parameter_logitistc import ThreeParameterLogistic
|
||||
from lib.irt.models.three_parameter_logistic import ThreeParameterLogistic
|
||||
|
||||
class ItemResponseFunction():
|
||||
def __init__(self, irt_model):
|
||||
@ -6,7 +6,7 @@ class ItemResponseFunction():
|
||||
|
||||
def calculate(self, **kwargs):
|
||||
if self.model_data.model == '3PL':
|
||||
return ThreeParameterLogistic.new(self.model_data, kwargs).result
|
||||
return ThreeParameterLogistic(self.model_data, kwargs).result()
|
||||
else:
|
||||
# potentially error out
|
||||
return None
|
||||
|
@ -7,7 +7,6 @@ class ThreeParameterLogistic:
|
||||
self.theta = kwargs['theta']
|
||||
|
||||
def result(self):
|
||||
a = self.model_params.a
|
||||
c = self.model_params.c
|
||||
|
||||
return c + (1 - c) * (1 / (1 + e**(-a * (self.theta - self.b_param))))
|
||||
a = self.model_params.a_param
|
||||
c = self.model_params.c_param
|
||||
return c + (1 - c) * (1 / (1 + self.e**(-a * (self.theta - self.b_param))))
|
||||
|
@ -1,12 +1,17 @@
|
||||
from lib.irt.item_response_function import ItemResponseFunction
|
||||
|
||||
# otherwise known as the Test Characteristic Curve (TCC)
|
||||
class TestResponseFunction():
|
||||
def __init__(self, irf):
|
||||
self.irf = irf
|
||||
def __init__(self, irt_model):
|
||||
self.irt_model = irt_model
|
||||
self.irf = ItemResponseFunction(irt_model)
|
||||
|
||||
def calculate(self, items, **kwargs):
|
||||
result = 0
|
||||
sum = 0
|
||||
|
||||
for item in items:
|
||||
result += irf.calculate(b_param=item.b_param, theta=kwargs['theta'])
|
||||
result = self.irf.calculate(b_param=item.b_param, theta=kwargs['theta'])
|
||||
item.irf = result
|
||||
sum += item.irf
|
||||
|
||||
return result
|
||||
return sum
|
||||
|
@ -4,5 +4,5 @@ from typing import List
|
||||
from models.item import Item
|
||||
|
||||
class Form(BaseModel):
|
||||
items: List[int]
|
||||
items: List[Item]
|
||||
cut_score: float
|
||||
|
@ -7,3 +7,4 @@ class Item(BaseModel):
|
||||
id: int
|
||||
attributes: List[Attribute]
|
||||
b_param: int
|
||||
irf: float = 0.00
|
||||
|
@ -45,12 +45,7 @@ class LoftService(Base):
|
||||
# real solver will return N forms and process a cut score, this is for mock purposes
|
||||
return Solution(
|
||||
response_id=random.randint(100,5000),
|
||||
forms=[
|
||||
Form(
|
||||
items=[item.id for item in random.sample(self.solver_run.items, self.solver_run.total_form_items)],
|
||||
cut_score=120
|
||||
) for x in range(form_count)
|
||||
]
|
||||
forms=[self.generate_forms(random.sample(self.solver_run.items, self.solver_run.total_form_items)) for x in range(form_count)]
|
||||
)
|
||||
|
||||
def stream_to_s3_bucket(self):
|
||||
@ -60,3 +55,9 @@ class LoftService(Base):
|
||||
|
||||
# upload generated file to s3 and return result
|
||||
return aws_helper.file_stream_upload(solution_file, f'{service_helper.key_to_uuid(self.key)}.csv', os.environ['MEASURE_PROCESSED_BUCKET'])
|
||||
|
||||
def generate_forms(self, items):
|
||||
return Form(
|
||||
items=items,
|
||||
cut_score=TestResponseFunction(self.solver_run.irt_model).calculate(items, theta=self.solver_run.theta_cut_score)
|
||||
)
|
||||
|
Loading…
x
Reference in New Issue
Block a user