Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/anthias_server/api/serializers/mixins.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,3 +230,7 @@ class RebootViewSerializerMixin(Serializer[Any]):

class ShutdownViewSerializerMixin(Serializer[Any]):
pass


class DisplayPowerViewSerializerMixin(Serializer[Any]):
pass
Comment thread
vpetersson marked this conversation as resolved.
Outdated
61 changes: 61 additions & 0 deletions src/anthias_server/api/tests/test_v2_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -508,3 +508,64 @@ def test_integrations_non_balena_environment(
'balena_host_os_version': None,
'balena_device_name_at_init': None,
}


# --- Display power (experimental, HDMI-CEC) -------------------------


@pytest.mark.django_db
@mock.patch(
'anthias_server.api.views.mixins.diagnostics.set_display_power',
return_value=(True, 'Display turn-on command sent.'),
)
def test_display_power_on_success(
set_display_power_mock: Any, api_client: APIClient
) -> None:
response = api_client.post(
reverse('api:display_power_v2', kwargs={'state': 'on'})
)
assert response.status_code == status.HTTP_200_OK
set_display_power_mock.assert_called_once_with(on=True)
assert 'sent' in response.data['message']


@pytest.mark.django_db
@mock.patch(
'anthias_server.api.views.mixins.diagnostics.set_display_power',
return_value=(True, 'Display turn-off command sent.'),
)
def test_display_power_off_success(
set_display_power_mock: Any, api_client: APIClient
) -> None:
response = api_client.post(
reverse('api:display_power_v2', kwargs={'state': 'off'})
)
assert response.status_code == status.HTTP_200_OK
set_display_power_mock.assert_called_once_with(on=False)


@pytest.mark.django_db
@mock.patch(
'anthias_server.api.views.mixins.diagnostics.set_display_power',
return_value=(False, 'Display turn-on failed: no adapter'),
)
def test_display_power_failure_returns_502(
set_display_power_mock: Any, api_client: APIClient
) -> None:
response = api_client.post(
reverse('api:display_power_v2', kwargs={'state': 'on'})
)
assert response.status_code == status.HTTP_502_BAD_GATEWAY
assert 'no adapter' in response.data['message']


@pytest.mark.django_db
@mock.patch('anthias_server.api.views.mixins.diagnostics.set_display_power')
def test_display_power_invalid_state_returns_400(
set_display_power_mock: Any, api_client: APIClient
) -> None:
response = api_client.post(
reverse('api:display_power_v2', kwargs={'state': 'foo'})
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
set_display_power_mock.assert_not_called()
6 changes: 6 additions & 0 deletions src/anthias_server/api/urls/v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
AssetViewV2,
BackupViewV2,
DeviceSettingsViewV2,
DisplayPowerViewV2,
FileAssetViewV2,
InfoViewV2,
IntegrationsViewV2,
Expand Down Expand Up @@ -43,6 +44,11 @@ def get_url_patterns() -> list[URLPattern | URLResolver]:
path('v2/recover', RecoverViewV2.as_view(), name='recover_v2'),
path('v2/reboot', RebootViewV2.as_view(), name='reboot_v2'),
path('v2/shutdown', ShutdownViewV2.as_view(), name='shutdown_v2'),
path(
'v2/display/<str:state>',
DisplayPowerViewV2.as_view(),
name='display_power_v2',
),
path('v2/file_asset', FileAssetViewV2.as_view(), name='file_asset_v2'),
path(
'v2/assets/<str:asset_id>/content',
Expand Down
33 changes: 33 additions & 0 deletions src/anthias_server/api/views/mixins.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from anthias_server.api.helpers import save_active_assets_ordering
from anthias_server.api.serializers.mixins import (
BackupViewSerializerMixin,
DisplayPowerViewSerializerMixin,
PlaylistOrderSerializerMixin,
RebootViewSerializerMixin,
ShutdownViewSerializerMixin,
Expand Down Expand Up @@ -179,6 +180,38 @@ def post(self, request: Request) -> Response:
return Response(status=status.HTTP_200_OK)


class DisplayPowerViewMixin(APIView):
serializer_class = DisplayPowerViewSerializerMixin

@extend_schema(
summary='Set display power state (experimental, HDMI-CEC)',
parameters=[
OpenApiParameter(
name='state',
location=OpenApiParameter.PATH,
type=OpenApiTypes.STR,
enum=['on', 'off'],
description=(
'Desired display power state. Only valid on '
'CEC-capable hardware.'
),
),
],
)
@authorized
def post(self, request: Request, state: str) -> Response:
if state not in ('on', 'off'):
return Response(
{'message': 'Invalid display state.'},
status=status.HTTP_400_BAD_REQUEST,
)
ok, msg = diagnostics.set_display_power(on=(state == 'on'))
if ok:
return Response({'message': msg}, status=status.HTTP_200_OK)
# 502: upstream CEC adapter / TV refused or didn't respond.
return Response({'message': msg}, status=status.HTTP_502_BAD_GATEWAY)

Comment thread
vpetersson marked this conversation as resolved.

Comment thread
vpetersson marked this conversation as resolved.
class FileAssetViewMixin(APIView):
@extend_schema(
summary='Upload file asset',
Expand Down
5 changes: 5 additions & 0 deletions src/anthias_server/api/views/v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
AssetsControlViewMixin,
BackupViewMixin,
DeleteAssetViewMixin,
DisplayPowerViewMixin,
FileAssetViewMixin,
InfoViewMixin,
PlaylistOrderViewMixin,
Expand Down Expand Up @@ -536,6 +537,10 @@ class ShutdownViewV2(ShutdownViewMixin):
pass


class DisplayPowerViewV2(DisplayPowerViewMixin):
pass


class FileAssetViewV2(FileAssetViewMixin):
pass

Expand Down
4 changes: 4 additions & 0 deletions src/anthias_server/app/page_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,10 @@ def device_settings() -> dict[str, Any]:
# on that revision (matches the React audio-output dropdown).
'is_pi5': 'Raspberry Pi 5' in device_model,
'date_format_options': _DATE_FORMAT_OPTIONS,
# Render-time gate for the experimental CEC display-power
# buttons; cec_available() only stats device nodes, so it's
# cheap enough to call on every settings render.
'cec_available': diagnostics.cec_available(),
}


Expand Down
13 changes: 13 additions & 0 deletions src/anthias_server/app/static/sass/_styles.scss
Original file line number Diff line number Diff line change
Expand Up @@ -1532,6 +1532,19 @@ label { text-align: left; }
display: flex;
gap: var(--space-2);
}
.settings-section__badge {
display: inline-block;
margin-left: var(--space-2);
padding: 0.125rem 0.5rem;
border-radius: var(--radius-md);
background: var(--color-warning-soft, #fef3c7);
color: var(--color-warning-strong, #92400e);
font-size: 0.6875rem;
font-weight: 600;
letter-spacing: 0.03em;
text-transform: uppercase;
vertical-align: middle;
}
}

.toggle-row {
Expand Down
30 changes: 30 additions & 0 deletions src/anthias_server/app/templates/settings.html
Original file line number Diff line number Diff line change
Expand Up @@ -247,5 +247,35 @@ <h2 x-text="pending === 'reboot' ? 'Reboot device?' : 'Shut down device?'"></h2>
</div>
</div>
</section>

{% if cec_available %}
{# ===== Display power (experimental, HDMI-CEC) ===== #}
<section class="settings-section mt-4">
<header>
<div>
<h2>Display power<span class="settings-section__badge">Experimental</span></h2>
Comment thread
vpetersson marked this conversation as resolved.
Outdated
<p class="settings-section__lede">
Toggle the connected display over HDMI-CEC. Works only on
CEC-capable hardware (most x86 PCs are not). The outcome
will appear as a toast.
</p>
</div>
<div class="settings-section__actions">
<form method="post" action="{% url 'anthias_app:settings_display_power' state='on' %}" class="inline m-0">
{% csrf_token %}
<button type="submit" class="app-btn app-btn-outline-dark">
<i class="ti ti-bulb"></i><span>Turn display on</span>
</button>
</form>
<form method="post" action="{% url 'anthias_app:settings_display_power' state='off' %}" class="inline m-0">
{% csrf_token %}
<button type="submit" class="app-btn app-btn-outline-dark">
<i class="ti ti-bulb-off"></i><span>Turn display off</span>
</button>
</form>
</div>
</header>
</section>
{% endif %}
</div>
{% endblock %}
5 changes: 5 additions & 0 deletions src/anthias_server/app/urls.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,11 @@
views.settings_shutdown,
name='settings_shutdown',
),
path(
'settings/display/<str:state>/',
views.settings_display_power,
name='settings_display_power',
),
path(
'settings/migrate-to-screenly/',
views.migrate_to_screenly,
Expand Down
11 changes: 11 additions & 0 deletions src/anthias_server/app/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -990,6 +990,17 @@ def settings_shutdown(request: HttpRequest) -> HttpResponse:
return redirect(reverse('anthias_app:settings'))


@authorized
@require_http_methods(['POST'])
def settings_display_power(request: HttpRequest, state: str) -> HttpResponse:
if state not in ('on', 'off'):
messages.error(request, 'Invalid display state.')
return redirect(reverse('anthias_app:settings'))
ok, msg = diagnostics.set_display_power(on=(state == 'on'))
(messages.success if ok else messages.error)(request, msg)
return redirect(reverse('anthias_app:settings'))


Comment thread
vpetersson marked this conversation as resolved.
@authorized
@require_http_methods(['GET'])
def system_info(request: HttpRequest) -> HttpResponse:
Expand Down
66 changes: 66 additions & 0 deletions src/anthias_server/lib/diagnostics.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,30 @@
sys.stdout.write('Unknown')
"""

# Issued from the settings page / REST endpoint, *not* from a celery
# worker, so a hung libcec call would block the request thread until
# the subprocess timeout fires. Same subprocess+timeout shape as
# `_CEC_QUERY_SCRIPT` for the same reason: libcec C calls don't
# honour Python signals.
_CEC_SET_SCRIPT = """
import sys
try:
import cec
cec.init()
tv = cec.Device(cec.CECDEVICE_TV)
except Exception as exc:
sys.stdout.write('ERROR: ' + (str(exc) or 'CEC stack unavailable'))
sys.exit(0)
try:
if {on}:
tv.power_on()
else:
tv.standby()
sys.stdout.write('OK')
except Exception as exc:
sys.stdout.write('ERROR: ' + (str(exc) or 'CEC command failed'))
"""


def get_display_power() -> str | bool:
"""
Expand Down Expand Up @@ -51,6 +75,48 @@ def get_display_power() -> str | bool:
return output or 'CEC error'


def set_display_power(on: bool) -> tuple[bool, str]:
"""Send a CEC power_on / standby to the connected TV.

Returns ``(ok, message)`` for direct surfacing to the operator as
a toast. Stays synchronous on purpose — the issue brief asks for
an immediate feedback loop so failed CEC commands aren't silent.
"""
script = _CEC_SET_SCRIPT.format(on='True' if on else 'False')
verb = 'on' if on else 'off'
try:
result = subprocess.run(
[sys.executable, '-c', script],
capture_output=True,
timeout=10,
)
except subprocess.TimeoutExpired:
return (
False,
f'Display turn-{verb} timed out — CEC adapter unresponsive.',
)

output = result.stdout.decode('utf-8', errors='replace').strip()
if output == 'OK':
return True, f'Display turn-{verb} command sent.'
if output.startswith('ERROR: '):
return False, f'Display turn-{verb} failed: {output[len("ERROR: ") :]}'
Comment thread
vpetersson marked this conversation as resolved.
Outdated
return False, f'Display turn-{verb} failed: unexpected CEC response.'

Comment thread
vpetersson marked this conversation as resolved.

def cec_available() -> bool:
"""Cheap render-time gate for whether to show CEC controls.

Probes only for the device nodes libcec consumes — `/dev/cec0`
on mainline kernels (Pi 5, x86 USB adapters when exposed) and
`/dev/vchiq` on Pi 1-4 (currently the only one passed into the
server container by `docker-compose.yml.tmpl`). A positive result
means the adapter *could* work, not that it will: the actual
success/failure is surfaced by ``set_display_power``'s toast.
"""
return os.path.exists('/dev/cec0') or os.path.exists('/dev/vchiq')


def get_uptime() -> float:
with open('/proc/uptime', 'r') as f:
uptime_seconds = float(f.readline().split()[0])
Expand Down
Loading