Scribd
Upload
20 views7 pages

Subprocess Management in Python Scripts

The document outlines a Python script that manages subprocesses for visualization and stimuli handling using PySide6. It includes classes for running subprocesses, logging messages from Brython, and managing visualizations with auto-resizing features. The script also handles loading Python scripts, checking their types, and managing web views for displaying outputs.

Uploaded by

YeisonEng
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
20 views7 pages

Subprocess Management in Python Scripts

The document outlines a Python script that manages subprocesses for visualization and stimuli handling using PySide6. It includes classes for running subprocesses, logging messages from Brython, and managing visualizations with auto-resizing features. The script also handles loading Python scripts, checking their types, and managing web views for displaying outputs.

Uploaded by

YeisonEng
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

"""

==================
Subprocess handler
==================
"""

import os
import sys
import socket
import logging
import subprocess
import [Link]
from urllib import request
from queue import Queue, Empty
from contextlib import closing
from typing import TypeVar, Optional

from [Link] import QApplication


from [Link] import QTimer, QSize
from [Link] import QWebEngineView
from [Link] import QWebEnginePage

from ..extensions import properties as prop


from .nbstreamreader import NonBlockingStreamReader as NBSR

PathLike = TypeVar('PathLike')
HostLike = TypeVar('HostLike')
Command = TypeVar('Command')

DEFAULT_LOCAL_IP = 'localhost'

# ----------------------------------------------------------------------
def run_subprocess(call: Command) -> [Link]:
"""Run a python script with non blocking debugger installed."""
my_env = [Link]()
my_env['PYTHONPATH'] = ":".join(
[Link] + [[Link]([Link]([Link][0]))])

sub = [Link](call,
stdout=[Link],
stderr=[Link],
env=my_env,
preexec_fn=[Link],
shell=False,
# universal_newlines=True,
# bufsize=1,
)
sub.nb_stdout = NBSR([Link])

return sub

########################################################################
class BrythonLogging:
"""Log messages from Brython."""

# ----------------------------------------------------------------------
def __init__(self):
""""""
[Link] = Queue()

# ----------------------------------------------------------------------
def feed(self, level: int, message: str, lineNumber: int, sourceID: str) ->
None:
"""Concatenae messages."""
[Link](message)

# ----------------------------------------------------------------------
def readline(self, timeout: Optional[int] = None) -> str:
"""Get mesage from JavaScriptConsole."""
if [Link]():
try:
return [Link](block=timeout is not None, timeout=timeout)
except Empty:
return None

########################################################################
class VisualizationSubprocess:
"""Define matplotlib properties and start the auto-resizer."""

# ----------------------------------------------------------------------
def viz_auto_size(self, timer: bool = True) -> None:
""""""
if [Link]:
return

dpi = [Link]
f = dpi / [Link]
try:
size = [Link].web_engine.size()

# reload if changes
if self.plot_size != size or self.plot_dpi != [Link] or
self.input_interact():
if 'light' in [Link]('QTMATERIAL_THEME'):
background = 'ffffff'
else:
background = '000000'

url = [Link] + \
f'/?width={f * [Link]() / dpi:.2f}&height={f *
[Link]() /
dpi:.2f}&dpi={dpi/f:.2f}&background={background}&{self.update_interact()}'
[Link].web_engine.setUrl(url)
self.plot_dpi = [Link]
self.plot_size = size
except:
pass

if timer:
[Link](1000, self.viz_auto_size)

# ----------------------------------------------------------------------
def input_interact(self) -> bool:
""""""
if not hasattr(self, 'interactive_copy'):
self.interactive_copy = [Link]()
return False
differents_items = {
k: self.interactive_copy[k] for k in self.interactive_copy if k in
[Link] and self.interactive_copy[k] != [Link][k]}

self.interactive_copy = [Link]()
return bool(len(differents_items))

# ----------------------------------------------------------------------
def update_interact(self) -> str:
""""""
get_args = [
f'{k}={[Link][k]}' for k in [Link]]
return '&'.join(get_args)

# ----------------------------------------------------------------------
def viz_debug(self) -> None:
""""""
[Link] = self.subprocess_script.nb_stdout

# ----------------------------------------------------------------------
def viz_start(self) -> None:
""""""
[Link](1000, self.viz_auto_size)

########################################################################
class StimuliSubprocess:
"""Connect with Brython logs."""

# ----------------------------------------------------------------------
def stm_debug(self) -> None:
""""""
console = BrythonLogging()
self.web_engine_page = QWebEnginePage([Link].web_engine)
self.web_engine_page.javaScriptConsoleMessage = [Link]
[Link].web_engine.setPage(self.web_engine_page)
[Link] = console
self.web_engine_page .profile().clearHttpCache()
self.stm_start()

# ----------------------------------------------------------------------
def stm_start(self) -> None:
""""""
[Link].web_engine.setUrl([Link])
with open([Link]([Link]('BCISTREAM_HOME'), '[Link]'), 'w') as
file:
[Link]([Link](
'/dashboard', '').replace('localhost',
self.get_local_ip_address()))

# ----------------------------------------------------------------------
def get_local_ip_address(self) -> HostLike:
"""Connect to internet for get the local IP."""

try:
s = [Link](socket.AF_INET, socket.SOCK_DGRAM)
[Link](("[Link]", 80))
local_ip_address = [Link]()[0]
[Link]()
return local_ip_address

except:
[Link]('Impossible to detect a network connection, the WiFi'
'module and this machine must share the same network.')
[Link](f'If you are using this machine as server (access
point) '
f'the address {DEFAULT_LOCAL_IP} will be used.')

return DEFAULT_LOCAL_IP

########################################################################
class LoadSubprocess(VisualizationSubprocess, StimuliSubprocess):
""""""

# ----------------------------------------------------------------------
def __init__(self, parent, path: Optional[PathLike] = None, use_webview:
Optional[bool] = True, debugger: Optional[bool] = False):
""""""
[Link] = parent
self.web_view = [Link].gridLayout_webview
self.plot_size = QSize(0, 0)
self.plot_dpi = 0
[Link] = False
[Link] = debugger

if path:
self.load_path(path)

# ----------------------------------------------------------------------
def load_path(self, path: PathLike) -> None:
"""Load Python scipt."""
[Link] = QTimer()

self.is_analysis = self.file_is_analysis(path)
self.is_visualization = self.file_is_visualization(path)
self.is_timelock = self.file_is_timelock(path)
self.is_stimuli = self.file_is_stimuli(path)

if any([self.is_stimuli, self.is_visualization]):
[Link] = self.get_free_port()
else:
[Link] = ''

if [Link]:
extra = '--debug'
else:
extra = ''

if not self.is_timelock:
self.subprocess_script = run_subprocess(
[[Link], path, [Link], extra])

if any([self.is_visualization, self.is_stimuli]):
self.prepare_webview()
elif self.is_timelock:
self.prepare_layout(path)

# ----------------------------------------------------------------------
def file_is_analysis(self, path: PathLike) -> bool:
""""""
with open(path, 'r') as file:
return 'data_analysis import DataAnalysis' in [Link]()

# ----------------------------------------------------------------------
def file_is_stimuli(self, path: PathLike) -> bool:
""""""
with open(path, 'r') as file:
return 'stimuli_delivery import StimuliAPI' in [Link]()

# ----------------------------------------------------------------------
def file_is_visualization(self, path: PathLike) -> bool:
""""""
with open(path, 'r') as file:
return 'visualizations import EEGStream' in [Link]()

# ----------------------------------------------------------------------
def file_is_timelock(self, path: PathLike) -> bool:
""""""
with open(path, 'r') as file:
return 'timelock_analysis import TimelockDashboard' in [Link]() or
'timelock_analysis import TimelockWidget' in [Link]()

# ----------------------------------------------------------------------
def prepare_webview(self) -> None:
"""Try to load the webview."""
# Try to get mode
try:
[Link] = [Link](
f'[Link] timeout=10).read().decode()
except: # if fail
[Link](100, self.prepare_webview) # call again
return

# and only when the mode is explicit...

if [Link] == 'visualization':
self.is_visualization = True
self.is_stimuli = False
self.is_analysis = False
endpoint = ''
elif [Link] == 'stimuli':
self.is_visualization = False
self.is_stimuli = True
self.is_analysis = False
endpoint = 'dashboard'

# [Link].widget_development_webview.show()
[Link] = f'[Link]
self.load_webview()

# ----------------------------------------------------------------------
def prepare_layout(self, path: PathLike) -> None:
""""""
spec = [Link].spec_from_file_location("Analysis", path)
foo = [Link].module_from_spec(spec)
[Link].exec_module(foo)
for i in range(self.web_view.count()):
self.web_view.itemAt(i).widget().deleteLater()

screen = [Link]()
size = [Link]()

[Link] = [Link]([Link]())
self.web_view.addWidget([Link])

# ----------------------------------------------------------------------
def clear_subprocess_script(self) -> None:
""""""
if hasattr(self, 'subprocess_script'):
self.__delattr__('subprocess_script')

# ----------------------------------------------------------------------
def stop_preview(self) -> None:
"""Kill the subprocess and crear the webview."""
[Link]()
[Link] = True
if hasattr(self, 'subprocess_script'):
self.subprocess_script.nb_stdout.stop()
self.subprocess_script.terminate()
if hasattr(self, 'subprocess_script'):
[Link](300, self.clear_subprocess_script)

# [Link](
# 300, lambda: delattr(self, 'subprocess_script'))

if hasattr(self, 'web_engine_page'):
try:
self.web_engine_page.deleteLater()
except: # already deleted.
pass

# TODO: A wait page could be a god idea


# [Link].widget_development_webview.hide()
if hasattr([Link], 'web_engine'):
[Link].web_engine.setUrl('about:blank')

# ----------------------------------------------------------------------
def load_webview(self) -> None:
"""After the process starting, set the URL into the webview."""
# [Link].widget_development_webview.show()

# Create main QWebEngineView object


if not hasattr([Link], 'web_engine'):
[Link].web_engine = QWebEngineView()
self.web_view.addWidget([Link].web_engine)

else:
# [Link].web_engine.deleteLater()
# [Link].web_engine = QWebEngineView()
self.web_view.addWidget([Link].web_engine)

# Set URL and start interface


if self.is_visualization:
self.viz_start()
elif self.is_stimuli:
self.stm_start()

# ----------------------------------------------------------------------
def start_debug(self) -> None:
"""Try to start the debugger."""
if [Link]:
try:
if self.is_stimuli:
self.stm_debug()
elif self.is_visualization or self.is_analysis:
self.viz_debug()
except Exception as e:
pass

# ----------------------------------------------------------------------
def reload(self) -> None:
"""Restart the webview."""
[Link].web_engine.setUrl([Link])

# ----------------------------------------------------------------------
def get_free_port(self) -> str:
"""Get any free port available."""
with closing([Link](socket.AF_INET, socket.SOCK_STREAM)) as s:
[Link](('', 0))
[Link](socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
port = str([Link]()[1])
[Link](f'Free port found in {port}')
return port

Common questions

Powered by AI

The LoadSubprocess class supports both stimuli and visualization functionalities by employing a design approach where it integrates the properties and methods from both VisualizationSubprocess and StimuliSubprocess classes. It distinguishes tasks by checking file content for specific imports and sets configuration accordingly. This implies a versatile and hierarchical design that allows each class to focus on a specific set of functionalities while maintaining straightforward integration. It effectively combines multiple aspects into a single cohesive framework, allowing it to manage diverse task types efficiently .

Exception handling strategies in the discussed classes contribute to system robustness by providing mechanisms to anticipate potential failures and ensure continuity of operations. For instance, network-related methods like get_local_ip_address capture exceptions when an operation can't be completed, defaulting to safe values instead. Additionally, during webview preparation or subprocess management, the classes handle unforeseen exceptions silently to prevent crashes, allowing the timer to retry operations or clear states as needed. This approach minimizes disruptions, allowing for graceful degradation and fault tolerance .

The LoadSubprocess class enhances modularity and extensibility by combining functionality from multiple subprocess management classes, such as VisualizationSubprocess and StimuliSubprocess, allowing for specialized processing tasks like visualization and stimuli delivery. It uses methods like load_path and prepare_webview to differentiate tasks and configure the environment and execution dynamically. By encapsulating subprocess handling, debugging, and environmental setup within a class structure, it can easily adapt to new types of tasks or integrate additional features without disrupting existing functionality .

The timer in the LoadSubprocess class is pivotal for asynchronous task management as it schedules periodic validation and updates required by the visualization tasks. For instance, the viz_auto_size method employs the timer to repeatedly adjust the plotting area dimensions and update the URL based on current settings. This ensures that visualization remains responsive to changes in size, DPI resolution, and input interactions without blocking the main thread, enhancing the efficiency of interactive graphics presentation .

The VisualizationSubprocess uses several mechanisms to ensure plots are correctly scaled and displayed. It checks if the actual plot dimensions or DPI settings differ from the expected values, then recalculates the size and associated parameters. The URL used to render the plot is updated with detailed query parameters specifying width, height, DPI, and background color to maintain visual fidelity. The viz_auto_size method, part of this process, ensures that these updates happen dynamically and periodically by utilizing a timer to trigger checks and adjustments .

The get_local_ip_address method defaults to using the predefined IP address 'localhost' when it cannot establish a network connection. This occurs if the method fails to connect to an external IP (like 8.8.8.8), which indicates that the device is either not connected to a network or is in a configuration where it acts as a server or access point without a recognized external IP .

The BrythonLogging class uses the feed method to handle messages sent from the JavaScript console. Its primary function is to concatenate messages into a queue for later retrieval with the readline method. This allows the system to log and process information for debugging purposes in a non-blocking manner .

The run_subprocess function is used to execute a Python script with a non-blocking debugger installed. It creates a subprocess using subprocess.Popen, configures the environment with an updated PYTHONPATH, and sets up non-blocking stdout reading using NonBlockingStreamReader. It is designed to handle the execution of scripts while allowing for output monitoring and debugging .

The LoadSubprocess class determines if a file is a visualization task by checking if the file contains the string 'visualizations import EEGStream'. This is done using the file_is_visualization method, which reads the file's content to identify specific import statements indicating its function .

When a visualization is detected, the LoadSubprocess class starts a webview by first determining the mode of operation (visualization or stimuli) using an HTTP request to check the mode endpoint. If visualization is confirmed, it sets relevant flags and constructs a URL for the local server hosting the visualization based on the acquired port and endpoint configuration. The webview component is set up using QWebEngineView, which is integrated into the main interface and linked to the URL. Finally, visualization is started through the viz_start method, enabling dynamic content rendering .

You might also like