calibration_server.py File Reference

Go to the source code of this file.

Classes
class	calibration_server.ElodinWriter
class	calibration_server.ServerState
class	calibration_server.CalibrationHTTPRequestHandler

Namespaces
namespace	calibration_server

Functions
Optional[int]	calibration_server._parse_raw_adc (bytes payload)
Optional[int]	calibration_server._parse_raw_signed (bytes payload)
Optional[int]	calibration_server._parse_int32_at_12 (bytes payload)
float	calibration_server._apply_ema (int key, float value, float alpha=_EMA_ALPHA_DEFAULT)
None	calibration_server._reset_ema_for_packet_channels (list packet_channels)
bool	calibration_server._apply_direct_offset_correction (tuple key, float adc_code, float target_psi)
Optional[int]	calibration_server._parse_pt_raw_for_packet_ch (bytes payload, int packet_ch)
str	calibration_server._throttle_key (str stype, int ch)
	calibration_server._get_raw_conversion_config ()
None	calibration_server._process_raw_and_write_calibrated (str stype, int packet_ch, int raw_val, Any writer, dict channel_to_key, *, int connector_id, int raw_packet_low)
None	calibration_server._auto_save_adjustments ()
	calibration_server.relay_subscriber_task ()
	calibration_server.run_http_server (host, port)
	calibration_server.ws_handler (websocket)
	calibration_server.broadcast_ws (message)
	calibration_server.run_ws_server (host, port)

Variables
	calibration_server.level
	calibration_server.INFO
	calibration_server.format
	calibration_server.logger = logging.getLogger("CalibrationServer")
	calibration_server.config = load_config()
dict	calibration_server._last_elodin_write = {}
set	calibration_server._first_calibrated_write_logged = set()
dict	calibration_server._hp_pt_channels_cache = {}
float	calibration_server._EMA_ALPHA_DEFAULT = 0.4
dict	calibration_server._ema_state = {}
dict	calibration_server._raw_conversion_config_cache = {}
str	calibration_server._ADJUSTMENTS_PATH = Path(__file__).resolve().parent / "calibrations" / "adjustments.json"
	calibration_server.state = ServerState()
	calibration_server.sidecar_cfg = config.get("calibration", {}).get("sidecar", {})
	calibration_server.host = sidecar_cfg.get("host", "127.0.0.1")
	calibration_server.port = sidecar_cfg.get("port", 8100)
	calibration_server.http_thread
	calibration_server.ws_thread
	calibration_server.pending_alerts
	calibration_server.running