mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-07-22 01:32:11 +08:00
happy saturday
This commit is contained in:
firestar5683
@@ -51,7 +51,9 @@
|
||||
{.msg = {{ 0xaa, 0, 8, 83U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, { 0 }, { 0 }}}, \
|
||||
{.msg = {{0x260, 0, 8, 50U, .ignore_counter = true, .ignore_quality_flag=!(lta)}, { 0 }, { 0 }}}, \
|
||||
/* StarPilot Variables */ \
|
||||
{.msg = {{0x1D3, 0, 8, 33U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, { 0 }, { 0 }}}, \
|
||||
{.msg = {{0x1D3, 0, 8, 33U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, \
|
||||
{0x1D3, 0, 5, 33U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}, \
|
||||
{0x365, 0, 7, 10U, .ignore_checksum = true, .ignore_counter = true, .ignore_quality_flag = true}}}, \
|
||||
|
||||
#define TOYOTA_RX_CHECKS(lta) \
|
||||
TOYOTA_COMMON_RX_CHECKS(lta) \
|
||||
@@ -108,6 +110,12 @@ static bool toyota_get_quality_flag_valid(const CANPacket_t *msg) {
|
||||
return valid;
|
||||
}
|
||||
|
||||
static void toyota_rx_all_hook(const CANPacket_t *msg) {
|
||||
if ((msg->bus == 0U) && (msg->addr == 0x365U) && (GET_LEN(msg) == 7U)) {
|
||||
acc_main_on = GET_BIT(msg, 0U); // DSU_CRUISE.MAIN_ON
|
||||
}
|
||||
}
|
||||
|
||||
static void toyota_rx_hook(const CANPacket_t *msg) {
|
||||
if (msg->bus == 0U) {
|
||||
|
||||
@@ -185,14 +193,10 @@ static void toyota_rx_hook(const CANPacket_t *msg) {
|
||||
UPDATE_VEHICLE_SPEED(speed / 4.0 * 0.01 * KPH_TO_MS);
|
||||
}
|
||||
|
||||
if (msg->addr == 0x1D3U) {
|
||||
if ((msg->addr == 0x1D3U) && (GET_LEN(msg) == 8U)) {
|
||||
acc_main_on = GET_BIT(msg, 15U);
|
||||
}
|
||||
|
||||
if (msg->addr == 0x365U) {
|
||||
acc_main_on = GET_BIT(msg, 0U);
|
||||
}
|
||||
|
||||
if (enable_gas_interceptor && (msg->addr == 0x201U)) {
|
||||
// Match the DBC's physical pedal threshold to avoid controls state mismatches.
|
||||
const int toyota_gas_interceptor_threshold = 805;
|
||||
@@ -542,6 +546,7 @@ static bool toyota_fwd_hook(int bus_num, int addr) {
|
||||
const safety_hooks toyota_hooks = {
|
||||
.init = toyota_init,
|
||||
.rx = toyota_rx_hook,
|
||||
.rx_all = toyota_rx_all_hook,
|
||||
.tx = toyota_tx_hook,
|
||||
.fwd = toyota_fwd_hook,
|
||||
.get_checksum = toyota_get_checksum,
|
||||
|
||||
@@ -155,6 +155,44 @@ class TestToyotaSafetyBase(common.CarSafetyTest, common.LongitudinalAccelSafetyT
|
||||
return self.packer.make_can_msg_panda("PCM_CRUISE_2", 0, values)
|
||||
|
||||
|
||||
def test_toyota_aol_rx_check_variants():
|
||||
packer = CANPackerSafety("toyota_nodsu_pt_generated")
|
||||
safety = libsafety_py.libsafety
|
||||
|
||||
for addr, length in ((0x1D3, 8), (0x1D3, 5), (0x365, 7)):
|
||||
safety.set_safety_hooks(CarParams.SafetyModel.toyota, TestToyotaSafetyBase.EPS_SCALE)
|
||||
safety.init_tests()
|
||||
|
||||
messages = (
|
||||
packer.make_can_msg_safety("WHEEL_SPEEDS", 0, {f"WHEEL_SPEED_{wheel}": 0 for wheel in ("FR", "FL", "RR", "RL")}),
|
||||
packer.make_can_msg_safety("STEER_TORQUE_SENSOR", 0, {"STEER_TORQUE_EPS": 0}),
|
||||
packer.make_can_msg_safety("PCM_CRUISE", 0, {"CRUISE_ACTIVE": 0}),
|
||||
packer.make_can_msg_safety("BRAKE_MODULE", 0, {"BRAKE_PRESSED": 0}),
|
||||
libsafety_py.make_CANPacket(addr, 0, bytes(length)),
|
||||
)
|
||||
for msg in messages:
|
||||
assert safety.safety_rx_hook(msg)
|
||||
assert safety.safety_config_valid(), f"RX checks invalid for {addr=:#x}, {length=}"
|
||||
|
||||
|
||||
def test_toyota_dsu_cruise_main_bit():
|
||||
safety = libsafety_py.libsafety
|
||||
safety.set_safety_hooks(CarParams.SafetyModel.toyota, TestToyotaSafetyBase.EPS_SCALE)
|
||||
safety.init_tests()
|
||||
|
||||
safety.safety_rx_hook(libsafety_py.make_CANPacket(0x1D3, 0, bytes(5)))
|
||||
assert not safety.get_acc_main_on()
|
||||
|
||||
safety.safety_rx_hook(libsafety_py.make_CANPacket(0x365, 0, b"\x01\x00\x00\x00\x00\x00\x00"))
|
||||
assert safety.get_acc_main_on()
|
||||
|
||||
safety.safety_rx_hook(libsafety_py.make_CANPacket(0x1D3, 0, bytes(5)))
|
||||
assert safety.get_acc_main_on()
|
||||
|
||||
safety.safety_rx_hook(libsafety_py.make_CANPacket(0x365, 0, bytes(7)))
|
||||
assert not safety.get_acc_main_on()
|
||||
|
||||
|
||||
class TestToyotaSafetyTorque(TestToyotaSafetyBase, common.MotorTorqueSteeringSafetyTest, common.SteerRequestCutSafetyTest):
|
||||
|
||||
MAX_RATE_UP = 15
|
||||
|
||||
@@ -335,8 +335,8 @@ KIA_NIRO_PHEV_2022_FRICTION_THRESHOLD_GAIN = 0.12
|
||||
KIA_CARNIVAL_CENTER_TAPER_MAX = 0.10
|
||||
KIA_CARNIVAL_CENTER_TAPER_LAT = 0.16
|
||||
KIA_CARNIVAL_CENTER_TAPER_LAT_WIDTH = 0.05
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED = 6.5
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED_WIDTH = 1.5
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED = 3.5
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED_WIDTH = 1.8
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED_MAX = 16.0
|
||||
KIA_CARNIVAL_CENTER_TAPER_SPEED_MAX_WIDTH = 2.5
|
||||
KIA_CARNIVAL_FRICTION_THRESHOLD_GAIN = 0.18
|
||||
|
||||
@@ -313,6 +313,9 @@ UNCERT_PANIC_MIN_CLOSING_SPEED = 2.0
|
||||
UNCERT_PANIC_MIN_CLOSING_SPEED_GAIN = 0.08
|
||||
UNCERT_PANIC_MAX_GAP_BUFFER_MIN = 8.0
|
||||
UNCERT_PANIC_MAX_GAP_BUFFER_GAIN = 0.35
|
||||
UNCERT_DUPLICATE_VISION_MIN_TTC = 6.0
|
||||
UNCERT_DUPLICATE_VISION_MIN_HEADWAY = 0.85
|
||||
UNCERT_DUPLICATE_VISION_HEADWAY_BELOW_TARGET = 0.45
|
||||
STEADY_FOLLOW_SMOOTHING_MIN_SPEED = 22.0
|
||||
STEADY_FOLLOW_SMOOTHING_MIN_CLOSING_SPEED = 0.15
|
||||
STEADY_FOLLOW_SMOOTHING_MAX_CLOSING_SPEED = 1.8
|
||||
@@ -1927,6 +1930,27 @@ class LongitudinalPlanner:
|
||||
self.duplicate_vision_comfort_lead_source = selected_source
|
||||
return selected_lead
|
||||
|
||||
def is_nonurgent_duplicate_vision_follow(self, v_ego, t_follow):
|
||||
if bool(getattr(self.lead_one, "radar", False)) or bool(getattr(self.lead_two, "radar", False)):
|
||||
return False
|
||||
if not self.mpc.leads_are_near_duplicates(
|
||||
self.lead_one,
|
||||
self.lead_two,
|
||||
v_ego,
|
||||
vision_min_speed=LOW_SPEED_MATCHED_FOLLOW_TRANSITION_MIN_SPEED,
|
||||
):
|
||||
return False
|
||||
|
||||
lead = self.lead_one
|
||||
closing_speed = max(0.0, float(v_ego) - float(lead.vLead))
|
||||
ttc = float(lead.dRel) / max(closing_speed, 0.1) if closing_speed > 0.1 else float("inf")
|
||||
actual_headway = float(lead.dRel) / max(float(v_ego), 1e-3)
|
||||
minimum_headway = max(
|
||||
UNCERT_DUPLICATE_VISION_MIN_HEADWAY,
|
||||
float(t_follow) - UNCERT_DUPLICATE_VISION_HEADWAY_BELOW_TARGET,
|
||||
)
|
||||
return ttc > UNCERT_DUPLICATE_VISION_MIN_TTC and actual_headway > minimum_headway
|
||||
|
||||
def lead_is_spacious_brake_cap_window(self, lead, v_ego, base_t_follow):
|
||||
if lead is None or not lead.status or v_ego < STEADY_FOLLOW_SMOOTHING_MIN_SPEED:
|
||||
return False
|
||||
@@ -2706,6 +2730,10 @@ class LongitudinalPlanner:
|
||||
panic_bypass = panic_close_window and closing_fast and (
|
||||
uncert_slope > UNCERT_SLOPE_TRIG or uncertainty >= UNCERT_MAG_TRIG
|
||||
)
|
||||
# Duplicate vision tracks can share the same noisy velocity spike. Keep the
|
||||
# comfort path unless distance, TTC, or lead braking makes the scene urgent.
|
||||
if panic_bypass and self.is_nonurgent_duplicate_vision_follow(scene_v_ego, effective_t_follow):
|
||||
panic_bypass = False
|
||||
|
||||
steady_follow_filter_floor = 0.0
|
||||
if lead_one_active and desired_gap is not None and not panic_bypass:
|
||||
|
||||
@@ -410,10 +410,14 @@ class TestLatControl:
|
||||
center_taper = get_kia_carnival_center_taper_scale(0.04, 8.5)
|
||||
turn_taper = get_kia_carnival_center_taper_scale(0.35, 8.5)
|
||||
low_speed_taper = get_kia_carnival_center_taper_scale(0.04, 2.0)
|
||||
neighborhood_taper = get_kia_carnival_center_taper_scale(0.04, 5.0)
|
||||
neighborhood_turn_taper = get_kia_carnival_center_taper_scale(0.35, 5.0)
|
||||
highway_taper = get_kia_carnival_center_taper_scale(0.04, 25.0)
|
||||
assert center_taper < turn_taper <= 1.0
|
||||
assert center_taper < low_speed_taper <= 1.0
|
||||
assert center_taper < highway_taper <= 1.0
|
||||
assert neighborhood_taper < 0.94
|
||||
assert neighborhood_turn_taper > 0.99
|
||||
|
||||
center_threshold = get_kia_carnival_friction_threshold(8.5, 0.04)
|
||||
turn_threshold = get_kia_carnival_friction_threshold(8.5, 0.35)
|
||||
|
||||
@@ -3812,6 +3812,55 @@ def test_duplicate_vision_comfort_lead_resets_when_duplicate_cluster_clears():
|
||||
assert planner.duplicate_vision_comfort_lead_source is None
|
||||
|
||||
|
||||
def test_nonurgent_duplicate_vision_follow_keeps_comfort_smoothing():
|
||||
v_ego = 25.0
|
||||
CP = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC)
|
||||
planner = LongitudinalPlanner(CP, init_v=v_ego)
|
||||
lead_one = make_lead(status=True, d_rel=35.0, v_lead=20.5, a_lead=-0.05, radar=False, model_prob=0.99)
|
||||
lead_two = make_lead(status=True, d_rel=35.2, v_lead=20.52, a_lead=-0.04, radar=False, model_prob=0.99)
|
||||
lead_one.vRel = lead_one.vLead - v_ego
|
||||
lead_two.vRel = lead_two.vLead - v_ego
|
||||
planner.lead_one = lead_one
|
||||
planner.lead_two = lead_two
|
||||
|
||||
assert planner.is_nonurgent_duplicate_vision_follow(v_ego, 1.45)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("d_rel,v_lead,a_lead", [
|
||||
(24.0, 20.0, -0.05), # Low TTC.
|
||||
(22.0, 23.0, -0.05), # Headway materially below the requested gap.
|
||||
(35.0, 20.5, -0.50), # The lead is braking.
|
||||
])
|
||||
def test_duplicate_vision_follow_preserves_urgent_panic_bypass(d_rel, v_lead, a_lead):
|
||||
v_ego = 25.0
|
||||
CP = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC)
|
||||
planner = LongitudinalPlanner(CP, init_v=v_ego)
|
||||
lead_one = make_lead(status=True, d_rel=d_rel, v_lead=v_lead, a_lead=a_lead, radar=False, model_prob=0.99)
|
||||
lead_two = make_lead(status=True, d_rel=d_rel + 0.2, v_lead=v_lead + 0.02, a_lead=a_lead, radar=False, model_prob=0.99)
|
||||
lead_one.vRel = lead_one.vLead - v_ego
|
||||
lead_two.vRel = lead_two.vLead - v_ego
|
||||
planner.lead_one = lead_one
|
||||
planner.lead_two = lead_two
|
||||
|
||||
assert not planner.is_nonurgent_duplicate_vision_follow(v_ego, 1.45)
|
||||
|
||||
|
||||
def test_radar_duplicates_preserve_panic_bypass():
|
||||
v_ego = 25.0
|
||||
CP = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC)
|
||||
planner = LongitudinalPlanner(CP, init_v=v_ego)
|
||||
lead_one = make_lead(status=True, d_rel=35.0, v_lead=20.5, a_lead=-0.05, radar=True, model_prob=1.0)
|
||||
lead_two = make_lead(status=True, d_rel=35.0, v_lead=20.5, a_lead=-0.05, radar=True, model_prob=1.0)
|
||||
lead_one.radarTrackId = 17
|
||||
lead_two.radarTrackId = 17
|
||||
lead_one.vRel = lead_one.vLead - v_ego
|
||||
lead_two.vRel = lead_two.vLead - v_ego
|
||||
planner.lead_one = lead_one
|
||||
planner.lead_two = lead_two
|
||||
|
||||
assert not planner.is_nonurgent_duplicate_vision_follow(v_ego, 1.45)
|
||||
|
||||
|
||||
def test_near_duplicate_lead_transition_target_damps_same_source_sign_flip():
|
||||
v_ego = 25.0
|
||||
CP = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC)
|
||||
|
||||
@@ -427,12 +427,14 @@ function renderVitals(device) {
|
||||
const uptime = device.uptimeSeconds == null ? "unknown" : formatDuration(device.uptimeSeconds);
|
||||
const cpu = device.cpuTempC == null ? "unknown" : `${formatInt(device.cpuTempC)} C`;
|
||||
const lanIp = device.lanIp || "unknown";
|
||||
const networkName = device.networkName || "No wireless connectivity";
|
||||
return `
|
||||
<section class="dashboard-card dashboard-device-card">
|
||||
<h2>Vitals</h2>
|
||||
<div class="dashboard-key-values">
|
||||
<div><span>Status</span><strong>${escapeHtml(device.status || "Parked")}</strong></div>
|
||||
<div><span>LAN IP</span><strong>${escapeHtml(lanIp)}</strong></div>
|
||||
<div><span>Network</span><strong>${escapeHtml(networkName)}</strong></div>
|
||||
<div><span>Uptime</span><strong>${escapeHtml(uptime)}</strong></div>
|
||||
<div><span>CPU temp</span><strong>${escapeHtml(cpu)}</strong></div>
|
||||
</div>
|
||||
|
||||
@@ -694,6 +694,57 @@ def test_cpu_temp_reader_ignores_non_cpu_thermal_zones(tmp_path):
|
||||
assert utilities._read_cpu_temp_c(tmp_path) == 61
|
||||
|
||||
|
||||
def test_network_name_uses_wifi_ssid(monkeypatch):
|
||||
monkeypatch.setattr(utilities, "HARDWARE", SimpleNamespace(get_network_type=lambda: 1))
|
||||
monkeypatch.setattr(utilities, "_read_active_wifi_ssid", lambda: "Garage Wi-Fi")
|
||||
utilities._NETWORK_STATUS_CACHE.update({"updated_at": 0.0, "value": None})
|
||||
|
||||
assert utilities.get_current_network_name() == "Garage Wi-Fi"
|
||||
|
||||
|
||||
def test_wifi_ssid_reader_uses_active_nmcli_row(monkeypatch):
|
||||
result = SimpleNamespace(
|
||||
returncode=0,
|
||||
stdout=" :Nearby Network\n*:Garage:Main Wi-Fi\n",
|
||||
)
|
||||
monkeypatch.setattr(utilities.subprocess, "run", lambda *args, **kwargs: result)
|
||||
|
||||
assert utilities._read_active_wifi_ssid() == "Garage:Main Wi-Fi"
|
||||
|
||||
|
||||
def test_network_name_labels_cellular_generations(monkeypatch):
|
||||
network_type = {"value": 4}
|
||||
monkeypatch.setattr(utilities, "HARDWARE", SimpleNamespace(get_network_type=lambda: network_type["value"]))
|
||||
monkeypatch.setattr(utilities, "_read_active_wifi_ssid", lambda: None)
|
||||
|
||||
utilities._NETWORK_STATUS_CACHE.update({"updated_at": 0.0, "value": None})
|
||||
assert utilities.get_current_network_name() == "Cellular (LTE)"
|
||||
|
||||
network_type["value"] = 5
|
||||
utilities._NETWORK_STATUS_CACHE.update({"updated_at": 0.0, "value": None})
|
||||
assert utilities.get_current_network_name() == "Cellular (5G)"
|
||||
|
||||
|
||||
def test_network_name_reports_no_wireless_connectivity(monkeypatch):
|
||||
monkeypatch.setattr(utilities, "HARDWARE", SimpleNamespace(get_network_type=lambda: 0))
|
||||
monkeypatch.setattr(utilities, "_read_active_wifi_ssid", lambda: None)
|
||||
utilities._NETWORK_STATUS_CACHE.update({"updated_at": 0.0, "value": None})
|
||||
|
||||
assert utilities.get_current_network_name() == "No wireless connectivity"
|
||||
|
||||
|
||||
def test_device_summary_includes_network_name(monkeypatch):
|
||||
monkeypatch.setattr(utilities, "_read_uptime_seconds", lambda: 120)
|
||||
monkeypatch.setattr(utilities, "_read_cpu_temp_c", lambda: 55)
|
||||
monkeypatch.setattr(utilities, "get_current_lan_ip", lambda: "192.168.1.10")
|
||||
monkeypatch.setattr(utilities, "get_current_network_name", lambda: "Home Network")
|
||||
|
||||
summary = utilities._build_device_summary(FakeParams({"IsOnroad": False}))
|
||||
|
||||
assert summary["networkName"] == "Home Network"
|
||||
assert summary["lanIp"] == "192.168.1.10"
|
||||
|
||||
|
||||
def test_persistent_loader_accepts_decoded_param_dict():
|
||||
params = FakeParams({
|
||||
utilities.DASHBOARD_PERSISTENT_STATS_PARAM: {
|
||||
|
||||
@@ -21,6 +21,7 @@ from urllib.parse import quote
|
||||
|
||||
from openpilot.common.constants import CV
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.system.hardware import HARDWARE
|
||||
from openpilot.system.loggerd.config import get_available_bytes, get_used_bytes
|
||||
from openpilot.system.loggerd.deleter import PRESERVE_ATTR_NAME, PRESERVE_ATTR_VALUE
|
||||
from openpilot.system.loggerd.uploader import listdir_by_creation
|
||||
@@ -78,6 +79,7 @@ DASHBOARD_ANALYZER_STATUS_PATH = Path("/tmp/galaxy_dashboard_analyzer_status.jso
|
||||
DASHBOARD_ANALYZER_STATUS_MAX_AGE_SECONDS = 30 * 60
|
||||
DASHBOARD_TOP_MODEL_LIMIT = 3
|
||||
LAN_IP_CACHE_TTL_SECONDS = 10.0
|
||||
NETWORK_STATUS_CACHE_TTL_SECONDS = 10.0
|
||||
DASHBOARD_EVENT_DISTRACTED = "driverDistracted2"
|
||||
DASHBOARD_EVENT_UNRESPONSIVE = "driverUnresponsive3"
|
||||
DASHBOARD_TIME_SOURCE_LOG = "log"
|
||||
@@ -111,6 +113,10 @@ _LAN_IP_CACHE = {
|
||||
"updated_at": 0.0,
|
||||
"value": None,
|
||||
}
|
||||
_NETWORK_STATUS_CACHE = {
|
||||
"updated_at": 0.0,
|
||||
"value": None,
|
||||
}
|
||||
_DASHBOARD_ANALYZER_LOCK = threading.Lock()
|
||||
_DASHBOARD_ANALYZER_PROCESS = None
|
||||
params = Params(return_defaults=True)
|
||||
@@ -206,6 +212,100 @@ def get_current_lan_ip():
|
||||
return None
|
||||
|
||||
|
||||
def _clean_network_name(value):
|
||||
text = "".join(character for character in str(value or "").strip().strip("\x00") if character.isprintable())
|
||||
return text[:128]
|
||||
|
||||
|
||||
def _read_active_wifi_ssid():
|
||||
commands = (
|
||||
(["nmcli", "-t", "--escape", "no", "-f", "IN-USE,SSID", "device", "wifi", "list", "--rescan", "no"], "nmcli"),
|
||||
(["iwgetid", "--raw"], "iwgetid"),
|
||||
(["iw", "dev", "wlan0", "link"], "iw"),
|
||||
)
|
||||
|
||||
for command, source in commands:
|
||||
try:
|
||||
result = subprocess.run(command, check=False, capture_output=True, text=True, timeout=0.5)
|
||||
except Exception:
|
||||
continue
|
||||
if result.returncode != 0:
|
||||
continue
|
||||
|
||||
if source == "nmcli":
|
||||
for line in result.stdout.splitlines():
|
||||
in_use, separator, ssid = line.partition(":")
|
||||
if separator and in_use.strip() == "*":
|
||||
name = _clean_network_name(ssid)
|
||||
if name:
|
||||
return name
|
||||
elif source == "iwgetid":
|
||||
name = _clean_network_name(result.stdout)
|
||||
if name:
|
||||
return name
|
||||
else:
|
||||
for line in result.stdout.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("SSID:"):
|
||||
name = _clean_network_name(stripped.partition(":")[2])
|
||||
if name:
|
||||
return name
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _network_type_name(network_type):
|
||||
raw_value = getattr(network_type, "raw", network_type)
|
||||
try:
|
||||
raw_value = int(raw_value)
|
||||
except (TypeError, ValueError):
|
||||
text = str(network_type or "").strip().lower()
|
||||
for name in ("none", "wifi", "cell2g", "cell3g", "cell4g", "cell5g", "ethernet"):
|
||||
if text == name or text.endswith(f".{name}"):
|
||||
return name
|
||||
return "unknown"
|
||||
|
||||
return {
|
||||
0: "none",
|
||||
1: "wifi",
|
||||
2: "cell2g",
|
||||
3: "cell3g",
|
||||
4: "cell4g",
|
||||
5: "cell5g",
|
||||
6: "ethernet",
|
||||
}.get(raw_value, "unknown")
|
||||
|
||||
|
||||
def get_current_network_name():
|
||||
now = time.monotonic()
|
||||
if now - _NETWORK_STATUS_CACHE["updated_at"] < NETWORK_STATUS_CACHE_TTL_SECONDS:
|
||||
return _NETWORK_STATUS_CACHE["value"]
|
||||
|
||||
try:
|
||||
network_type = _network_type_name(HARDWARE.get_network_type())
|
||||
except Exception:
|
||||
network_type = "unknown"
|
||||
|
||||
if network_type == "wifi":
|
||||
value = _read_active_wifi_ssid() or "Wi-Fi"
|
||||
elif network_type == "cell2g":
|
||||
value = "Cellular (2G)"
|
||||
elif network_type == "cell3g":
|
||||
value = "Cellular (3G)"
|
||||
elif network_type == "cell4g":
|
||||
value = "Cellular (LTE)"
|
||||
elif network_type == "cell5g":
|
||||
value = "Cellular (5G)"
|
||||
elif network_type == "ethernet":
|
||||
value = "Ethernet"
|
||||
else:
|
||||
value = _read_active_wifi_ssid() or "No wireless connectivity"
|
||||
|
||||
_NETWORK_STATUS_CACHE["updated_at"] = time.monotonic()
|
||||
_NETWORK_STATUS_CACHE["value"] = value
|
||||
return value
|
||||
|
||||
|
||||
def secure_filename(filename):
|
||||
safe = os.path.basename(str(filename or ""))
|
||||
safe = safe.replace(" ", "_").strip()
|
||||
@@ -2508,12 +2608,14 @@ def _build_device_summary(params_obj):
|
||||
uptime_seconds = _read_uptime_seconds()
|
||||
cpu_temp_c = _read_cpu_temp_c()
|
||||
lan_ip = get_current_lan_ip()
|
||||
network_name = get_current_network_name()
|
||||
return {
|
||||
"status": "Driving" if is_onroad else "Parked",
|
||||
"online": True,
|
||||
"uptimeSeconds": uptime_seconds,
|
||||
"cpuTempC": cpu_temp_c,
|
||||
"lanIp": lan_ip,
|
||||
"networkName": network_name,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user