Merge branch 'PLAT-276-add-oem-mode-endpoint' into oem-mode-integration

api/src/opentrons/config/feature_flags.py

@@ -80,3 +80,9 @@ def enable_error_recovery_experiments() -> bool:
 
 def enable_performance_metrics(robot_type: RobotTypeEnum) -> bool:
     return advs.get_setting_with_env_overload("enablePerformanceMetrics", robot_type)
+
+
+def oem_mode_enabled() -> bool:
+    return advs.get_setting_with_env_overload(
+        "enableOEMMode", RobotTypeEnum.FLEX
+    )

robot-server/robot_server/service/legacy/routers/settings.py

@@ -1,7 +1,7 @@
-from dataclasses import asdict
+import aiohttp
 import logging
+from dataclasses import asdict
 from typing import cast, Any, Dict, List, Optional, Union
-
 from starlette import status
 from fastapi import APIRouter, Depends
 
@@ -64,6 +64,15 @@
 router = APIRouter()
 
 
+async def set_oem_mode_request(enable):
+    """PUT request to set the OEM Mode for the system server."""
+    async with aiohttp.ClientSession() as session:
+        async with session.put(
+            "http://127.0.0.1:31950/system/oem_mode/enable", json={"enable": enable}
+        ) as resp:
+            return resp.status
+
+
 @router.post(
     path="/settings",
     summary="Change a setting",
@@ -82,6 +91,13 @@ async def post_settings(
 ) -> AdvancedSettingsResponse:
     """Update advanced setting (feature flag)"""
     try:
+        # send request to system server if this is the enableOEMMode setting
+        if update.id == "enableOEMMode":
+            resp = await set_oem_mode_request(update.value)
+            if resp != 200:
+                # TODO: raise correct error here
+                raise Exception(f"Something went wrong setting OEM Mode. err: {resp}")
+
         await advanced_settings.set_adv_setting(update.id, update.value)
         hardware.hardware_feature_flags = HardwareFeatureFlags.build_from_ff()
         await hardware.set_status_bar_enabled(ff.status_bar_enabled())

system-server/settings_schema.json

@@ -9,6 +9,19 @@
       "default": "/var/lib/opentrons-system-server/",
       "env_names": ["ot_system_server_persistence_directory"],
       "type": "string"
+    },
+    "oem_mode_enabled": {
+        "title": "OEM Mode Enabled",
+        "description": "A flag used to change the default splash screen on system startup. If this flag is disabled (default), the Opentrons loading video will be shown. If this flag is enabled but `oem_mode_splash_custom` is not set, then the default OEM Mode splash screen will be shown. If this flag is enabled and `oem_mode_splash_custom` is set to a PNG filepath, the custom splash screen will be shown.",
+        "default": false,
+        "env_names": ["ot_system_server_oem_mode_enabled"],
+        "type": "bool"
+    },
+    "oem_mode_splash_custom": {
+        "description": "The filepath of the PNG image used as the custom splash screen. Read the description of the `oem_mode_enabled` flag to know how the splash screen changes when the flag is enabled/disabled.",
+        "default": null,
+        "env_names": ["ot_system_server_oem_mode_splash_custom"],
+        "type": "string"
     }
   },
   "additionalProperties": false

system-server/system_server/settings/__init__.py

@@ -1,6 +1,10 @@
 """system_server.settings: Provides an interface to get server settings."""
 
-from .settings import get_settings, SystemServerSettings
+from .settings import (
+    save_settings,
+    get_settings,
+    SystemServerSettings,
+)
 
 
-__all__ = ["get_settings", "SystemServerSettings"]
+__all__ = ["save_settings", "get_settings", "SystemServerSettings"]

system-server/system_server/settings/settings.py

@@ -1,12 +1,9 @@
 """System server configuration options."""
 import typing
-import logging
 from functools import lru_cache
 
 from pydantic import BaseSettings, Field
-from dotenv import load_dotenv
-
-log = logging.getLogger(__name__)
+from dotenv import load_dotenv, set_key
 
 
 @lru_cache(maxsize=1)
@@ -56,7 +53,43 @@ class SystemServerSettings(BaseSettings):
         ),
     )
 
+    oem_mode_enabled: typing.Optional[bool] = Field(
+        default=False,
+        description=(
+            "A flag used to change the default splash screen on system startup."
+            " If this flag is disabled (default), the Opentrons loading video will be shown."
+            " If this flag is enabled but `oem_mode_splash_custom` is not set,"
+            " then the default OEM Mode splash screen will be shown."
+            " If this flag is enabled and `oem_mode_splash_custom` is set to a"
+            " PNG filepath, the custom splash screen will be shown."
+        ),
+    )
+
+    oem_mode_splash_custom: typing.Optional[str] = Field(
+        default=None,
+        description=(
+            "The filepath of the PNG image used as the custom splash screen."
+            " Read the description of the `oem_mode_enabled` flag to know how"
+            " the splash screen changes when the flag is enabled/disabled."
+        ),
+    )
+
     class Config:
         """Prefix configuration for environment variables."""
 
         env_prefix = "OT_SYSTEM_SERVER_"
+
+
+def save_settings(settings: SystemServerSettings) -> bool:
+    """Save the settings to the dotenv file."""
+    env_path = Environment().dot_env_path
+    env_path = env_path or f"{settings.persistence_directory}/system.env"
+    prefix = settings.Config.env_prefix
+    try:
+        for key, val in settings.dict().items():
+            name = f"{prefix}{key}"
+            value = str(val) if val is not None else ""
+            set_key(env_path, name, value)
+        return True
+    except (IOError, ValueError):
+        return False

system-server/system_server/system/oem_mode/__init__.py

@@ -0,0 +1,5 @@
+"""OEM Mode endpoint."""
+
+from .router import oem_mode_router
+
+__all__ = ["oem_mode_router"]

system-server/system_server/system/oem_mode/dependencies.py

@@ -0,0 +1,21 @@
+"""Dependencies for /system/register endpoints."""
+from system_server.jwt import Registrant
+from fastapi import Query
+
+
+def create_registrant(
+    subject: str = Query(
+        ..., description="Identifies the human intending to register with the robot"
+    ),
+    agent: str = Query(..., description="Identifies the app type making the request"),
+    agentId: str = Query(
+        ..., description="A unique identifier for the instance of the agent"
+    ),
+) -> Registrant:
+    """Define a unique Registrant to create a registration token for.
+
+    A registrant is defined by a set of unique identifiers that remain
+    persistent indefinitely for the same person using the same method of
+    access to the system.
+    """
+    return Registrant(subject=subject, agent=agent, agent_id=agentId)

system-server/system_server/system/oem_mode/models.py

@@ -0,0 +1,9 @@
+"""Models for /system/oem_mode."""
+
+from pydantic import BaseModel, Field
+
+
+class EnableOEMMode(BaseModel):
+    """Enable OEM Mode model."""
+
+    enable: bool = Field(..., description="Enable or Disable OEM Mode.")

system-server/system_server/system/oem_mode/router.py

@@ -0,0 +1,38 @@
+"""Router for /system/register endpoint."""
+
+from fastapi import APIRouter, Depends, status, Response
+from .models import EnableOEMMode
+from ...settings import SystemServerSettings, get_settings, save_settings
+
+
+oem_mode_router = APIRouter()
+
+
+@oem_mode_router.put(
+    "/system/oem_mode/enable",
+    summary="Enable or Disable OEM Mode for this robot.",
+    responses={
+        status.HTTP_200_OK: {"message": "OEM Mode changed successfully."},
+        status.HTTP_400_BAD_REQUEST: {"message": "OEM Mode did not changed."},
+        status.HTTP_500_INTERNAL_SERVER_ERROR: {
+            "message": "OEM Mode unhandled exception."
+        },
+    },
+)
+async def enable_oem_mode_endpoint(
+    response: Response,
+    enableRequest: EnableOEMMode,
+    settings: SystemServerSettings = Depends(get_settings),
+) -> Response:
+    """Router for /system/oem_mode/enable endpoint."""
+
+    enable = enableRequest.enable
+    try:
+        settings.oem_mode_enabled = enable
+        success = save_settings(settings)
+        response.status_code = (
+            status.HTTP_200_OK if success else status.HTTP_400_BAD_REQUEST
+        )
+    except Exception:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+    return response

system-server/system_server/system/router.py

@@ -3,6 +3,7 @@
 from .register.router import register_router
 from .authorize.router import authorize_router
 from .connected.router import connected_router
+from .oem_mode.router import oem_mode_router
 
 system_router = APIRouter()
 
@@ -11,3 +12,5 @@
 system_router.include_router(router=authorize_router)
 
 system_router.include_router(router=connected_router)
+
+system_router.include_router(router=oem_mode_router)