diff --git a/optuna_dashboard/_app.py b/optuna_dashboard/_app.py index cbf257a8..5f433072 100644 --- a/optuna_dashboard/_app.py +++ b/optuna_dashboard/_app.py @@ -27,6 +27,7 @@ from ._bottle_util import json_api_view from ._cached_extra_study_property import get_cached_extra_study_property from ._importance import get_param_importance_from_trials_cache from ._pareto_front import get_pareto_front_trials +from ._preferential_history import NewHistory from ._preferential_history import report_history from ._rdb_migration import register_rdb_migration_route from ._serializer import serialize_study_detail @@ -291,11 +292,11 @@ def create_app( report_history( study_id, storage, - { - "mode": mode, - "candidates": candidates, - "clicked": clicked, - }, + NewHistory( + mode=mode, + candidates=candidates, + clicked=clicked, + ), ) response.status = 204 diff --git a/optuna_dashboard/_preferential_history.py b/optuna_dashboard/_preferential_history.py index 00375e56..a246c166 100644 --- a/optuna_dashboard/_preferential_history.py +++ b/optuna_dashboard/_preferential_history.py @@ -16,58 +16,48 @@ _SYSTEM_ATTR_PREFIX_HISTORY = "preference:history" if TYPE_CHECKING: from typing import Literal + from typing import TypedDict FeedbackMode = Literal["ChooseWorst"] - if TYPE_CHECKING: - from typing import TypedDict - - NewHistoryJSON = TypedDict( - "NewHistoryJSON", - { - "mode": FeedbackMode, - "candidates": list[int], - "clicked": int, - }, - ) + ChooseWorstHistory = TypedDict( + "ChooseWorstHistory", + { + "mode": FeedbackMode, + "id": str, + "preference_id": str, + "timestamp": str, + "candidates": list[int], + "clicked": int, + }, + ) + History = ChooseWorstHistory +else: + ChooseWorstHistory = Any + History = Any -@dataclass(frozen=True) -class ChooseWorstHistory: - mode: Literal["ChooseWorst"] - id: str - preference_id: str # making it possible to remove the preference - timestamp: datetime - candidates: list[int] # a list of trial number - clicked: int # The worst trial number in the candidates. - - def to_dict(self) -> dict[str, Any]: - return { - "mode": self.mode, - "id": self.id, - "preference_id": self.preference_id, - "timestamp": self.timestamp.isoformat(), - "candidates": self.candidates, - "clicked": self.clicked, - } - - -History = ChooseWorstHistory +@dataclass +class NewHistory: + mode: FeedbackMode + candidates: list[int] + clicked: int def report_history( study_id: int, storage: BaseStorage, - input_data: NewHistoryJSON, + input_data: NewHistory, ) -> None: preferences = [] - if input_data["mode"] == "ChooseWorst": + # TODO(moririn): Use TypeGuard after adding other history types. + if input_data.mode == "ChooseWorst": preferences = [ - (best, input_data["clicked"]) - for best in input_data["candidates"] - if best != input_data["clicked"] + (best, input_data.clicked) + for best in input_data.candidates + if best != input_data.clicked ] else: - assert False, f"Unknown mode: {input_data['mode']}" + assert False, f"Unknown data: {input_data}" preference_id = report_preferences( study_id=study_id, @@ -76,43 +66,19 @@ def report_history( ) history_id = str(uuid.uuid4()) - if input_data["mode"] == "ChooseWorst": - history = ChooseWorstHistory( - mode="ChooseWorst", - id=history_id, - preference_id=preference_id, - timestamp=datetime.now(), - candidates=input_data["candidates"], - clicked=input_data["clicked"], - ) + if input_data.mode == "ChooseWorst": + history: ChooseWorstHistory = { + "mode": "ChooseWorst", + "id": history_id, + "preference_id": preference_id, + "timestamp": datetime.now().isoformat(), + "candidates": input_data.candidates, + "clicked": input_data.clicked, + } key = _SYSTEM_ATTR_PREFIX_HISTORY + history_id storage.set_study_system_attr( study_id=study_id, key=key, - value=json.dumps(history.to_dict()), + value=json.dumps(history), ) - - -def serialize_preference_history( - system_attrs: dict[str, Any], -) -> list[dict[str, Any]]: - histories: list[History] = [] - for k, v in system_attrs.items(): - if not k.startswith(_SYSTEM_ATTR_PREFIX_HISTORY): - continue - choice: dict[str, Any] = json.loads(v) - if choice["mode"] == "ChooseWorst": - histories.append( - ChooseWorstHistory( - mode="ChooseWorst", - id=choice["id"], - preference_id=choice["preference_id"], - timestamp=datetime.fromisoformat(choice["timestamp"]), - candidates=choice["candidates"], - clicked=choice["clicked"], - ) - ) - - histories.sort(key=lambda c: c.timestamp) - return [history.to_dict() for history in histories] diff --git a/optuna_dashboard/_serializer.py b/optuna_dashboard/_serializer.py index 19ceb19b..a0999b1f 100644 --- a/optuna_dashboard/_serializer.py +++ b/optuna_dashboard/_serializer.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime import json from typing import Any from typing import TYPE_CHECKING @@ -14,7 +15,9 @@ from optuna.trial import FrozenTrial from . import _note as note from ._form_widget import get_form_widgets_json from ._named_objectives import get_objective_names -from ._preferential_history import serialize_preference_history +from ._preferential_history import _SYSTEM_ATTR_PREFIX_HISTORY +from ._preferential_history import ChooseWorstHistory +from ._preferential_history import History from .artifact._backend import list_trial_artifacts from .preferential._study import _SYSTEM_ATTR_PREFERENTIAL_STUDY @@ -161,6 +164,29 @@ def serialize_study_detail( return serialized +def serialize_preference_history( + system_attrs: dict[str, Any], +) -> list[History]: + histories: list[History] = [] + for k, v in system_attrs.items(): + if not k.startswith(_SYSTEM_ATTR_PREFIX_HISTORY): + continue + choice: dict[str, Any] = json.loads(v) + if choice["mode"] == "ChooseWorst": + history: ChooseWorstHistory = { + "mode": "ChooseWorst", + "id": choice["id"], + "preference_id": choice["preference_id"], + "timestamp": choice["timestamp"], + "candidates": choice["candidates"], + "clicked": choice["clicked"], + } + histories.append(history) + + histories.sort(key=lambda c: datetime.fromisoformat(c["timestamp"])) + return histories + + def serialize_frozen_trial( study_id: int, trial: FrozenTrial, study_system_attrs: dict[str, Any] ) -> dict[str, Any]: diff --git a/optuna_dashboard/preferential/_system_attrs.py b/optuna_dashboard/preferential/_system_attrs.py index 00347d7e..e65a935d 100644 --- a/optuna_dashboard/preferential/_system_attrs.py +++ b/optuna_dashboard/preferential/_system_attrs.py @@ -16,8 +16,8 @@ def report_preferences( storage: BaseStorage, preferences: list[tuple[int, int]], ) -> str: - preference_uuid = str(uuid.uuid4()) - key = _SYSTEM_ATTR_PREFIX_PREFERENCE + preference_uuid + preference_id = str(uuid.uuid4()) + key = _SYSTEM_ATTR_PREFIX_PREFERENCE + preference_id storage.set_study_system_attr( study_id=study_id, key=key, @@ -31,7 +31,7 @@ def report_preferences( trial_id = trials[number]._trial_id if trials[number].state != TrialState.COMPLETE: storage.set_trial_state_values(trial_id, TrialState.COMPLETE, values) - return preference_uuid + return preference_id def get_preferences( diff --git a/optuna_dashboard/ts/components/PreferenceHistory.tsx b/optuna_dashboard/ts/components/PreferenceHistory.tsx index f21fd326..3bc5a750 100644 --- a/optuna_dashboard/ts/components/PreferenceHistory.tsx +++ b/optuna_dashboard/ts/components/PreferenceHistory.tsx @@ -208,7 +208,7 @@ export const PreferenceHistory: FC<{ studyDetail: StudyDetail | null }> = ({ > {preference_histories.reverse().map((choice) => ( diff --git a/python_tests/test_preferential_history.py b/python_tests/test_preferential_history.py index b23dee9e..7f184e51 100644 --- a/python_tests/test_preferential_history.py +++ b/python_tests/test_preferential_history.py @@ -2,6 +2,7 @@ from __future__ import annotations from typing import Callable +from optuna_dashboard._preferential_history import NewHistory from optuna_dashboard._preferential_history import report_history from optuna_dashboard._serializer import serialize_preference_history from optuna_dashboard.preferential import create_study @@ -25,20 +26,20 @@ def test_report_and_get_choices(storage_supplier: Callable[[], StorageSupplier]) report_history( study_id=study_id, storage=storage, - input_data={ - "mode": "ChooseWorst", - "candidates": [0, 1, 2], - "clicked": 1, - }, + input_data=NewHistory( + mode="ChooseWorst", + candidates=[0, 1, 2], + clicked=1, + ), ) report_history( study_id=study_id, storage=storage, - input_data={ - "mode": "ChooseWorst", - "candidates": [0, 2, 3, 4], - "clicked": 0, - }, + input_data=NewHistory( + mode="ChooseWorst", + candidates=[0, 2, 3, 4], + clicked=0, + ), ) history = serialize_preference_history(storage.get_study_system_attrs(study_id)) sys_attrs = storage.get_study_system_attrs(study_id)