Skip to content

Commit

Permalink
Merge pull request #109 from ImperialCollegeLondon/buttons
Browse files Browse the repository at this point in the history
Assign functionality to start & stop buttons
  • Loading branch information
tsmbland authored Dec 15, 2023
2 parents a3f21ea + 8130b9e commit 07ea5ac
Show file tree
Hide file tree
Showing 3 changed files with 156 additions and 103 deletions.
30 changes: 30 additions & 0 deletions app/datahub_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,3 +94,33 @@ def get_wesim_data() -> dict[str, dict]: # type: ignore[type-arg]
req = request_datahub("wesim")

return req.json()["data"]


def start_model() -> str:
"""Function for starting the model.
Returns:
str: Message to display on the control app
"""
try:
if request_datahub("start").json():
return "Model is already running"
response = requests.post(f"{DH_URL}/set_model_signals", params={"start": True})
return response.text
except (DataHubConnectionError, DataHubRequestError):
return "Failed to connect to the DataHub"


def stop_model() -> str:
"""Function for stopping the model.
Returns:
str: Message to display on the control app
"""
try:
if request_datahub("stop").json():
return "Model is not running"
response = requests.post(f"{DH_URL}/set_model_signals", params={"start": False})
return response.text
except (DataHubConnectionError, DataHubRequestError):
return "Failed to connect to the DataHub"
145 changes: 97 additions & 48 deletions app/pages/control.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
"""Controller Page for Dash app."""

import dash # type: ignore
from dash import Input, Output, State, callback, ctx, dcc, html # type: ignore
import requests
from dash import Input, Output, State, callback, dcc, html # type: ignore
from dash_iconify import DashIconify # type: ignore

from .. import LIVE_MODEL, log
from .. import core_api as core
from ..data import data_interval
from ..datahub_api import start_model, stop_model

dash.register_page(__name__)

Expand Down Expand Up @@ -224,9 +226,6 @@ def get_button(func: str, icon: str) -> html.Button:
Output("message", "children", allow_duplicate=True),
[
Input("button_update", "n_clicks"),
Input("button_start", "n_clicks"),
Input("button_stop", "n_clicks"),
Input("button_restart", "n_clicks"),
],
[
State("Hub01_dropdown", "value"),
Expand All @@ -241,10 +240,7 @@ def get_button(func: str, icon: str) -> html.Button:
prevent_initial_call=True,
)
def update_button_click(
button_update: int | None,
button_start: int | None,
button_stop: int | None,
button_restart: int | None,
n_clicks: int | None,
Hub01_dropdown: str,
Hub02_dropdown: str,
PC01_Top_dropdown: str,
Expand All @@ -254,47 +250,24 @@ def update_button_click(
PC02_Left_dropdown: str,
PC02_Right_dropdown: str,
) -> list[str]:
"""Placeholder function for buttons."""
button_id = ctx.triggered_id[7:]

if button_id == "update":
"""Will make an API call to set up OVE sections accoding to dropdowns.
Args: Value inputs for the 8 dropdown menus
"""
log.debug("Clicked Update Button!")
message = core.assign_sections(
{
"Hub01": Hub01_dropdown,
"Hub02": Hub02_dropdown,
"PC01-Top": PC01_Top_dropdown,
"PC01-Left": PC01_Left_dropdown,
"PC01-Right": PC01_Right_dropdown,
"PC02-Top": PC02_Top_dropdown,
"PC02-Left": PC02_Left_dropdown,
"PC02-Right": PC02_Right_dropdown,
}
)
return [message]

elif button_id == "start":
"""Will make an API call to start the Gridlington simulation and Datahub."""
log.debug("Clicked Start Button!")
return ["Clicked Start Button!"]

elif button_id == "stop":
"""Will make an API call to stop the Gridlington simulation and Datahub."""
log.debug("Clicked Stop Button!")
return ["Clicked Stop Button!"]

elif button_id == "restart":
"""Will make an API call to restart the Gridlington simulation and Datahub."""
log.debug("Clicked Restart Button!")
core.refresh_sections()
return ["Clicked Restart Button!"]
"""Will make an API call to set up OVE sections accoding to dropdowns.
else:
return [""]
Args: Value inputs for the 8 dropdown menus
"""
log.debug("Clicked Update Button!")
message = core.assign_sections(
{
"Hub01": Hub01_dropdown,
"Hub02": Hub02_dropdown,
"PC01-Top": PC01_Top_dropdown,
"PC01-Left": PC01_Left_dropdown,
"PC01-Right": PC01_Right_dropdown,
"PC02-Top": PC02_Top_dropdown,
"PC02-Left": PC02_Left_dropdown,
"PC02-Right": PC02_Right_dropdown,
}
)
return [message]


@callback(
Expand Down Expand Up @@ -328,6 +301,82 @@ def default_button_click(n_clicks: int | None) -> list[str]:
]


@callback(
[
Output("message", "children", allow_duplicate=True),
Output("data_interval", "disabled", allow_duplicate=True),
],
[Input("button_start", "n_clicks")],
prevent_initial_call=True,
)
def start_button_click(n_clicks: int | None) -> tuple[str, bool]:
"""Function for start button.
Args:
n_clicks (int | None): Number of times the button has been clicked
Returns:
str: Message to display on the control app
bool: Whether to disable data updates
"""
message = start_model() if LIVE_MODEL else "Playback started"
log.debug(message)
return message, False


@callback(
[
Output("message", "children", allow_duplicate=True),
Output("data_interval", "disabled", allow_duplicate=True),
],
[Input("button_stop", "n_clicks")],
prevent_initial_call=True,
)
def stop_button_click(n_clicks: int | None) -> tuple[str, bool]:
"""Function for stop button.
Args:
n_clicks (int | None): Number of times the button has been clicked
Returns:
str: Message to display on the control app
bool: Whether to disable data updates
"""
message = stop_model() if LIVE_MODEL else "Playback stopped"
log.debug(message)
return message, True


@callback(
[
Output("message", "children", allow_duplicate=True),
Output("data_interval", "n_intervals"),
Output("data_interval", "disabled", allow_duplicate=True),
],
[Input("button_restart", "n_clicks")],
prevent_initial_call=True,
)
def restart_button_click(n_clicks: int | None) -> tuple[str, int, bool]:
"""Function for restart button.
TODO: this should send a signal to DataHub to restart the model
Args:
n_clicks (int | None): Number of times the button has been clicked
Returns:
str: Message for the control app
int: 0 sends data interval back to the beginning
bool: False (re-)enables data updates
"""
log.debug("Clicked Restart Button!")
try:
core.refresh_sections()
except requests.exceptions.ConnectionError:
pass
return "Clicked Restart Button!", 0, False


@callback(
[Output("data_interval", "interval")], [Input("update-interval-slider", "value")]
)
Expand Down
84 changes: 29 additions & 55 deletions tests/test_control_view.py
Original file line number Diff line number Diff line change
@@ -1,76 +1,50 @@
from contextvars import copy_context
from app.pages.control import (
default_button_click,
restart_button_click,
start_button_click,
stop_button_click,
update_button_click,
update_data_interval,
)

from dash._callback_context import context_value # type: ignore
from dash._utils import AttributeDict # type: ignore


def test_control_view_callback(mocker):
"""Test for Control View callback function and buttons."""
from app.pages.control import update_button_click

def run_callback():
context_value.set(
AttributeDict(
**{"triggered_inputs": [{"prop_id": f"{button_id}.n_clicks"}]}
)
)
return update_button_click(
buttons[0],
buttons[1],
buttons[2],
buttons[3],
"",
"",
"",
"",
"",
"",
"",
"",
)

def test_update_button_callback(mocker):
"""Test Update Button."""
buttons = [1, None, None, None]
button_id = "button_update"

ctx = copy_context()
patched_assign_sections = mocker.patch(
"app.core_api.assign_sections", return_value=button_id
"app.core_api.assign_sections", return_value="button_update"
)
output = ctx.run(run_callback)
output = update_button_click(0, "", "", "", "", "", "", "", "")
patched_assign_sections.assert_called_once()
assert output[0] == button_id
assert output[0] == "button_update"


def test_start_button_callback():
"""Test Start Button."""
buttons = [None, 1, None, None]
button_id = "button_start"
output = start_button_click(0)
assert output[1] is False

ctx = copy_context()
output = ctx.run(run_callback)
assert output[0] == "Clicked Start Button!"

def test_stop_button_callback():
"""Test Stop Button."""
buttons = [None, None, 1, None]
button_id = "button_stop"
output = stop_button_click(0)
assert output[1] is True

ctx = copy_context()
output = ctx.run(run_callback)
assert output[0] == "Clicked Stop Button!"

"""Test Restart Button."""
buttons = [None, None, None, 1]
button_id = "button_restart"

ctx = copy_context()
def test_restart_button_callback(mocker):
"""Test Reset Button."""
patched_refresh_sections = mocker.patch("app.core_api.refresh_sections")
output = ctx.run(run_callback)
output = restart_button_click(0)
patched_refresh_sections.assert_called_once()
assert output[0] == "Clicked Restart Button!"
assert output[1] == 0


def test_default_button_callback():
"""Test Default Button."""
from app.pages.control import default_button_click

output = default_button_click(0)
assert output[0] == "Dropdowns returned to default values. Click tick to assign."


def test_data_interval_slider_callback():
"""Test Interval Slider."""
output = update_data_interval(2)
assert output[0] == 2000

0 comments on commit 07ea5ac

Please sign in to comment.