[docs]classMonitor(Thread):""" QCodes Monitor - WebSockets server to monitor qcodes parameters. """running:Monitor|None=Nonedef__init__(self,*parameters:Parameter,interval:float=1,use_root_instrument:bool=True,):""" Monitor qcodes parameters. Args: *parameters: Parameters to monitor. interval: How often one wants to refresh the values. use_root_instrument: Defines if parameters are grouped according to parameter.root_instrument or parameter.instrument """super().__init__(daemon=True)# Check that all values are valid parametersforparameterinparameters:ifnotisinstance(parameter,Parameter):raiseTypeError(f"We can only monitor QCodes Parameters, not {type(parameter)}")self.loop:asyncio.AbstractEventLoop|None=Noneself._stop_loop_future:asyncio.Future|None=Noneself._parameters=parametersself.loop_is_closed=Event()self.server_is_started=Event()self.handler=_handler(parameters,interval=interval,use_root_instrument=use_root_instrument)log.debug("Start monitoring thread")ifMonitor.running:# stop the old serverlog.debug("Stopping and restarting server")Monitor.running.stop()self.start()# Wait until the loop is runningself.server_is_started.wait(timeout=5)ifnotself.server_is_started.is_set():raiseRuntimeError("Failed to start server")Monitor.running=self
[docs]defrun(self)->None:""" Start the event loop and run forever. """log.debug("Running Websocket server")asyncdefrun_loop()->None:self.loop=asyncio.get_running_loop()self._stop_loop_future=self.loop.create_future()asyncwithwebsockets.server.serve(self.handler,"127.0.0.1",WEBSOCKET_PORT,close_timeout=1):self.server_is_started.set()awaitself._stop_loop_futurelog.debug("Websocket server thread shutting down")try:asyncio.run(run_loop())finally:self.loop_is_closed.set()
[docs]defupdate_all(self)->None:""" Update all parameters in the monitor. """forparameterinself._parameters:# call get if it can be called without argumentswithsuppress(TypeError):parameter.get()
[docs]defstop(self)->None:""" Shutdown the server, close the event loop and join the thread. Setting active Monitor to ``None``. """self.join()Monitor.running=None
[docs]defjoin(self,timeout:float|None=None)->None:""" Overwrite ``Thread.join`` to make sure server is stopped before joining avoiding a potential deadlock. """log.debug("Shutting down server")ifnotself.is_alive():# we run this check before trying to run to prevent a cryptic# error messagelog.debug("monitor is dead")returntry:ifself.loopisnotNoneandself._stop_loop_futureisnotNone:log.debug("Instructing server to stop event loop.")self.loop.call_soon_threadsafe(self._stop_loop_future.set_result,True)else:log.debug("No event loop found. Cannot stop event loop.")exceptRuntimeError:# the above may throw a runtime error if the loop is already# stopped in which case there is nothing more to dolog.exception("Could not close loop")self.loop_is_closed.wait(timeout=5)ifnotself.loop_is_closed.is_set():raiseRuntimeError("Failed to join loop")log.debug("Loop reported closed")super().join(timeout=timeout)log.debug("Monitor Thread has joined")
[docs]@staticmethoddefshow()->None:""" Overwrite this method to show/raise your monitor GUI F.ex. :: import webbrowser url = "localhost:3000" # Open URL in new window, raising the window if possible. webbrowser.open_new(url) """webbrowser.open(f"http://localhost:{SERVER_PORT}")
defmain()->None:importhttp.server# If this file is run, create a simple webserver that serves a simple# website that can be used to view monitored parameters.# # https://github.com/python/mypy/issues/4182parent_module=".".join(__loader__.name.split(".")[:-1])# type: ignore[name-defined]static_dir=files(parent_module).joinpath("dist")try:withas_file(static_dir)asextracted_dir:os.chdir(extracted_dir)log.info("Starting HTTP Server at http://localhost:%i",SERVER_PORT)withsocketserver.TCPServer(("",SERVER_PORT),http.server.SimpleHTTPRequestHandler)ashttpd:log.debug("serving directory %s",static_dir)webbrowser.open(f"http://localhost:{SERVER_PORT}")httpd.serve_forever()exceptKeyboardInterrupt:log.info("Shutting Down HTTP Server")if__name__=="__main__":main()