Skip to content

Integrated Variant

The optional webatm_integrated package used by the integrated build. It is wired into the core app only when WEBATM_INTEGRATED=1.

webatm_integrated

webatm_integrated

Provide the WebATM integrated extensions.

Shipped ONLY in the webatm-integrated build variant. Adds BlueSky server lifecycle control (start/stop/restart/kill), live, in-order log streaming of the bluesky --headless process tree to the web UI, and -- because BlueSky runs in this same container -- pre-wires WebATM's file management directly to BlueSky's own scenario / plugins / output directories (no manual base-path step).

The core webatm package never imports this package. It is wired in through a single env-guarded hook in WebATM.app.create_app that calls register only when WEBATM_INTEGRATED=1 and this package is installed.

LogStreamer 

LogStreamer(
    socketio,
    max_history: int = 2000,
    batch_ms: int = 100,
    batch_max: int = 200,
)

Buffer process output and broadcast it as ordered, batched events.

Initialize the streamer.

Parameters:

Name Type Description Default
socketio SocketIO

Instance used to emit batches.

 required
max_history int

Maximum lines retained for history replay.

 2000
batch_ms int

Delay in milliseconds used to coalesce a batch.

 100
batch_max int

Maximum lines per emitted batch chunk.

 200
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    socketio,
    max_history: int = 2000,
    batch_ms: int = 100,
    batch_max: int = 200,
):
    """Initialize the streamer.

    Args:
        socketio (flask_socketio.SocketIO): Instance used to emit batches.
        max_history (int): Maximum lines retained for history replay.
        batch_ms (int): Delay in milliseconds used to coalesce a batch.
        batch_max (int): Maximum lines per emitted batch chunk.
    """
    self._sio = socketio
    self._lock = threading.Lock()
    self._history: collections.deque[dict] = collections.deque(maxlen=max_history)
    self._pending: list[dict] = []
    self._seq = 0
    self._flush_scheduled = False
    self._batch_ms = batch_ms
    self._batch_max = batch_max

feed_line 

feed_line(line: str) -> None

Ingest one output line, assign its order, and schedule a flush.

Parameters:

Name Type Description Default
line str

The process output line to broadcast.

 required
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def feed_line(self, line: str) -> None:
    """Ingest one output line, assign its order, and schedule a flush.

    Args:
        line (str): The process output line to broadcast.
    """
    with self._lock:
        self._seq += 1
        item = {"seq": self._seq, "t": time.time(), "line": line}
        self._history.append(item)
        self._pending.append(item)
        if not self._flush_scheduled:
            self._flush_scheduled = True
            try:
                self._sio.start_background_task(self._flush_after_delay)
            except Exception:
                # Un-wedge the scheduler: the line stays pending and the
                # next feed_line retries; a stuck True flag would silence
                # the stream forever.
                self._flush_scheduled = False
                raise

history 

history() -> list[dict]

Return a snapshot of buffered lines for late-joining clients.

Returns:

Type Description
list[dict]

Buffered items with seq, t and line keys.
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
79
80
81
82
83
84
85
86
def history(self) -> list[dict]:
    """Return a snapshot of buffered lines for late-joining clients.

    Returns:
        list[dict]: Buffered items with ``seq``, ``t`` and ``line`` keys.
    """
    with self._lock:
        return list(self._history)

on_process_exit 

on_process_exit(return_code: int) -> None

Emit an end-of-stream marker when the server process exits.

Parameters:

Name Type Description Default
return_code int

Exit code of the BlueSky server process.

 required
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
88
89
90
91
92
93
94
def on_process_exit(self, return_code: int) -> None:
    """Emit an end-of-stream marker when the server process exits.

    Args:
        return_code (int): Exit code of the BlueSky server process.
    """
    self.feed_line(f"--- bluesky server exited (return code {return_code}) ---")

BlueSkyProcessManager 

BlueSkyProcessManager(
    on_line: Callable[[str], None] | None = None,
    on_exit: Callable[[int], None] | None = None,
    spawn: Callable | None = None,
    cmd: list[str] | None = None,
)

Thread-safe lifecycle manager for the bluesky --headless process tree.

Tracks only the parent process; signals (stop/kill) address the whole process group so node children are reaped together with the server.

Initialize the process manager.

Parameters:

Name Type Description Default
on_line Callable[[str], None] | None

Callback invoked with each output line of the process tree (newline stripped).

 None
on_exit Callable[[int], None] | None

Callback invoked with the return code when the server process exits.

 None
spawn Callable | None

Spawn primitive for the reader task (e.g. socketio.start_background_task); defaults to a plain daemon thread.

 None
cmd list[str] | None

Command to launch; defaults to ["bluesky", "--headless"].

 None
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    on_line: Callable[[str], None] | None = None,
    on_exit: Callable[[int], None] | None = None,
    spawn: Callable | None = None,
    cmd: list[str] | None = None,
):
    """Initialize the process manager.

    Args:
        on_line: Callback invoked with each output line of the process
            tree (newline stripped).
        on_exit: Callback invoked with the return code when the server
            process exits.
        spawn: Spawn primitive for the reader task (e.g.
            ``socketio.start_background_task``); defaults to a plain
            daemon thread.
        cmd (list[str] | None): Command to launch; defaults to
            ``["bluesky", "--headless"]``.
    """
    self._lock = threading.RLock()
    self._proc: subprocess.Popen | None = None
    self._on_line = on_line
    self._on_exit = on_exit
    self._spawn = spawn or _default_spawn
    self._cmd = cmd or ["bluesky", "--headless"]
    self._state = "stopped"  # stopped | starting | running | stopping

start 

start() -> dict

Spawn the headless server (in its own process group) and a reader.

The process is started in a new session with merged, line-buffered stdout/stderr so the reader receives one ordered stream for the server and all node children. A no-op if the server is already running. If a concurrent :meth:stop is mid-shutdown, waits for it to finish and then starts a fresh server instead of reporting the doomed process as "already running".

Returns:

Type Description
dict

Result with success, status, pid and message (plus error on failure).
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def start(self) -> dict:
    """Spawn the headless server (in its own process group) and a reader.

    The process is started in a new session with merged, line-buffered
    stdout/stderr so the reader receives one ordered stream for the
    server and all node children. A no-op if the server is already
    running. If a concurrent :meth:`stop` is mid-shutdown, waits for it
    to finish and then starts a fresh server instead of reporting the
    doomed process as "already running".

    Returns:
        dict: Result with ``success``, ``status``, ``pid`` and
            ``message`` (plus ``error`` on failure).
    """
    with self._lock:
        stopping_proc = self._proc if self._state == "stopping" else None
    if stopping_proc is not None:
        # A concurrent stop() owns this process's shutdown; treating it as
        # "already running" would return a pid that is about to die. Wait
        # out the stop (its worst case is escalate_after + the 5s SIGKILL
        # wait), then start fresh below.
        try:
            stopping_proc.wait(timeout=15)
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server is still stopping; retry shortly",
            }

    with self._lock:
        if self._proc is not None and self._proc.poll() is None:
            return {
                "success": True,
                "status": "running",
                "pid": self._proc.pid,
                "message": "BlueSky server already running",
            }
        self._state = "starting"
        # PYTHONUNBUFFERED keeps the server's (and inherited children's)
        # stdout line-buffered so log lines arrive promptly and in order.
        env = dict(os.environ, PYTHONUNBUFFERED="1")
        try:
            self._proc = subprocess.Popen(
                self._cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,  # merge -> single ordered stream
                stdin=subprocess.DEVNULL,
                bufsize=1,
                text=True,
                env=env,
                start_new_session=True,  # own process group / session
                close_fds=True,
            )
        except Exception as e:
            self._state = "stopped"
            return {
                "success": False,
                "status": "error",
                "message": f"Failed to start BlueSky: {e}",
                "error": str(e),
            }
        self._state = "running"
        proc = self._proc

    # Launch the reader outside the lock.
    self._spawn(self._read_loop, proc)
    return {
        "success": True,
        "status": "running",
        "pid": proc.pid,
        "message": "BlueSky server started",
    }

stop 

stop(
    sig: int = signal.SIGTERM, escalate_after: float = 5.0
) -> dict

Signal the whole process group, escalating to SIGKILL if needed.

Parameters:

Name Type Description Default
sig int

Signal sent to the process group first.

 SIGTERM
escalate_after float

Seconds to wait for exit before force-killing the group with SIGKILL.

 5.0

Returns:

Type Description
dict

Result with success, status and message. success is False if the process survives even SIGKILL.
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def stop(self, sig: int = signal.SIGTERM, escalate_after: float = 5.0) -> dict:
    """Signal the whole process group, escalating to SIGKILL if needed.

    Args:
        sig (int): Signal sent to the process group first.
        escalate_after (float): Seconds to wait for exit before
            force-killing the group with SIGKILL.

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
            ``success`` is False if the process survives even SIGKILL.
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server is not running",
            }
        self._state = "stopping"
        try:
            pgid = os.getpgid(proc.pid)
        except ProcessLookupError:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server already exited",
            }

    # Signal the whole group (server + all node children) outside the lock.
    try:
        os.killpg(pgid, sig)
    except ProcessLookupError:
        pass

    try:
        proc.wait(timeout=escalate_after)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(pgid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            logger.error("BlueSky process group %s survived SIGKILL", pgid)
            with self._lock:
                if self._proc is proc:
                    self._state = "running"
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server did not exit after SIGKILL",
            }

    with self._lock:
        if self._proc is proc:
            self._state = "stopped"
    return {
        "success": True,
        "status": "stopped",
        "message": "BlueSky server stopped",
    }

kill 

kill() -> dict

Force-kill the whole process group immediately (no graceful wait).

Returns:

Type Description
dict

Result with success, status and message.
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
237
238
239
240
241
242
243
244
245
246
def kill(self) -> dict:
    """Force-kill the whole process group immediately (no graceful wait).

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
    """
    result = self.stop(sig=signal.SIGKILL, escalate_after=2.0)
    if result.get("success") and result.get("status") == "stopped":
        result["message"] = "BlueSky server killed"
    return result

restart 

restart() -> dict

Stop the current tree (if any) and start a fresh one.

Returns:

Type Description
dict

The start() result, with the message adjusted to "restarted" on success. If the old tree could not be stopped, its failure result is returned instead of starting a new one (start() would just report the surviving process as "already running").
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def restart(self) -> dict:
    """Stop the current tree (if any) and start a fresh one.

    Returns:
        dict: The ``start()`` result, with the message adjusted to
            "restarted" on success. If the old tree could not be stopped,
            its failure result is returned instead of starting a new one
            (``start()`` would just report the surviving process as
            "already running").
    """
    stop_result = self.stop()
    if not stop_result.get("success"):
        return stop_result
    result = self.start()
    if result.get("success"):
        result["message"] = "BlueSky server restarted"
    return result

status 

status() -> dict

Report whether the server is running, with its pid and state.

Returns:

Type Description
dict

Result with success, running, status and pid (None when stopped).
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def status(self) -> dict:
    """Report whether the server is running, with its pid and state.

    Returns:
        dict: Result with ``success``, ``running``, ``status`` and
            ``pid`` (None when stopped).
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            return {
                "success": True,
                "running": False,
                "status": "stopped",
                "pid": None,
            }
        return {
            "success": True,
            "running": True,
            "status": self._state,
            "pid": proc.pid,
        }

register 

register(
    app,
    socketio,
    *,
    session_manager=None,
    bluesky_proxy=None,
)

Wire the integrated features into an existing WebATM app.

Called by WebATM.app.create_app when WEBATM_INTEGRATED=1. Points the file-management routes at BlueSky's fixed working directory, creates the process manager and log streamer (stashed on app as bluesky_process_manager / bluesky_log_streamer), registers the integrated REST routes and Socket.IO handlers, and arranges for the whole BlueSky process group to be reaped when the worker exits. On the first boot only (guarded by claim_first_boot(), so a replaced gunicorn worker never resurrects a manually-stopped server) it also schedules the background auto-start of the bundled BlueSky server; disable that with WEBATM_AUTO_START=0.

Parameters:

Name Type Description Default
app Flask

Flask application instance.

 required
socketio SocketIO

Flask-SocketIO instance (async_mode="threading").

 required
session_manager SessionManager

Core session manager (accepted for forward-compat; currently unused).

 None
bluesky_proxy BlueSkyProxy

Core proxy, used for the first-boot auto-connect.

 None

Returns:

Type Description
dict

The created manager (BlueSkyProcessManager) and streamer (LogStreamer), handy for tests.
Source code in WebATM-integrated/webatm_integrated/__init__.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def register(app, socketio, *, session_manager=None, bluesky_proxy=None):
    """Wire the integrated features into an existing WebATM app.

    Called by ``WebATM.app.create_app`` when ``WEBATM_INTEGRATED=1``. Points the
    file-management routes at BlueSky's fixed working directory, creates the
    process manager and log streamer (stashed on ``app`` as
    ``bluesky_process_manager`` / ``bluesky_log_streamer``), registers the
    integrated REST routes and Socket.IO handlers, and arranges for the whole
    BlueSky process group to be reaped when the worker exits. On the first boot
    only (guarded by ``claim_first_boot()``, so a replaced gunicorn worker never
    resurrects a manually-stopped server) it also schedules the background
    auto-start of the bundled BlueSky server; disable that with
    ``WEBATM_AUTO_START=0``.

    Args:
        app (flask.Flask): Flask application instance.
        socketio (flask_socketio.SocketIO): Flask-SocketIO instance
            (``async_mode="threading"``).
        session_manager (SessionManager): Core session manager (accepted for
            forward-compat; currently unused).
        bluesky_proxy (BlueSkyProxy): Core proxy, used for the first-boot
            auto-connect.

    Returns:
        dict: The created ``manager`` (BlueSkyProcessManager) and ``streamer``
            (LogStreamer), handy for tests.
    """
    # Imported lazily so the Flask-importing modules stay out of flask-free
    # unit-test imports of process_manager / log_streamer.
    from .auto_start import auto_start_enabled, claim_first_boot, schedule_auto_start
    from .bluesky_paths import configure_file_management
    from .routes import register_integrated_routes
    from .socket_handlers import register_integrated_socket_handlers

    configure_file_management(app)

    streamer = LogStreamer(socketio)
    manager = BlueSkyProcessManager(
        on_line=streamer.feed_line,
        on_exit=streamer.on_process_exit,
        spawn=socketio.start_background_task,
    )

    # Stash on the app so blueprint-free route/handler closures can reach them
    # via flask.current_app if needed.
    app.bluesky_process_manager = manager
    app.bluesky_log_streamer = streamer

    register_integrated_routes(app, manager)
    register_integrated_socket_handlers(socketio, streamer)

    # Reap the whole bluesky process group if the worker process exits.
    atexit.register(manager.kill)

    # First boot only (see the docstring above): auto-start BlueSky and connect
    # the proxy in the background so app creation returns promptly.
    if auto_start_enabled() and claim_first_boot():
        schedule_auto_start(socketio, manager, bluesky_proxy)

    return {"manager": manager, "streamer": streamer}

webatm_integrated.auto_start

webatm_integrated.auto_start

Auto-start the bundled BlueSky server and connect the proxy on first boot.

Shipped ONLY in the webatm-integrated build. In this variant BlueSky runs inside the same container as the WebATM backend, so there is no reason to make the user open Settings and click Start/Connect before anything works: on start-up we spawn the bluesky --headless process tree and, once its command/data ports accept connections, connect the WebATM proxy to it. The user then lands on a live, already-connected map.

This mirrors -- server-side and automatically -- the exact connect sequence the manual /api/server/config route performs (start_client then register_subscribers; subscribers can only attach once the client exists).

Opt out with WEBATM_AUTO_START=0 (e.g. for tests, or deployments that want the manual Start button to drive the lifecycle). The core webatm package never imports this module; it is reached only via webatm_integrated.register (env-guarded on WEBATM_INTEGRATED=1).

auto_start_enabled 

auto_start_enabled() -> bool

Report whether to auto-start BlueSky and auto-connect on boot.

On by default; set WEBATM_AUTO_START=0 to disable.

Returns:

Type Description
bool

True unless the WEBATM_AUTO_START environment variable is set to "0".
Source code in WebATM-integrated/webatm_integrated/auto_start.py 
38
39
40
41
42
43
44
45
46
47
def auto_start_enabled() -> bool:
    """Report whether to auto-start BlueSky and auto-connect on boot.

    On by default; set ``WEBATM_AUTO_START=0`` to disable.

    Returns:
        bool: True unless the ``WEBATM_AUTO_START`` environment variable is
            set to ``"0"``.
    """
    return os.environ.get("WEBATM_AUTO_START", "1") != "0"

claim_first_boot 

claim_first_boot(marker_path: str | None = None) -> bool

Atomically claim the one-shot auto-start for this boot.

Creates the marker file with O_CREAT | O_EXCL so only the first caller per boot wins; every later caller (e.g. a replaced gunicorn worker re-running register()) stands down, and auto-start never fights the manual Start/Stop controls. The default marker lives on tmpfs (/dev/shm) so it survives worker replacement but clears on a fresh container start.

Parameters:

Name Type Description Default
marker_path str | None

Marker file location. Defaults to the WEBATM_AUTOSTART_MARKER environment variable, falling back to /dev/shm/webatm_autostart.done.

 None

Returns:

Type Description
bool

True for the first caller to create the marker file, False thereafter. If the marker cannot be created at all (e.g. no /dev/shm on a dev box) it degrades to True, proceeding without the once-per-boot guard.
Source code in WebATM-integrated/webatm_integrated/auto_start.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def claim_first_boot(marker_path: str | None = None) -> bool:
    """Atomically claim the one-shot auto-start for this boot.

    Creates the marker file with ``O_CREAT | O_EXCL`` so only the first caller
    per boot wins; every later caller (e.g. a replaced gunicorn worker re-running
    ``register()``) stands down, and auto-start never fights the manual
    Start/Stop controls. The default marker lives on tmpfs (``/dev/shm``) so it
    survives worker replacement but clears on a fresh container start.

    Args:
        marker_path (str | None): Marker file location. Defaults to the
            ``WEBATM_AUTOSTART_MARKER`` environment variable, falling back to
            ``/dev/shm/webatm_autostart.done``.

    Returns:
        bool: True for the first caller to create the marker file, False
            thereafter. If the marker cannot be created at all (e.g. no
            ``/dev/shm`` on a dev box) it degrades to True, proceeding without
            the once-per-boot guard.
    """
    path = marker_path or os.environ.get("WEBATM_AUTOSTART_MARKER", _DEFAULT_MARKER)
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError:
        logger.info("Auto-start: already performed this boot; skipping")
        return False
    except OSError as e:
        logger.warning(
            f"Auto-start: could not create boot marker '{path}' ({e}); "
            "proceeding without the once-per-boot guard"
        )
        return True
    try:
        os.write(fd, f"pid={os.getpid()}\n".encode())
    finally:
        os.close(fd)
    return True

schedule_auto_start 

schedule_auto_start(
    socketio, manager, bluesky_proxy
) -> None

Run the auto-start sequence on a background task.

Backgrounded so register() (which runs during app creation) returns immediately: waiting for BlueSky's ports can take several seconds on a cold start, and we must not block the worker from beginning to serve requests.

Parameters:

Name Type Description Default
socketio SocketIO

Socket.IO instance whose start_background_task runs the sequence.

 required
manager BlueSkyProcessManager

Process manager used to start the BlueSky server.

 required
bluesky_proxy BlueSkyProxy | None

Core proxy to connect once BlueSky is ready, or None to skip the auto-connect.

 required
Source code in WebATM-integrated/webatm_integrated/auto_start.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def schedule_auto_start(socketio, manager, bluesky_proxy) -> None:
    """Run the auto-start sequence on a background task.

    Backgrounded so ``register()`` (which runs during app creation) returns
    immediately: waiting for BlueSky's ports can take several seconds on a cold
    start, and we must not block the worker from beginning to serve requests.

    Args:
        socketio (flask_socketio.SocketIO): Socket.IO instance whose
            ``start_background_task`` runs the sequence.
        manager (BlueSkyProcessManager): Process manager used to start the
            BlueSky server.
        bluesky_proxy (BlueSkyProxy | None): Core proxy to connect once
            BlueSky is ready, or None to skip the auto-connect.
    """
    socketio.start_background_task(_run_auto_start, manager, bluesky_proxy)

connect_proxy_when_ready 

connect_proxy_when_ready(
    bluesky_proxy,
    *,
    host: str | None = None,
    ready_timeout: float = 60.0,
    poll_interval: float = 0.5,
    is_port_listening: Callable[..., bool] | None = None,
    register_subscribers: Callable[[], None] | None = None,
    sleep: Callable[[float], None] | None = None,
) -> bool

Wait for BlueSky to accept connections, then connect the WebATM proxy.

Polls BlueSky's command/data ports until one is listening (or the timeout elapses), then performs the same connect sequence as the manual route: start_client followed by register_subscribers (subscribers attach to the client created by start_client).

The port probe, subscriber registration and sleep are injectable so this can be unit-tested without a real BlueSky server or wall-clock delays.

Parameters:

Name Type Description Default
bluesky_proxy BlueSkyProxy

Core proxy to connect.

 required
host str | None

BlueSky server host. Defaults to the proxy's server_ip, then the BLUESKY_SERVER_HOST environment variable, then "localhost".

 None
ready_timeout float

Maximum seconds to wait for a BlueSky port to start listening.

 60.0
poll_interval float

Seconds to sleep between port probes.

 0.5
is_port_listening Callable | None

Port probe taking (port, timeout, host). Defaults to WebATM.server.bluesky_server_status.is_port_listening.

 None
register_subscribers Callable | None

Subscriber-registration hook. Defaults to WebATM.proxy.register_subscribers.

 None
sleep Callable | None

Sleep function. Defaults to time.sleep.

 None

Returns:

Type Description
bool

True if the proxy connect succeeded, False if BlueSky never came up within the timeout or the connect attempt raised.
Source code in WebATM-integrated/webatm_integrated/auto_start.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def connect_proxy_when_ready(
    bluesky_proxy,
    *,
    host: str | None = None,
    ready_timeout: float = 60.0,
    poll_interval: float = 0.5,
    is_port_listening: Callable[..., bool] | None = None,
    register_subscribers: Callable[[], None] | None = None,
    sleep: Callable[[float], None] | None = None,
) -> bool:
    """Wait for BlueSky to accept connections, then connect the WebATM proxy.

    Polls BlueSky's command/data ports until one is listening (or the timeout
    elapses), then performs the same connect sequence as the manual route:
    ``start_client`` followed by ``register_subscribers`` (subscribers attach to
    the client created by ``start_client``).

    The port probe, subscriber registration and sleep are injectable so this can
    be unit-tested without a real BlueSky server or wall-clock delays.

    Args:
        bluesky_proxy (BlueSkyProxy): Core proxy to connect.
        host (str | None): BlueSky server host. Defaults to the proxy's
            ``server_ip``, then the ``BLUESKY_SERVER_HOST`` environment
            variable, then ``"localhost"``.
        ready_timeout (float): Maximum seconds to wait for a BlueSky port to
            start listening.
        poll_interval (float): Seconds to sleep between port probes.
        is_port_listening (Callable | None): Port probe taking
            ``(port, timeout, host)``. Defaults to
            ``WebATM.server.bluesky_server_status.is_port_listening``.
        register_subscribers (Callable | None): Subscriber-registration hook.
            Defaults to ``WebATM.proxy.register_subscribers``.
        sleep (Callable | None): Sleep function. Defaults to ``time.sleep``.

    Returns:
        bool: True if the proxy connect succeeded, False if BlueSky never came
            up within the timeout or the connect attempt raised.
    """
    # Deferred imports: keep this module light for unit tests and avoid pulling
    # the Flask/ZMQ-laden core packages unless we actually connect.
    if is_port_listening is None:
        from WebATM.server.bluesky_server_status import is_port_listening
    if register_subscribers is None:
        from WebATM.proxy import register_subscribers
    if sleep is None:
        sleep = time.sleep

    host = (
        host
        or getattr(bluesky_proxy, "server_ip", None)
        or os.environ.get("BLUESKY_SERVER_HOST", "localhost")
    )

    if not _wait_for_ports(
        host, ready_timeout, poll_interval, is_port_listening, sleep
    ):
        logger.error(
            f"Auto-start: BlueSky ports {_BLUESKY_PORTS} not listening on '{host}' "
            f"after {ready_timeout:.0f}s; proxy not connected"
        )
        return False

    try:
        bluesky_proxy.server_ip = host
        bluesky_proxy.start_client(hostname=host)
        # Subscribers can only be registered once the client exists, which
        # start_client creates -- this is the same ordering the manual
        # /api/server/config route relies on.
        register_subscribers()
        logger.info(f"Auto-start: WebATM proxy connected to BlueSky at '{host}'")
        return True
    except Exception as e:
        logger.error(f"Auto-start: failed to connect proxy to BlueSky: {e}")
        return False

webatm_integrated.process_manager

webatm_integrated.process_manager

BlueSky headless server process manager.

Owns the lifecycle of the bluesky --headless process tree: the headless server plus every node child process it spawns. On POSIX, BlueSky spawns node children as ordinary subprocesses that inherit the parent's stdout/stderr and live in the parent's process group, so:

  • a single merged pipe on the parent captures the server and all node-child output, already interleaved in order; and
  • launching the parent in its own session (start_new_session=True) lets us reap the entire tree with one os.killpg.

BlueSkyProcessManager 

BlueSkyProcessManager(
    on_line: Callable[[str], None] | None = None,
    on_exit: Callable[[int], None] | None = None,
    spawn: Callable | None = None,
    cmd: list[str] | None = None,
)

Thread-safe lifecycle manager for the bluesky --headless process tree.

Tracks only the parent process; signals (stop/kill) address the whole process group so node children are reaped together with the server.

Initialize the process manager.

Parameters:

Name Type Description Default
on_line Callable[[str], None] | None

Callback invoked with each output line of the process tree (newline stripped).

 None
on_exit Callable[[int], None] | None

Callback invoked with the return code when the server process exits.

 None
spawn Callable | None

Spawn primitive for the reader task (e.g. socketio.start_background_task); defaults to a plain daemon thread.

 None
cmd list[str] | None

Command to launch; defaults to ["bluesky", "--headless"].

 None
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    on_line: Callable[[str], None] | None = None,
    on_exit: Callable[[int], None] | None = None,
    spawn: Callable | None = None,
    cmd: list[str] | None = None,
):
    """Initialize the process manager.

    Args:
        on_line: Callback invoked with each output line of the process
            tree (newline stripped).
        on_exit: Callback invoked with the return code when the server
            process exits.
        spawn: Spawn primitive for the reader task (e.g.
            ``socketio.start_background_task``); defaults to a plain
            daemon thread.
        cmd (list[str] | None): Command to launch; defaults to
            ``["bluesky", "--headless"]``.
    """
    self._lock = threading.RLock()
    self._proc: subprocess.Popen | None = None
    self._on_line = on_line
    self._on_exit = on_exit
    self._spawn = spawn or _default_spawn
    self._cmd = cmd or ["bluesky", "--headless"]
    self._state = "stopped"  # stopped | starting | running | stopping

start 

start() -> dict

Spawn the headless server (in its own process group) and a reader.

The process is started in a new session with merged, line-buffered stdout/stderr so the reader receives one ordered stream for the server and all node children. A no-op if the server is already running. If a concurrent :meth:stop is mid-shutdown, waits for it to finish and then starts a fresh server instead of reporting the doomed process as "already running".

Returns:

Type Description
dict

Result with success, status, pid and message (plus error on failure).
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def start(self) -> dict:
    """Spawn the headless server (in its own process group) and a reader.

    The process is started in a new session with merged, line-buffered
    stdout/stderr so the reader receives one ordered stream for the
    server and all node children. A no-op if the server is already
    running. If a concurrent :meth:`stop` is mid-shutdown, waits for it
    to finish and then starts a fresh server instead of reporting the
    doomed process as "already running".

    Returns:
        dict: Result with ``success``, ``status``, ``pid`` and
            ``message`` (plus ``error`` on failure).
    """
    with self._lock:
        stopping_proc = self._proc if self._state == "stopping" else None
    if stopping_proc is not None:
        # A concurrent stop() owns this process's shutdown; treating it as
        # "already running" would return a pid that is about to die. Wait
        # out the stop (its worst case is escalate_after + the 5s SIGKILL
        # wait), then start fresh below.
        try:
            stopping_proc.wait(timeout=15)
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server is still stopping; retry shortly",
            }

    with self._lock:
        if self._proc is not None and self._proc.poll() is None:
            return {
                "success": True,
                "status": "running",
                "pid": self._proc.pid,
                "message": "BlueSky server already running",
            }
        self._state = "starting"
        # PYTHONUNBUFFERED keeps the server's (and inherited children's)
        # stdout line-buffered so log lines arrive promptly and in order.
        env = dict(os.environ, PYTHONUNBUFFERED="1")
        try:
            self._proc = subprocess.Popen(
                self._cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,  # merge -> single ordered stream
                stdin=subprocess.DEVNULL,
                bufsize=1,
                text=True,
                env=env,
                start_new_session=True,  # own process group / session
                close_fds=True,
            )
        except Exception as e:
            self._state = "stopped"
            return {
                "success": False,
                "status": "error",
                "message": f"Failed to start BlueSky: {e}",
                "error": str(e),
            }
        self._state = "running"
        proc = self._proc

    # Launch the reader outside the lock.
    self._spawn(self._read_loop, proc)
    return {
        "success": True,
        "status": "running",
        "pid": proc.pid,
        "message": "BlueSky server started",
    }

stop 

stop(
    sig: int = signal.SIGTERM, escalate_after: float = 5.0
) -> dict

Signal the whole process group, escalating to SIGKILL if needed.

Parameters:

Name Type Description Default
sig int

Signal sent to the process group first.

 SIGTERM
escalate_after float

Seconds to wait for exit before force-killing the group with SIGKILL.

 5.0

Returns:

Type Description
dict

Result with success, status and message. success is False if the process survives even SIGKILL.
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def stop(self, sig: int = signal.SIGTERM, escalate_after: float = 5.0) -> dict:
    """Signal the whole process group, escalating to SIGKILL if needed.

    Args:
        sig (int): Signal sent to the process group first.
        escalate_after (float): Seconds to wait for exit before
            force-killing the group with SIGKILL.

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
            ``success`` is False if the process survives even SIGKILL.
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server is not running",
            }
        self._state = "stopping"
        try:
            pgid = os.getpgid(proc.pid)
        except ProcessLookupError:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server already exited",
            }

    # Signal the whole group (server + all node children) outside the lock.
    try:
        os.killpg(pgid, sig)
    except ProcessLookupError:
        pass

    try:
        proc.wait(timeout=escalate_after)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(pgid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            logger.error("BlueSky process group %s survived SIGKILL", pgid)
            with self._lock:
                if self._proc is proc:
                    self._state = "running"
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server did not exit after SIGKILL",
            }

    with self._lock:
        if self._proc is proc:
            self._state = "stopped"
    return {
        "success": True,
        "status": "stopped",
        "message": "BlueSky server stopped",
    }

kill 

kill() -> dict

Force-kill the whole process group immediately (no graceful wait).

Returns:

Type Description
dict

Result with success, status and message.
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
237
238
239
240
241
242
243
244
245
246
def kill(self) -> dict:
    """Force-kill the whole process group immediately (no graceful wait).

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
    """
    result = self.stop(sig=signal.SIGKILL, escalate_after=2.0)
    if result.get("success") and result.get("status") == "stopped":
        result["message"] = "BlueSky server killed"
    return result

restart 

restart() -> dict

Stop the current tree (if any) and start a fresh one.

Returns:

Type Description
dict

The start() result, with the message adjusted to "restarted" on success. If the old tree could not be stopped, its failure result is returned instead of starting a new one (start() would just report the surviving process as "already running").
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def restart(self) -> dict:
    """Stop the current tree (if any) and start a fresh one.

    Returns:
        dict: The ``start()`` result, with the message adjusted to
            "restarted" on success. If the old tree could not be stopped,
            its failure result is returned instead of starting a new one
            (``start()`` would just report the surviving process as
            "already running").
    """
    stop_result = self.stop()
    if not stop_result.get("success"):
        return stop_result
    result = self.start()
    if result.get("success"):
        result["message"] = "BlueSky server restarted"
    return result

status 

status() -> dict

Report whether the server is running, with its pid and state.

Returns:

Type Description
dict

Result with success, running, status and pid (None when stopped).
Source code in WebATM-integrated/webatm_integrated/process_manager.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def status(self) -> dict:
    """Report whether the server is running, with its pid and state.

    Returns:
        dict: Result with ``success``, ``running``, ``status`` and
            ``pid`` (None when stopped).
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            return {
                "success": True,
                "running": False,
                "status": "stopped",
                "pid": None,
            }
        return {
            "success": True,
            "running": True,
            "status": self._state,
            "pid": proc.pid,
        }

webatm_integrated.log_streamer

webatm_integrated.log_streamer

Live, in-order log streaming of the BlueSky process tree to web clients.

A single server-wide stream (one subprocess) is broadcast to all connected browsers over the server_log Socket.IO event. Ordering is guaranteed by a monotonic sequence number assigned under a lock at ingest, before any async hop. Bursts (for example, creating many nodes at once) are coalesced into batches so a flood of lines cannot overwhelm Socket.IO.

LogStreamer 

LogStreamer(
    socketio,
    max_history: int = 2000,
    batch_ms: int = 100,
    batch_max: int = 200,
)

Buffer process output and broadcast it as ordered, batched events.

Initialize the streamer.

Parameters:

Name Type Description Default
socketio SocketIO

Instance used to emit batches.

 required
max_history int

Maximum lines retained for history replay.

 2000
batch_ms int

Delay in milliseconds used to coalesce a batch.

 100
batch_max int

Maximum lines per emitted batch chunk.

 200
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    socketio,
    max_history: int = 2000,
    batch_ms: int = 100,
    batch_max: int = 200,
):
    """Initialize the streamer.

    Args:
        socketio (flask_socketio.SocketIO): Instance used to emit batches.
        max_history (int): Maximum lines retained for history replay.
        batch_ms (int): Delay in milliseconds used to coalesce a batch.
        batch_max (int): Maximum lines per emitted batch chunk.
    """
    self._sio = socketio
    self._lock = threading.Lock()
    self._history: collections.deque[dict] = collections.deque(maxlen=max_history)
    self._pending: list[dict] = []
    self._seq = 0
    self._flush_scheduled = False
    self._batch_ms = batch_ms
    self._batch_max = batch_max

feed_line 

feed_line(line: str) -> None

Ingest one output line, assign its order, and schedule a flush.

Parameters:

Name Type Description Default
line str

The process output line to broadcast.

 required
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def feed_line(self, line: str) -> None:
    """Ingest one output line, assign its order, and schedule a flush.

    Args:
        line (str): The process output line to broadcast.
    """
    with self._lock:
        self._seq += 1
        item = {"seq": self._seq, "t": time.time(), "line": line}
        self._history.append(item)
        self._pending.append(item)
        if not self._flush_scheduled:
            self._flush_scheduled = True
            try:
                self._sio.start_background_task(self._flush_after_delay)
            except Exception:
                # Un-wedge the scheduler: the line stays pending and the
                # next feed_line retries; a stuck True flag would silence
                # the stream forever.
                self._flush_scheduled = False
                raise

history 

history() -> list[dict]

Return a snapshot of buffered lines for late-joining clients.

Returns:

Type Description
list[dict]

Buffered items with seq, t and line keys.
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
79
80
81
82
83
84
85
86
def history(self) -> list[dict]:
    """Return a snapshot of buffered lines for late-joining clients.

    Returns:
        list[dict]: Buffered items with ``seq``, ``t`` and ``line`` keys.
    """
    with self._lock:
        return list(self._history)

on_process_exit 

on_process_exit(return_code: int) -> None

Emit an end-of-stream marker when the server process exits.

Parameters:

Name Type Description Default
return_code int

Exit code of the BlueSky server process.

 required
Source code in WebATM-integrated/webatm_integrated/log_streamer.py 
88
89
90
91
92
93
94
def on_process_exit(self, return_code: int) -> None:
    """Emit an end-of-stream marker when the server process exits.

    Args:
        return_code (int): Exit code of the BlueSky server process.
    """
    self.feed_line(f"--- bluesky server exited (return code {return_code}) ---")

webatm_integrated.bluesky_paths

webatm_integrated.bluesky_paths

BlueSky file-management path wiring for the integrated build.

In the integrated variant BlueSky runs inside the same container as the WebATM backend, so its scenario / plugins / output directories live on the same filesystem and at a fixed, known location -- there is nothing for a user to configure. This module resolves that location and points WebATM's existing file-management routes (/api/bluesky/...) straight at it, replacing the manual "configure base path" step the standalone build relies on.

BlueSky, installed as a pip package (which is exactly how the integrated image ships it), keeps its working directory at ~/bluesky and maintains scenario, plugins and output subdirectories under it (see BlueSky's pathfinder). Pointing the file manager at that same working directory means uploads and browsing land precisely where the running server reads them, with no chance of the two disagreeing.

The core webatm package never imports this module; it is reached only via webatm_integrated.register (env-guarded on WEBATM_INTEGRATED=1).

resolve_bluesky_workdir 

resolve_bluesky_workdir() -> Path

Return BlueSky's working directory (where scenario/plugins/output live).

For a pip-package install -- the only way the integrated build ships BlueSky -- BlueSky uses ~/bluesky as its working directory. We deliberately mirror that exact rule (rather than expose a separate, overridable setting) so WebATM and the BlueSky server can never point at different directories.

Returns:

Type Description
Path

BlueSky's working directory (~/bluesky).
Source code in WebATM-integrated/webatm_integrated/bluesky_paths.py 
35
36
37
38
39
40
41
42
43
44
45
46
def resolve_bluesky_workdir() -> Path:
    """Return BlueSky's working directory (where scenario/plugins/output live).

    For a pip-package install -- the only way the integrated build ships BlueSky
    -- BlueSky uses ``~/bluesky`` as its working directory. We deliberately
    mirror that exact rule (rather than expose a separate, overridable setting)
    so WebATM and the BlueSky server can never point at different directories.

    Returns:
        Path: BlueSky's working directory (``~/bluesky``).
    """
    return Path.home() / "bluesky"

configure_file_management 

configure_file_management(app) -> str

Pre-configure WebATM's file-management routes for the integrated build.

Sets app.bluesky_base_path -- the very same attribute the standalone build's /api/bluesky/configure-base-path route sets -- so every existing file route (filestatus, upload, browse, list, delete, output) keeps working unchanged, just pre-wired to BlueSky's working directory. The managed subdirectories are best-effort created so the UI can browse and upload even before the BlueSky server's first start.

Parameters:

Name Type Description Default
app Flask

Flask application instance.

 required

Returns:

Type Description
str

The configured base path (BlueSky's working directory).
Source code in WebATM-integrated/webatm_integrated/bluesky_paths.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def configure_file_management(app) -> str:
    """Pre-configure WebATM's file-management routes for the integrated build.

    Sets ``app.bluesky_base_path`` -- the very same attribute the standalone
    build's ``/api/bluesky/configure-base-path`` route sets -- so every existing
    file route (filestatus, upload, browse, list, delete, output) keeps working
    unchanged, just pre-wired to BlueSky's working directory. The managed
    subdirectories are best-effort created so the UI can browse and upload even
    before the BlueSky server's first start.

    Args:
        app (flask.Flask): Flask application instance.

    Returns:
        str: The configured base path (BlueSky's working directory).
    """
    workdir = resolve_bluesky_workdir()
    base_path = str(workdir)
    app.bluesky_base_path = base_path

    try:
        workdir.mkdir(parents=True, exist_ok=True)
        for subdir in _MANAGED_SUBDIRS:
            (workdir / subdir).mkdir(exist_ok=True)
    except OSError as e:
        # Non-fatal: BlueSky itself (re)creates these on its first start, and the
        # file routes degrade gracefully when a directory is missing. We still
        # keep base_path set so the UI reports the correct, fixed location.
        logger.warning(
            f"Could not pre-create BlueSky file directories under {workdir}: {e}"
        )

    logger.info(f"Integrated: BlueSky file management configured at {base_path}")
    return base_path

webatm_integrated.routes

webatm_integrated.routes

REST control routes for the integrated BlueSky server.

Namespaced under /api/integrated/ so they cannot collide with core routes and are simply absent from the default build.

register_integrated_routes 

register_integrated_routes(app, manager)

Register server lifecycle-control routes on the Flask app.

Parameters:

Name Type Description Default
app Flask

Flask application instance.

 required
manager BlueSkyProcessManager

Controls the bundled server.

 required
Source code in WebATM-integrated/webatm_integrated/routes.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def register_integrated_routes(app, manager):
    """Register server lifecycle-control routes on the Flask app.

    Args:
        app (flask.Flask): Flask application instance.
        manager (BlueSkyProcessManager): Controls the bundled server.
    """

    @app.route("/api/integrated/server/start", methods=["POST"])
    def integrated_server_start():
        """Start the bundled BlueSky server (POST /api/integrated/server/start)."""
        logger.info("Integrated: start BlueSky server requested")
        return jsonify(manager.start())

    @app.route("/api/integrated/server/stop", methods=["POST"])
    def integrated_server_stop():
        """Stop the bundled BlueSky server (POST /api/integrated/server/stop)."""
        logger.info("Integrated: stop BlueSky server requested")
        return jsonify(manager.stop())

    @app.route("/api/integrated/server/restart", methods=["POST"])
    def integrated_server_restart():
        """Restart the bundled BlueSky server (POST /api/integrated/server/restart)."""
        logger.info("Integrated: restart BlueSky server requested")
        return jsonify(manager.restart())

    @app.route("/api/integrated/server/kill", methods=["POST"])
    def integrated_server_kill():
        """Kill the whole BlueSky process group (POST /api/integrated/server/kill)."""
        logger.info("Integrated: kill BlueSky server requested")
        return jsonify(manager.kill())

    @app.route("/api/integrated/server/status", methods=["GET"])
    def integrated_server_status():
        """Report the server process state (GET /api/integrated/server/status)."""
        return jsonify(manager.status())

webatm_integrated.socket_handlers

webatm_integrated.socket_handlers

Socket.IO handlers for the integrated build.

Provides live-log history replay for late-joining clients. The lifecycle actions themselves are REST-only (see :mod:.routes).

register_integrated_socket_handlers 

register_integrated_socket_handlers(socketio, streamer)

Register the integrated build's Socket.IO event handlers.

Parameters:

Name Type Description Default
socketio SocketIO

The Flask-SocketIO instance.

 required
streamer LogStreamer

Holds the server-log history.

 required
Source code in WebATM-integrated/webatm_integrated/socket_handlers.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def register_integrated_socket_handlers(socketio, streamer):
    """Register the integrated build's Socket.IO event handlers.

    Args:
        socketio (flask_socketio.SocketIO): The Flask-SocketIO instance.
        streamer (LogStreamer): Holds the server-log history.
    """

    @socketio.on("request_log_history")
    def on_request_log_history():
        """Replay buffered server-log lines (``request_log_history`` event).

        Emits the log history to the requesting client only (flask_socketio's
        ``emit`` defaults to the sender's room), marked with ``replay: True``;
        clients de-duplicate by ``seq``.
        """
        emit(EVENT, {"lines": streamer.history(), "replay": True})