We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Ignored exception on ci in windows related to monitor and socket. 
This may be related to the intermitted failures of monitor tests on windows

Observing the following traceback
```
Exception ignored in: <socket.socket fd=1116, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 55090), raddr=('127.0.0.1', 55091)>
Traceback (most recent call last):
  File "C:\hostedtoolcache\windows\Python\3.9.13\x64\lib\threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
ResourceWarning: unclosed <socket.socket fd=1116, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 55090), raddr=('127.0.0.1', 55091)>
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/qcodes/monitor/monitor.py`
Content:
```
1 #! /usr/bin/env python
2 # vim:fenc=utf-8
3 #
4 # Copyright © 2017 unga <giulioungaretti@me.com>
5 #
6 # Distributed under terms of the MIT license.
7 """
8 Monitor a set of parameters in a background thread
9 stream output over websocket
10 
11 To start monitor, run this file, or if qcodes is installed as a module:
12 
13 ``% python -m qcodes.monitor.monitor``
14 
15 Add parameters to monitor in your measurement by creating a new monitor with a
16 list of parameters to monitor:
17 
18 ``monitor = qcodes.Monitor(param1, param2, param3, ...)``
19 """
20 from __future__ import annotations
21 
22 import asyncio
23 import json
24 import logging
25 import os
26 import socketserver
27 import time
28 import webbrowser
29 from asyncio import CancelledError
30 from collections import defaultdict
31 from contextlib import suppress
32 from importlib.resources import as_file, files
33 from threading import Event, Thread
34 from typing import TYPE_CHECKING, Any, Callable
35 
36 import websockets
37 import websockets.exceptions
38 import websockets.server
39 
40 from qcodes.parameters import Parameter
41 
42 if TYPE_CHECKING:
43     from collections.abc import Awaitable, Sequence
44 
45 WEBSOCKET_PORT = 5678
46 SERVER_PORT = 3000
47 
48 log = logging.getLogger(__name__)
49 
50 
51 def _get_metadata(
52     *parameters: Parameter, use_root_instrument: bool = True
53 ) -> dict[str, Any]:
54     """
55     Return a dictionary that contains the parameter metadata grouped by the
56     instrument it belongs to.
57     """
58     metadata_timestamp = time.time()
59     # group metadata by instrument
60     metas: dict[Any, Any] = defaultdict(list)
61     for parameter in parameters:
62         # Get the latest value from the parameter,
63         # respecting the max_val_age parameter
64         meta: dict[str, float | str | None] = {}
65         meta["value"] = str(parameter.get_latest())
66         timestamp = parameter.get_latest.get_timestamp()
67         if timestamp is not None:
68             meta["ts"] = timestamp.timestamp()
69         else:
70             meta["ts"] = None
71         meta["name"] = parameter.label or parameter.name
72         meta["unit"] = parameter.unit
73 
74         # find the base instrument that this parameter belongs to
75         if use_root_instrument:
76             baseinst = parameter.root_instrument
77         else:
78             baseinst = parameter.instrument
79         if baseinst is None:
80             metas["Unbound Parameter"].append(meta)
81         else:
82             metas[str(baseinst)].append(meta)
83 
84     # Create list of parameters, grouped by instrument
85     parameters_out = []
86     for instrument in metas:
87         temp = {"instrument": instrument, "parameters": metas[instrument]}
88         parameters_out.append(temp)
89 
90     state = {"ts": metadata_timestamp, "parameters": parameters_out}
91     return state
92 
93 
94 def _handler(
95     parameters: Sequence[Parameter], interval: float, use_root_instrument: bool = True
96 ) -> Callable[[websockets.server.WebSocketServerProtocol, str], Awaitable[None]]:
97     """
98     Return the websockets server handler.
99     """
100 
101     async def server_func(
102         websocket: websockets.server.WebSocketServerProtocol, _: str
103     ) -> None:
104         """
105         Create a websockets handler that sends parameter values to a listener
106         every "interval" seconds.
107         """
108         while True:
109             try:
110                 # Update the parameter values
111                 try:
112                     meta = _get_metadata(
113                         *parameters, use_root_instrument=use_root_instrument
114                     )
115                 except ValueError:
116                     log.exception("Error getting parameters")
117                     break
118                 log.debug("sending.. to %r", websocket)
119                 await websocket.send(json.dumps(meta))
120                 # Wait for interval seconds and then send again
121                 await asyncio.sleep(interval)
122             except (CancelledError, websockets.exceptions.ConnectionClosed):
123                 log.debug("Got CancelledError or ConnectionClosed",
124                           exc_info=True)
125                 break
126         log.debug("Closing websockets connection")
127 
128     return server_func
129 
130 
131 class Monitor(Thread):
132     """
133     QCodes Monitor - WebSockets server to monitor qcodes parameters.
134     """
135 
136     running: Monitor | None = None
137 
138     def __init__(
139         self,
140         *parameters: Parameter,
141         interval: float = 1,
142         use_root_instrument: bool = True,
143     ):
144         """
145         Monitor qcodes parameters.
146 
147         Args:
148             *parameters: Parameters to monitor.
149             interval: How often one wants to refresh the values.
150             use_root_instrument: Defines if parameters are grouped according to
151                                 parameter.root_instrument or parameter.instrument
152         """
153         super().__init__(daemon=True)
154 
155         # Check that all values are valid parameters
156         for parameter in parameters:
157             if not isinstance(parameter, Parameter):
158                 raise TypeError(f"We can only monitor QCodes "
159                                 f"Parameters, not {type(parameter)}")
160 
161         self.loop: asyncio.AbstractEventLoop | None = None
162         self._stop_loop_future: asyncio.Future | None = None
163         self._parameters = parameters
164         self.loop_is_closed = Event()
165         self.server_is_started = Event()
166         self.handler = _handler(
167             parameters, interval=interval, use_root_instrument=use_root_instrument
168         )
169         log.debug("Start monitoring thread")
170         if Monitor.running:
171             # stop the old server
172             log.debug("Stopping and restarting server")
173             Monitor.running.stop()
174         self.start()
175 
176         # Wait until the loop is running
177         self.server_is_started.wait(timeout=5)
178         if not self.server_is_started.is_set():
179             raise RuntimeError("Failed to start server")
180         Monitor.running = self
181 
182     def run(self) -> None:
183         """
184         Start the event loop and run forever.
185         """
186         log.debug("Running Websocket server")
187 
188         async def run_loop() -> None:
189             self.loop = asyncio.get_running_loop()
190             self._stop_loop_future = self.loop.create_future()
191 
192             async with websockets.server.serve(
193                 self.handler, "127.0.0.1", WEBSOCKET_PORT, close_timeout=1
194             ):
195                 self.server_is_started.set()
196                 try:
197                     await self._stop_loop_future
198                 except asyncio.CancelledError:
199                     log.debug("Websocket server thread shutting down")
200 
201         try:
202             asyncio.run(run_loop())
203         finally:
204             self.loop_is_closed.set()
205 
206     def update_all(self) -> None:
207         """
208         Update all parameters in the monitor.
209         """
210         for parameter in self._parameters:
211             # call get if it can be called without arguments
212             with suppress(TypeError):
213                 parameter.get()
214 
215     def stop(self) -> None:
216         """
217         Shutdown the server, close the event loop and join the thread.
218         Setting active Monitor to ``None``.
219         """
220         self.join()
221         Monitor.running = None
222 
223     def join(self, timeout: float | None = None) -> None:
224         """
225         Overwrite ``Thread.join`` to make sure server is stopped before
226         joining avoiding a potential deadlock.
227         """
228         log.debug("Shutting down server")
229         if not self.is_alive():
230             # we run this check before trying to run to prevent a cryptic
231             # error message
232             log.debug("monitor is dead")
233             return
234         try:
235             if self.loop is not None and self._stop_loop_future is not None:
236                 log.debug("Instructing server to stop event loop.")
237                 self.loop.call_soon_threadsafe(self._stop_loop_future.cancel)
238             else:
239                 log.debug("No event loop found. Cannot stop event loop.")
240         except RuntimeError:
241             # the above may throw a runtime error if the loop is already
242             # stopped in which case there is nothing more to do
243             log.exception("Could not close loop")
244         self.loop_is_closed.wait(timeout=5)
245         if not self.loop_is_closed.is_set():
246             raise RuntimeError("Failed to join loop")
247         log.debug("Loop reported closed")
248         super().join(timeout=timeout)
249         log.debug("Monitor Thread has joined")
250 
251     @staticmethod
252     def show() -> None:
253         """
254         Overwrite this method to show/raise your monitor GUI
255         F.ex.
256 
257         ::
258 
259             import webbrowser
260             url = "localhost:3000"
261             # Open URL in new window, raising the window if possible.
262             webbrowser.open_new(url)
263 
264         """
265         webbrowser.open(f"http://localhost:{SERVER_PORT}")
266 
267 
268 def main() -> None:
269     import http.server
270 
271     # If this file is run, create a simple webserver that serves a simple
272     # website that can be used to view monitored parameters.
273     # # https://github.com/python/mypy/issues/4182
274     parent_module = ".".join(__loader__.name.split(".")[:-1])  # type: ignore[name-defined]
275 
276     static_dir = files(parent_module).joinpath("dist")
277     try:
278         with as_file(static_dir) as extracted_dir:
279             os.chdir(extracted_dir)
280             log.info("Starting HTTP Server at http://localhost:%i", SERVER_PORT)
281             with socketserver.TCPServer(
282                 ("", SERVER_PORT), http.server.SimpleHTTPRequestHandler
283             ) as httpd:
284                 log.debug("serving directory %s", static_dir)
285                 webbrowser.open(f"http://localhost:{SERVER_PORT}")
286                 httpd.serve_forever()
287     except KeyboardInterrupt:
288         log.info("Shutting Down HTTP Server")
289 
290 
291 if __name__ == "__main__":
292     main()
293
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/qcodes/monitor/monitor.py b/src/qcodes/monitor/monitor.py
--- a/src/qcodes/monitor/monitor.py
+++ b/src/qcodes/monitor/monitor.py
@@ -193,10 +193,8 @@
                 self.handler, "127.0.0.1", WEBSOCKET_PORT, close_timeout=1
             ):
                 self.server_is_started.set()
-                try:
-                    await self._stop_loop_future
-                except asyncio.CancelledError:
-                    log.debug("Websocket server thread shutting down")
+                await self._stop_loop_future
+                log.debug("Websocket server thread shutting down")
 
         try:
             asyncio.run(run_loop())
@@ -234,7 +232,7 @@
         try:
             if self.loop is not None and self._stop_loop_future is not None:
                 log.debug("Instructing server to stop event loop.")
-                self.loop.call_soon_threadsafe(self._stop_loop_future.cancel)
+                self.loop.call_soon_threadsafe(self._stop_loop_future.set_result, True)
             else:
                 log.debug("No event loop found. Cannot stop event loop.")
         except RuntimeError:

{"golden_diff": "diff --git a/src/qcodes/monitor/monitor.py b/src/qcodes/monitor/monitor.py\n--- a/src/qcodes/monitor/monitor.py\n+++ b/src/qcodes/monitor/monitor.py\n@@ -193,10 +193,8 @@\n                 self.handler, \"127.0.0.1\", WEBSOCKET_PORT, close_timeout=1\n             ):\n                 self.server_is_started.set()\n-                try:\n-                    await self._stop_loop_future\n-                except asyncio.CancelledError:\n-                    log.debug(\"Websocket server thread shutting down\")\n+                await self._stop_loop_future\n+                log.debug(\"Websocket server thread shutting down\")\n \n         try:\n             asyncio.run(run_loop())\n@@ -234,7 +232,7 @@\n         try:\n             if self.loop is not None and self._stop_loop_future is not None:\n                 log.debug(\"Instructing server to stop event loop.\")\n-                self.loop.call_soon_threadsafe(self._stop_loop_future.cancel)\n+                self.loop.call_soon_threadsafe(self._stop_loop_future.set_result, True)\n             else:\n                 log.debug(\"No event loop found. Cannot stop event loop.\")\n         except RuntimeError:\n", "issue": "Ignored exception on ci in windows related to monitor and socket. \nThis may be related to the intermitted failures of monitor tests on windows\r\n\r\nObserving the following traceback\r\n```\r\nException ignored in: <socket.socket fd=1116, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 55090), raddr=('127.0.0.1', 55091)>\r\nTraceback (most recent call last):\r\n  File \"C:\\hostedtoolcache\\windows\\Python\\3.9.13\\x64\\lib\\threading.py\", line 917, in run\r\n    self._target(*self._args, **self._kwargs)\r\nResourceWarning: unclosed <socket.socket fd=1116, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 55090), raddr=('127.0.0.1', 55091)>\r\n```\n", "before_files": [{"content": "#! /usr/bin/env python\n# vim:fenc=utf-8\n#\n# Copyright \u00a9 2017 unga <giulioungaretti@me.com>\n#\n# Distributed under terms of the MIT license.\n\"\"\"\nMonitor a set of parameters in a background thread\nstream output over websocket\n\nTo start monitor, run this file, or if qcodes is installed as a module:\n\n``% python -m qcodes.monitor.monitor``\n\nAdd parameters to monitor in your measurement by creating a new monitor with a\nlist of parameters to monitor:\n\n``monitor = qcodes.Monitor(param1, param2, param3, ...)``\n\"\"\"\nfrom __future__ import annotations\n\nimport asyncio\nimport json\nimport logging\nimport os\nimport socketserver\nimport time\nimport webbrowser\nfrom asyncio import CancelledError\nfrom collections import defaultdict\nfrom contextlib import suppress\nfrom importlib.resources import as_file, files\nfrom threading import Event, Thread\nfrom typing import TYPE_CHECKING, Any, Callable\n\nimport websockets\nimport websockets.exceptions\nimport websockets.server\n\nfrom qcodes.parameters import Parameter\n\nif TYPE_CHECKING:\n    from collections.abc import Awaitable, Sequence\n\nWEBSOCKET_PORT = 5678\nSERVER_PORT = 3000\n\nlog = logging.getLogger(__name__)\n\n\ndef _get_metadata(\n    *parameters: Parameter, use_root_instrument: bool = True\n) -> dict[str, Any]:\n    \"\"\"\n    Return a dictionary that contains the parameter metadata grouped by the\n    instrument it belongs to.\n    \"\"\"\n    metadata_timestamp = time.time()\n    # group metadata by instrument\n    metas: dict[Any, Any] = defaultdict(list)\n    for parameter in parameters:\n        # Get the latest value from the parameter,\n        # respecting the max_val_age parameter\n        meta: dict[str, float | str | None] = {}\n        meta[\"value\"] = str(parameter.get_latest())\n        timestamp = parameter.get_latest.get_timestamp()\n        if timestamp is not None:\n            meta[\"ts\"] = timestamp.timestamp()\n        else:\n            meta[\"ts\"] = None\n        meta[\"name\"] = parameter.label or parameter.name\n        meta[\"unit\"] = parameter.unit\n\n        # find the base instrument that this parameter belongs to\n        if use_root_instrument:\n            baseinst = parameter.root_instrument\n        else:\n            baseinst = parameter.instrument\n        if baseinst is None:\n            metas[\"Unbound Parameter\"].append(meta)\n        else:\n            metas[str(baseinst)].append(meta)\n\n    # Create list of parameters, grouped by instrument\n    parameters_out = []\n    for instrument in metas:\n        temp = {\"instrument\": instrument, \"parameters\": metas[instrument]}\n        parameters_out.append(temp)\n\n    state = {\"ts\": metadata_timestamp, \"parameters\": parameters_out}\n    return state\n\n\ndef _handler(\n    parameters: Sequence[Parameter], interval: float, use_root_instrument: bool = True\n) -> Callable[[websockets.server.WebSocketServerProtocol, str], Awaitable[None]]:\n    \"\"\"\n    Return the websockets server handler.\n    \"\"\"\n\n    async def server_func(\n        websocket: websockets.server.WebSocketServerProtocol, _: str\n    ) -> None:\n        \"\"\"\n        Create a websockets handler that sends parameter values to a listener\n        every \"interval\" seconds.\n        \"\"\"\n        while True:\n            try:\n                # Update the parameter values\n                try:\n                    meta = _get_metadata(\n                        *parameters, use_root_instrument=use_root_instrument\n                    )\n                except ValueError:\n                    log.exception(\"Error getting parameters\")\n                    break\n                log.debug(\"sending.. to %r\", websocket)\n                await websocket.send(json.dumps(meta))\n                # Wait for interval seconds and then send again\n                await asyncio.sleep(interval)\n            except (CancelledError, websockets.exceptions.ConnectionClosed):\n                log.debug(\"Got CancelledError or ConnectionClosed\",\n                          exc_info=True)\n                break\n        log.debug(\"Closing websockets connection\")\n\n    return server_func\n\n\nclass Monitor(Thread):\n    \"\"\"\n    QCodes Monitor - WebSockets server to monitor qcodes parameters.\n    \"\"\"\n\n    running: Monitor | None = None\n\n    def __init__(\n        self,\n        *parameters: Parameter,\n        interval: float = 1,\n        use_root_instrument: bool = True,\n    ):\n        \"\"\"\n        Monitor qcodes parameters.\n\n        Args:\n            *parameters: Parameters to monitor.\n            interval: How often one wants to refresh the values.\n            use_root_instrument: Defines if parameters are grouped according to\n                                parameter.root_instrument or parameter.instrument\n        \"\"\"\n        super().__init__(daemon=True)\n\n        # Check that all values are valid parameters\n        for parameter in parameters:\n            if not isinstance(parameter, Parameter):\n                raise TypeError(f\"We can only monitor QCodes \"\n                                f\"Parameters, not {type(parameter)}\")\n\n        self.loop: asyncio.AbstractEventLoop | None = None\n        self._stop_loop_future: asyncio.Future | None = None\n        self._parameters = parameters\n        self.loop_is_closed = Event()\n        self.server_is_started = Event()\n        self.handler = _handler(\n            parameters, interval=interval, use_root_instrument=use_root_instrument\n        )\n        log.debug(\"Start monitoring thread\")\n        if Monitor.running:\n            # stop the old server\n            log.debug(\"Stopping and restarting server\")\n            Monitor.running.stop()\n        self.start()\n\n        # Wait until the loop is running\n        self.server_is_started.wait(timeout=5)\n        if not self.server_is_started.is_set():\n            raise RuntimeError(\"Failed to start server\")\n        Monitor.running = self\n\n    def run(self) -> None:\n        \"\"\"\n        Start the event loop and run forever.\n        \"\"\"\n        log.debug(\"Running Websocket server\")\n\n        async def run_loop() -> None:\n            self.loop = asyncio.get_running_loop()\n            self._stop_loop_future = self.loop.create_future()\n\n            async with websockets.server.serve(\n                self.handler, \"127.0.0.1\", WEBSOCKET_PORT, close_timeout=1\n            ):\n                self.server_is_started.set()\n                try:\n                    await self._stop_loop_future\n                except asyncio.CancelledError:\n                    log.debug(\"Websocket server thread shutting down\")\n\n        try:\n            asyncio.run(run_loop())\n        finally:\n            self.loop_is_closed.set()\n\n    def update_all(self) -> None:\n        \"\"\"\n        Update all parameters in the monitor.\n        \"\"\"\n        for parameter in self._parameters:\n            # call get if it can be called without arguments\n            with suppress(TypeError):\n                parameter.get()\n\n    def stop(self) -> None:\n        \"\"\"\n        Shutdown the server, close the event loop and join the thread.\n        Setting active Monitor to ``None``.\n        \"\"\"\n        self.join()\n        Monitor.running = None\n\n    def join(self, timeout: float | None = None) -> None:\n        \"\"\"\n        Overwrite ``Thread.join`` to make sure server is stopped before\n        joining avoiding a potential deadlock.\n        \"\"\"\n        log.debug(\"Shutting down server\")\n        if not self.is_alive():\n            # we run this check before trying to run to prevent a cryptic\n            # error message\n            log.debug(\"monitor is dead\")\n            return\n        try:\n            if self.loop is not None and self._stop_loop_future is not None:\n                log.debug(\"Instructing server to stop event loop.\")\n                self.loop.call_soon_threadsafe(self._stop_loop_future.cancel)\n            else:\n                log.debug(\"No event loop found. Cannot stop event loop.\")\n        except RuntimeError:\n            # the above may throw a runtime error if the loop is already\n            # stopped in which case there is nothing more to do\n            log.exception(\"Could not close loop\")\n        self.loop_is_closed.wait(timeout=5)\n        if not self.loop_is_closed.is_set():\n            raise RuntimeError(\"Failed to join loop\")\n        log.debug(\"Loop reported closed\")\n        super().join(timeout=timeout)\n        log.debug(\"Monitor Thread has joined\")\n\n    @staticmethod\n    def show() -> None:\n        \"\"\"\n        Overwrite this method to show/raise your monitor GUI\n        F.ex.\n\n        ::\n\n            import webbrowser\n            url = \"localhost:3000\"\n            # Open URL in new window, raising the window if possible.\n            webbrowser.open_new(url)\n\n        \"\"\"\n        webbrowser.open(f\"http://localhost:{SERVER_PORT}\")\n\n\ndef main() -> None:\n    import http.server\n\n    # If this file is run, create a simple webserver that serves a simple\n    # website that can be used to view monitored parameters.\n    # # https://github.com/python/mypy/issues/4182\n    parent_module = \".\".join(__loader__.name.split(\".\")[:-1])  # type: ignore[name-defined]\n\n    static_dir = files(parent_module).joinpath(\"dist\")\n    try:\n        with as_file(static_dir) as extracted_dir:\n            os.chdir(extracted_dir)\n            log.info(\"Starting HTTP Server at http://localhost:%i\", SERVER_PORT)\n            with socketserver.TCPServer(\n                (\"\", SERVER_PORT), http.server.SimpleHTTPRequestHandler\n            ) as httpd:\n                log.debug(\"serving directory %s\", static_dir)\n                webbrowser.open(f\"http://localhost:{SERVER_PORT}\")\n                httpd.serve_forever()\n    except KeyboardInterrupt:\n        log.info(\"Shutting Down HTTP Server\")\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "src/qcodes/monitor/monitor.py"}], "after_files": [{"content": "#! /usr/bin/env python\n# vim:fenc=utf-8\n#\n# Copyright \u00a9 2017 unga <giulioungaretti@me.com>\n#\n# Distributed under terms of the MIT license.\n\"\"\"\nMonitor a set of parameters in a background thread\nstream output over websocket\n\nTo start monitor, run this file, or if qcodes is installed as a module:\n\n``% python -m qcodes.monitor.monitor``\n\nAdd parameters to monitor in your measurement by creating a new monitor with a\nlist of parameters to monitor:\n\n``monitor = qcodes.Monitor(param1, param2, param3, ...)``\n\"\"\"\nfrom __future__ import annotations\n\nimport asyncio\nimport json\nimport logging\nimport os\nimport socketserver\nimport time\nimport webbrowser\nfrom asyncio import CancelledError\nfrom collections import defaultdict\nfrom contextlib import suppress\nfrom importlib.resources import as_file, files\nfrom threading import Event, Thread\nfrom typing import TYPE_CHECKING, Any, Callable\n\nimport websockets\nimport websockets.exceptions\nimport websockets.server\n\nfrom qcodes.parameters import Parameter\n\nif TYPE_CHECKING:\n    from collections.abc import Awaitable, Sequence\n\nWEBSOCKET_PORT = 5678\nSERVER_PORT = 3000\n\nlog = logging.getLogger(__name__)\n\n\ndef _get_metadata(\n    *parameters: Parameter, use_root_instrument: bool = True\n) -> dict[str, Any]:\n    \"\"\"\n    Return a dictionary that contains the parameter metadata grouped by the\n    instrument it belongs to.\n    \"\"\"\n    metadata_timestamp = time.time()\n    # group metadata by instrument\n    metas: dict[Any, Any] = defaultdict(list)\n    for parameter in parameters:\n        # Get the latest value from the parameter,\n        # respecting the max_val_age parameter\n        meta: dict[str, float | str | None] = {}\n        meta[\"value\"] = str(parameter.get_latest())\n        timestamp = parameter.get_latest.get_timestamp()\n        if timestamp is not None:\n            meta[\"ts\"] = timestamp.timestamp()\n        else:\n            meta[\"ts\"] = None\n        meta[\"name\"] = parameter.label or parameter.name\n        meta[\"unit\"] = parameter.unit\n\n        # find the base instrument that this parameter belongs to\n        if use_root_instrument:\n            baseinst = parameter.root_instrument\n        else:\n            baseinst = parameter.instrument\n        if baseinst is None:\n            metas[\"Unbound Parameter\"].append(meta)\n        else:\n            metas[str(baseinst)].append(meta)\n\n    # Create list of parameters, grouped by instrument\n    parameters_out = []\n    for instrument in metas:\n        temp = {\"instrument\": instrument, \"parameters\": metas[instrument]}\n        parameters_out.append(temp)\n\n    state = {\"ts\": metadata_timestamp, \"parameters\": parameters_out}\n    return state\n\n\ndef _handler(\n    parameters: Sequence[Parameter], interval: float, use_root_instrument: bool = True\n) -> Callable[[websockets.server.WebSocketServerProtocol, str], Awaitable[None]]:\n    \"\"\"\n    Return the websockets server handler.\n    \"\"\"\n\n    async def server_func(\n        websocket: websockets.server.WebSocketServerProtocol, _: str\n    ) -> None:\n        \"\"\"\n        Create a websockets handler that sends parameter values to a listener\n        every \"interval\" seconds.\n        \"\"\"\n        while True:\n            try:\n                # Update the parameter values\n                try:\n                    meta = _get_metadata(\n                        *parameters, use_root_instrument=use_root_instrument\n                    )\n                except ValueError:\n                    log.exception(\"Error getting parameters\")\n                    break\n                log.debug(\"sending.. to %r\", websocket)\n                await websocket.send(json.dumps(meta))\n                # Wait for interval seconds and then send again\n                await asyncio.sleep(interval)\n            except (CancelledError, websockets.exceptions.ConnectionClosed):\n                log.debug(\"Got CancelledError or ConnectionClosed\",\n                          exc_info=True)\n                break\n        log.debug(\"Closing websockets connection\")\n\n    return server_func\n\n\nclass Monitor(Thread):\n    \"\"\"\n    QCodes Monitor - WebSockets server to monitor qcodes parameters.\n    \"\"\"\n\n    running: Monitor | None = None\n\n    def __init__(\n        self,\n        *parameters: Parameter,\n        interval: float = 1,\n        use_root_instrument: bool = True,\n    ):\n        \"\"\"\n        Monitor qcodes parameters.\n\n        Args:\n            *parameters: Parameters to monitor.\n            interval: How often one wants to refresh the values.\n            use_root_instrument: Defines if parameters are grouped according to\n                                parameter.root_instrument or parameter.instrument\n        \"\"\"\n        super().__init__(daemon=True)\n\n        # Check that all values are valid parameters\n        for parameter in parameters:\n            if not isinstance(parameter, Parameter):\n                raise TypeError(f\"We can only monitor QCodes \"\n                                f\"Parameters, not {type(parameter)}\")\n\n        self.loop: asyncio.AbstractEventLoop | None = None\n        self._stop_loop_future: asyncio.Future | None = None\n        self._parameters = parameters\n        self.loop_is_closed = Event()\n        self.server_is_started = Event()\n        self.handler = _handler(\n            parameters, interval=interval, use_root_instrument=use_root_instrument\n        )\n        log.debug(\"Start monitoring thread\")\n        if Monitor.running:\n            # stop the old server\n            log.debug(\"Stopping and restarting server\")\n            Monitor.running.stop()\n        self.start()\n\n        # Wait until the loop is running\n        self.server_is_started.wait(timeout=5)\n        if not self.server_is_started.is_set():\n            raise RuntimeError(\"Failed to start server\")\n        Monitor.running = self\n\n    def run(self) -> None:\n        \"\"\"\n        Start the event loop and run forever.\n        \"\"\"\n        log.debug(\"Running Websocket server\")\n\n        async def run_loop() -> None:\n            self.loop = asyncio.get_running_loop()\n            self._stop_loop_future = self.loop.create_future()\n\n            async with websockets.server.serve(\n                self.handler, \"127.0.0.1\", WEBSOCKET_PORT, close_timeout=1\n            ):\n                self.server_is_started.set()\n                await self._stop_loop_future\n                log.debug(\"Websocket server thread shutting down\")\n\n        try:\n            asyncio.run(run_loop())\n        finally:\n            self.loop_is_closed.set()\n\n    def update_all(self) -> None:\n        \"\"\"\n        Update all parameters in the monitor.\n        \"\"\"\n        for parameter in self._parameters:\n            # call get if it can be called without arguments\n            with suppress(TypeError):\n                parameter.get()\n\n    def stop(self) -> None:\n        \"\"\"\n        Shutdown the server, close the event loop and join the thread.\n        Setting active Monitor to ``None``.\n        \"\"\"\n        self.join()\n        Monitor.running = None\n\n    def join(self, timeout: float | None = None) -> None:\n        \"\"\"\n        Overwrite ``Thread.join`` to make sure server is stopped before\n        joining avoiding a potential deadlock.\n        \"\"\"\n        log.debug(\"Shutting down server\")\n        if not self.is_alive():\n            # we run this check before trying to run to prevent a cryptic\n            # error message\n            log.debug(\"monitor is dead\")\n            return\n        try:\n            if self.loop is not None and self._stop_loop_future is not None:\n                log.debug(\"Instructing server to stop event loop.\")\n                self.loop.call_soon_threadsafe(self._stop_loop_future.set_result, True)\n            else:\n                log.debug(\"No event loop found. Cannot stop event loop.\")\n        except RuntimeError:\n            # the above may throw a runtime error if the loop is already\n            # stopped in which case there is nothing more to do\n            log.exception(\"Could not close loop\")\n        self.loop_is_closed.wait(timeout=5)\n        if not self.loop_is_closed.is_set():\n            raise RuntimeError(\"Failed to join loop\")\n        log.debug(\"Loop reported closed\")\n        super().join(timeout=timeout)\n        log.debug(\"Monitor Thread has joined\")\n\n    @staticmethod\n    def show() -> None:\n        \"\"\"\n        Overwrite this method to show/raise your monitor GUI\n        F.ex.\n\n        ::\n\n            import webbrowser\n            url = \"localhost:3000\"\n            # Open URL in new window, raising the window if possible.\n            webbrowser.open_new(url)\n\n        \"\"\"\n        webbrowser.open(f\"http://localhost:{SERVER_PORT}\")\n\n\ndef main() -> None:\n    import http.server\n\n    # If this file is run, create a simple webserver that serves a simple\n    # website that can be used to view monitored parameters.\n    # # https://github.com/python/mypy/issues/4182\n    parent_module = \".\".join(__loader__.name.split(\".\")[:-1])  # type: ignore[name-defined]\n\n    static_dir = files(parent_module).joinpath(\"dist\")\n    try:\n        with as_file(static_dir) as extracted_dir:\n            os.chdir(extracted_dir)\n            log.info(\"Starting HTTP Server at http://localhost:%i\", SERVER_PORT)\n            with socketserver.TCPServer(\n                (\"\", SERVER_PORT), http.server.SimpleHTTPRequestHandler\n            ) as httpd:\n                log.debug(\"serving directory %s\", static_dir)\n                webbrowser.open(f\"http://localhost:{SERVER_PORT}\")\n                httpd.serve_forever()\n    except KeyboardInterrupt:\n        log.info(\"Shutting Down HTTP Server\")\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "src/qcodes/monitor/monitor.py"}]}