Skip to content
Open
Show file tree
Hide file tree
Changes from all 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
42 changes: 28 additions & 14 deletions src/pycharting/api/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,20 @@
_active_server: ChartServer | None = None


def _notify(message: str) -> None:
"""Emit a user-facing console message.

All direct, user-facing output (chart URLs, status banners) is funneled
through this single helper rather than scattered ``print`` calls, so the
output channel is consistent and can be changed in one place. Diagnostic
detail continues to flow through ``logger``.

Args:
message (str): The message to display to the user.
"""
print(message)


def plot(
index: np.ndarray | pd.Series | list,
open: np.ndarray | pd.Series | list | None = None,
Expand Down Expand Up @@ -211,7 +225,7 @@ def plot(
webbrowser.open(chart_url)
except Exception as e:
logger.warning(f"Could not open browser: {e}")
print(f"Please open this URL manually: {chart_url}")
_notify(f"Please open this URL manually: {chart_url}")

result = {
"status": "success",
Expand All @@ -223,31 +237,31 @@ def plot(
}

# Print user-friendly message
print("\n✓ Chart created successfully!")
print(f" URL: {chart_url}")
print(f" Data points: {data_manager.length:,}")
_notify("\n✓ Chart created successfully!")
_notify(f" URL: {chart_url}")
_notify(f" Data points: {data_manager.length:,}")
if not open_browser:
print(" Open the URL above in your browser to view the chart.")
print()
_notify(" Open the URL above in your browser to view the chart.")
_notify("")

# Block until server shutdown if requested
if block and _active_server: # pragma: no cover
logger.info("Blocking until chart is closed (press Ctrl+C to force exit)...")
print("Keeping server alive until you close the browser page...")
_notify("Keeping server alive until you close the browser page...")
try:
# Wait with timeout so Ctrl+C can interrupt
while not _active_server._shutdown_event.is_set():
_active_server._shutdown_event.wait(timeout=0.5)
logger.info("Server shutdown detected, returning control")
print("\n✓ Chart closed, server stopped")
_notify("\n✓ Chart closed, server stopped")
except KeyboardInterrupt:
logger.info("Interrupted by user")
print("\n⚠️ Interrupted - stopping server...")
_notify("\n⚠️ Interrupted - stopping server...")
_active_server.stop_server()

except Exception as e:
logger.error(f"Error creating chart: {e}", exc_info=True)
print(f"\n✗ Error creating chart: {e}\n")
_notify(f"\n✗ Error creating chart: {e}\n")

return {
"status": "error",
Expand All @@ -258,7 +272,7 @@ def plot(
return result


def stop_server():
def stop_server() -> None:
"""Manually shut down the active chart server.

While the server has an auto-shutdown feature (triggered after a timeout when no clients are connected),
Expand All @@ -280,9 +294,9 @@ def stop_server():
if _active_server and _active_server.is_running:
logger.info("Stopping ChartServer...")
_active_server.stop_server()
print("✓ Chart server stopped")
_notify("✓ Chart server stopped")
else:
print("ⓘ No active server to stop")
_notify("ⓘ No active server to stop")


def get_server_status() -> dict[str, Any]:
Expand Down Expand Up @@ -322,7 +336,7 @@ def get_server_status() -> dict[str, Any]:


# Jupyter notebook support
def _repr_html_():
def _repr_html_() -> str:
"""Jupyter notebook representation."""
status = get_server_status()
if status["running"]:
Expand Down
50 changes: 31 additions & 19 deletions tests/test_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,28 +5,30 @@
import numpy as np
import pytest

from pycharting.api.interface import _active_server, _repr_html_, get_server_status, plot, stop_server
from pycharting.api.interface import _repr_html_, get_server_status, plot, stop_server
from pycharting.api.routes import _data_managers


@pytest.fixture(autouse=True)
def cleanup_globals():
"""Clean up global state between tests."""
global _active_server

# Stop any active server before test
if _active_server and _active_server.is_running:
_active_server.stop_server()

# Clear data managers
_data_managers.clear()
"""Clean up global state between tests.

Operates on the live ``interface`` module global rather than a name
imported into this module — a plain ``global _active_server`` here would
rebind the test module's own copy and leave the running server (and the
real ``interface._active_server``) untouched, leaking state between tests.
"""
import pycharting.api.interface as iface

def _reset() -> None:
if iface._active_server and iface._active_server.is_running:
iface._active_server.stop_server()
iface._active_server = None
_data_managers.clear()

_reset()
yield

# Clean up after test
if _active_server and _active_server.is_running:
_active_server.stop_server()
_data_managers.clear()
_reset()


class TestPlotFunction:
Expand Down Expand Up @@ -165,10 +167,20 @@ def test_stop_server_when_running(self):
status = get_server_status()
assert status["running"] is False

def test_stop_server_when_not_running(self):
"""Test stopping when no server is active."""
# Should not raise an error
stop_server() # Should print info message
def test_stop_server_when_not_running(self, capsys):
"""Test stopping when no server is active reports the no-op message."""
import pycharting.api.interface as iface

# Force the module global to None so the "no active server" branch is
# exercised deterministically, regardless of test execution order /
# xdist worker state.
original = iface._active_server
iface._active_server = None
try:
stop_server() # Should not raise; should emit the info message
assert "No active server to stop" in capsys.readouterr().out
finally:
iface._active_server = original


class TestGetServerStatus:
Expand Down
Loading