Glassvision/IfcOpenShell
1
0
Fork 0
mirror of https://github.com/IfcOpenShell/IfcOpenShell.git synced 2026-08-09 09:21:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

add type hints and docstrings

Browse Source
This commit is contained in:
Ziad-I
2024-07-21 14:31:18 +03:00
committed by Dion Moult
parent 276c4b871e
commit 975b215bd0
1 changed files with 105 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/blenderbim/blenderbim/tool/web.py
+105 -25
View File
@@ -49,7 +49,16 @@ IFC_TASK_ATTRIBUTE_MAP = {
class Web(blenderbim.core.tool.Web):
@classmethod
def generate_port_number(cls):
def generate_port_number(cls) -> int:
"""
Generate a free port number.
This method creates a temporary socket to bind to a free port.
It then retrieves the port number, and returns it.
Returns:
int: The port number that was generated.
"""
print("Generating port number")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("localhost", 0)) # Bind to a free port
@@ -58,16 +67,38 @@ class Web(blenderbim.core.tool.Web):
return port
@classmethod
def is_port_available(cls, port):
def is_port_available(cls, port: int) -> bool:
"""
Attempts to connect to the specified port on localhost.
If the connection is refused, the port is available for use; otherwise, it is in use.
Args:
port (int): The port number to check.
Returns:
bool: True if the port is available, False if it is in use.
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
# connect_ex returns errno.SUCCESS (0) if the connection succeeds
# indicating the port is in use
# otherwise returns errno.ECONNREFUSED (111 or 10061) if no server is listening
# indicating the port is available for use
return s.connect_ex(("localhost", port)) == errno.ECONNREFUSED
@classmethod
def start_websocket_server(cls, port):
def start_websocket_server(cls, port: int) -> None:
"""
Starts a WebSocket server on the specified port.
This method sets up the environment, locates paths, and starts
the WebSocket server process. It also handles the creation and updating of a PID file to keep track
of running server instances.
Args:
port (int): The port number on which to start the WebSocket server.
Returns:
None
"""
import addon_utils
global ws_process
@@ -82,13 +113,11 @@ class Web(blenderbim.core.tool.Web):
env["blenderbim_path"] = blenderbim_path
ws_process = subprocess.Popen(
[sys.executable, ws_path, "--p", str(port), "--host", "127.0.0.1"], cwd=webui_path, env=env
[sys.executable, ws_path, "--p", str(port), "--host", "127.0.0.1"],
cwd=webui_path,
env=env,
)
# moved intial opening of web browser to server on startup event
# to handle race condition between starting server and openning browser
# cls.open_web_browser(port)
pid_file = os.path.join(webui_path, "running_pid.json")
if os.path.exists(pid_file):
@@ -105,7 +134,19 @@ class Web(blenderbim.core.tool.Web):
cls.set_is_running(True)
@classmethod
def connect_websocket_server(cls, port):
def connect_websocket_server(cls, port: int) -> None:
"""
Connect to a WebSocket server on the specified port.
This method sets up an asynchronous Socket.IO client with
reconnection attempts, starts an asyncio thread, connects to the WebSocket server, and sets the connection status.
Args:
port (int): The port number to connect to the WebSocket server.
Returns:
None
"""
global ws_thread, sio
if bpy.context.scene.WebProperties.is_connected:
@@ -124,7 +165,13 @@ class Web(blenderbim.core.tool.Web):
bpy.app.timers.register(cls.check_operator_queue)
@classmethod
def disconnect_websocket_server(cls):
def disconnect_websocket_server(cls) -> None:
"""
Disconnects the WebSocket server and stops the associated thread.
This method is responsible for disconnecting the WebSocket server, stopping the asyncio thread,
and resetting the global variables related to the WebSocket connection.
"""
global ws_thread, sio
ws_thread.run_coro(cls.sio_disconnect())
ws_thread.stop()
@@ -133,7 +180,16 @@ class Web(blenderbim.core.tool.Web):
cls.set_is_connected(False)
@classmethod
def kill_websocket_server(cls):
def kill_websocket_server(cls) -> None:
"""
Terminate the currently running WebSocket server.
This method checks if there is an active WebSocket server process. If so, it disconnects it (if connected),
removes its PID from the PID file, terminates the process, and updates the server's running status.
Returns:
None
"""
global ws_process
if ws_process is None:
@@ -198,7 +254,7 @@ class Web(blenderbim.core.tool.Web):
ws_thread.run_coro(cls.sio_send(payload, event, namespace))
@classmethod
def check_operator_queue(cls):
def check_operator_queue(cls) -> None | float:
if not bpy.context.scene.WebProperties.is_connected:
with web_operator_queue.mutex:
web_operator_queue.queue.clear()
@@ -215,7 +271,7 @@ class Web(blenderbim.core.tool.Web):
return 1.0
@classmethod
def handle_csv_operator(cls, operator_data):
def handle_csv_operator(cls, operator_data: dict) -> None:
if operator_data["type"] == "selection":
bpy.ops.object.select_all(action="DESELECT")
guid = operator_data["globalId"]
@@ -225,7 +281,7 @@ class Web(blenderbim.core.tool.Web):
obj.select_set(True)
@classmethod
def handle_gantt_operator(cls, operator_data):
def handle_gantt_operator(cls, operator_data: dict) -> None:
ifc_file = tool.Ifc.get()
if operator_data["type"] == "editTask":
task_id = int(operator_data["taskId"])
@@ -257,52 +313,76 @@ class Web(blenderbim.core.tool.Web):
cls.send_webui_data(data=gantt_data, data_key="gantt_data", event="gantt_data")
@classmethod
def open_web_browser(cls, port):
def open_web_browser(cls, port: int) -> None:
webbrowser.open(f"http://127.0.0.1:{port}/")
@classmethod
async def sio_connect(cls, url):
async def sio_connect(cls, url: str) -> None:
await sio.connect(url, transports=["websocket"], namespaces="/blender")
sio.on("web_operator", cls.sio_listen_web_operator, namespace="/blender")
@classmethod
async def sio_disconnect(cls):
async def sio_disconnect(cls) -> None:
await sio.disconnect()
@classmethod
async def sio_send(cls, data, event="data", namespace="/blender"):
async def sio_send(cls, data: Any, event: str = "data", namespace: str = "/blender") -> None:
await sio.emit(event, data, namespace=namespace)
@classmethod
async def sio_listen_web_operator(cls, data):
async def sio_listen_web_operator(cls, data: dict) -> None:
try:
web_operator_queue.put_nowait(data)
except queue.Full:
pass
@classmethod
def set_is_running(cls, is_running):
def set_is_running(cls, is_running: bool) -> None:
bpy.context.scene.WebProperties.is_running = is_running
@classmethod
def set_is_connected(cls, is_connected):
def set_is_connected(cls, is_connected: bool) -> None:
bpy.context.scene.WebProperties.is_connected = is_connected
class AsyncioThread(threading.Thread):
"""
A thread that runs an asyncio event loop.
Args:
*args: Variable length argument list.
loop: An existing asyncio event loop. If None, a new event loop is created.
**kwargs: Arbitrary keyword arguments.
"""
def __init__(self, *args, loop=None, **kwargs):
super().__init__(*args, **kwargs)
self.loop = loop or asyncio.new_event_loop()
self.running = False
def run(self):
"""
Start the asyncio event loop and mark the thread as running.
"""
self.running = True
self.loop.run_forever()
def run_coro(self, coro):
def run_coro(self, coro) -> Any:
"""
Run a coroutine in the asyncio event loop from a separate thread.
Args:
coro: The coroutine to be run.
Returns:
The result of the coroutine.
"""
return asyncio.run_coroutine_threadsafe(coro, loop=self.loop).result()
def stop(self):
def stop(self) -> None:
"""
Stop the asyncio event loop and join the thread.
"""
self.loop.call_soon_threadsafe(self.loop.stop)
self.join()
self.running = False