skills/officialpm_my-tesla
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
officialpm_my-tesla/scripts/tesla.py

2313 lines
78 KiB
Python
Raw Permalink Normal View History

Initial commit with translated description
2026-03-29 13:13:35 +08:00
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "teslapy>=2.0.0",
# ]
# ///
"""
Tesla vehicle control via unofficial API.
Supports multiple vehicles.
"""
import argparse
import json
import os
import sys
import sqlite3
import time
import traceback
from datetime import datetime, timezone
from pathlib import Path
# Keep the repo clean: don't write __pycache__/ bytecode files when running the CLI.
# (Also helps keep private repos from accumulating noisy artifacts.)
sys.dont_write_bytecode = True
CACHE_FILE = Path.home() / ".tesla_cache.json"
DEFAULTS_FILE = Path.home() / ".my_tesla.json"
SKILL_DIR = Path(__file__).resolve().parent.parent
def _invocation(extra: str = "") -> str:
"""Return a copy/pastable invocation string for help/error messages.
Use the absolute path to this script so the suggestion works even when the
current working directory is not the repo root.
"""
prog = str(Path(__file__).resolve())
return f"python3 {prog}{(' ' + extra) if extra else ''}"
def read_skill_version() -> str:
"""Read the skill version from VERSION.txt/VERSION in the repo.
ClawdHub ignores extensionless files like `VERSION`, so published artifacts
also include `VERSION.txt`. Prefer VERSION.txt when present.
"""
for name in ("VERSION.txt", "VERSION"):
p = SKILL_DIR / name
try:
if p.exists():
v = p.read_text().strip()
if v:
return v
except Exception:
continue
return "(unknown)"
def resolve_email(args, prompt: bool = True) -> str:
"""Resolve Tesla account email from args/env, optionally prompting."""
email = getattr(args, "email", None) or os.environ.get("TESLA_EMAIL")
if isinstance(email, str) and email.strip():
return email.strip()
if not prompt:
return None
return input("Tesla email: ").strip()
def require_email(args) -> str:
"""Require a Tesla email to be provided via --email or TESLA_EMAIL."""
email = resolve_email(args, prompt=False)
if not email:
print(
"❌ Missing Tesla email. Set TESLA_EMAIL or pass --email\n"
f" Example: TESLA_EMAIL=\"you@email.com\" {_invocation('list')}",
file=sys.stderr,
)
sys.exit(2)
return email
def get_tesla(email: str):
"""Get authenticated Tesla instance."""
import teslapy
def custom_auth(url):
print(f"\n🔐 Open this URL in your browser:\n{url}\n")
print("Log in to Tesla, then paste the final URL here")
print("(it will start with https://auth.tesla.com/void/callback?...)")
return input("\nCallback URL: ").strip()
tesla = teslapy.Tesla(email, authenticator=custom_auth, cache_file=str(CACHE_FILE))
if not tesla.authorized:
tesla.fetch_token()
print("✅ Authenticated successfully!")
# Best-effort: keep the local OAuth cache file private.
if CACHE_FILE.exists():
_chmod_0600(CACHE_FILE)
return tesla
def load_defaults():
"""Load optional user defaults from ~/.my_tesla.json (local only)."""
try:
if DEFAULTS_FILE.exists():
return json.loads(DEFAULTS_FILE.read_text())
except Exception:
pass
return {}
def _chmod_0600(path: Path):
"""Best-effort: set file permissions to user read/write only."""
try:
path.chmod(0o600)
except Exception:
# Non-POSIX FS or permission error; ignore.
pass
def save_defaults(obj: dict):
# Defaults can include human-readable vehicle names; keep them private.
DEFAULTS_FILE.write_text(json.dumps(obj, indent=2) + "\n")
_chmod_0600(DEFAULTS_FILE)
def resolve_default_car_name():
# Highest priority: env var
env_name = os.environ.get("MY_TESLA_DEFAULT_CAR")
if env_name:
return env_name.strip()
defaults = load_defaults()
name = defaults.get("default_car")
return name.strip() if isinstance(name, str) and name.strip() else None
def _select_vehicle(vehicles, target_name: str):
"""Select a vehicle from a list by name (exact/partial) or 1-based index.
- Exact match is case-insensitive.
- If no exact match, a case-insensitive *substring* match is attempted.
- If target_name is a digit (e.g., "1"), it's treated as a 1-based index.
"""
if not vehicles:
return None
if not target_name:
return vehicles[0]
s = target_name.strip()
if s.isdigit():
idx = int(s) - 1
if 0 <= idx < len(vehicles):
return vehicles[idx]
return None
s_l = s.lower()
# 1) Exact match (case-insensitive)
for v in vehicles:
if v.get('display_name', '').lower() == s_l:
return v
# 2) Substring match (case-insensitive)
matches = [v for v in vehicles if s_l in v.get('display_name', '').lower()]
if len(matches) == 1:
return matches[0]
# Ambiguous / not found
return None
def get_vehicle(tesla, name: str = None):
"""Get vehicle by name/index, else default car, else first vehicle."""
vehicles = tesla.vehicle_list()
if not vehicles:
print("❌ No vehicles found on this account", file=sys.stderr)
sys.exit(1)
target_name = name or resolve_default_car_name()
if target_name:
selected = _select_vehicle(vehicles, target_name)
if selected:
return selected
# Give a more helpful error (and show numeric indices too).
s = str(target_name).strip()
ambiguous = False
matches = []
if s and not s.isdigit():
s_l = s.lower()
matches = [
(i + 1, v) for i, v in enumerate(vehicles)
if s_l in v.get('display_name', '').lower()
]
ambiguous = len(matches) > 1
options = "\n".join(
f" {i+1}. {v.get('display_name')}" for i, v in enumerate(vehicles)
)
if ambiguous:
match_lines = "\n".join(
f" {idx}. {v.get('display_name')}" for idx, v in matches
)
print(
f"❌ Vehicle '{target_name}' is ambiguous (matched multiple vehicles).\n"
" Tip: use a more specific name, or choose by index: --car <N>\n"
f"Matches:\n{match_lines}\n\n"
f"All vehicles:\n{options}",
file=sys.stderr,
)
else:
print(
f"❌ Vehicle '{target_name}' not found.\n"
" Tip: you can pass --car with a partial name (substring match) or a 1-based index.\n"
f"Available vehicles:\n{options}",
file=sys.stderr,
)
sys.exit(1)
return vehicles[0]
def wake_vehicle(vehicle, allow_wake: bool = True) -> bool:
"""Wake vehicle if asleep.
Returns True if the vehicle is (or becomes) online.
If allow_wake is False and the vehicle is not online, returns False.
"""
state = vehicle.get('state')
if state == 'online':
return True
if not allow_wake:
return False
try:
print("⏳ Waking vehicle...", file=sys.stderr)
vehicle.sync_wake_up()
return True
except Exception as e:
print(
f"❌ Failed to wake vehicle (state was: {state}). Try again, or run: {_invocation('wake')}\n"
f" Details: {e}",
file=sys.stderr,
)
sys.exit(1)
def cmd_auth(args):
"""Authenticate with Tesla."""
email = resolve_email(args)
if not email:
print("❌ Missing Tesla email. Set TESLA_EMAIL or pass --email", file=sys.stderr)
sys.exit(2)
tesla = get_tesla(email)
vehicles = tesla.vehicle_list()
print(f"\n✅ Authentication cached at {CACHE_FILE}")
print(f"\n🚗 Found {len(vehicles)} vehicle(s):")
for v in vehicles:
# Avoid printing VINs by default.
print(f" - {v['display_name']} ({v['state']})")
def cmd_list(args):
"""List all vehicles."""
tesla = get_tesla(require_email(args))
vehicles = tesla.vehicle_list()
default_name = resolve_default_car_name()
if getattr(args, "json", False):
# Keep JSON output small + privacy-safe (no VINs).
out = []
for i, v in enumerate(vehicles):
name = v.get('display_name')
out.append({
'index': i + 1,
'display_name': name,
'state': v.get('state'),
'is_default': bool(default_name and isinstance(name, str) and name.lower() == default_name.lower()),
})
print(json.dumps({'vehicles': out, 'default_car': default_name}, indent=2))
return
print(f"Found {len(vehicles)} vehicle(s):\n")
for i, v in enumerate(vehicles):
star = " (default)" if default_name and v['display_name'].lower() == default_name.lower() else ""
print(f"{i+1}. {v['display_name']}{star}")
# Avoid printing VIN in normal output (privacy).
print(f" State: {v['state']}")
print()
if default_name:
print(f"Default car: {default_name}")
else:
print(f"Default car: (none) — set with: {_invocation('default-car "Name"')}")
def _c_to_f(c):
try:
return c * 9 / 5 + 32
except Exception:
return None
def _fmt_bool(b, yes="Yes", no="No"):
return yes if b else no
def _short_status(vehicle, data):
charge = data.get('charge_state', {})
climate = data.get('climate_state', {})
vs = data.get('vehicle_state', {})
batt = charge.get('battery_level')
rng = charge.get('battery_range')
charging = charge.get('charging_state')
locked = vs.get('locked')
inside_c = climate.get('inside_temp')
inside_f = _c_to_f(inside_c) if inside_c is not None else None
climate_on = climate.get('is_climate_on')
parts = [f"🚗 {vehicle['display_name']}"]
if locked is not None:
parts.append(f"🔒 {_fmt_bool(locked, 'Locked', 'Unlocked')}")
if batt is not None:
if rng is not None:
parts.append(f"🔋 {batt}% ({rng:.0f} mi)")
else:
parts.append(f"🔋 {batt}%")
if charging:
parts.append(f"⚡ {charging}")
if inside_c is not None and inside_f is not None:
parts.append(f"🌡️ {inside_f:.0f}°F")
if climate_on is not None:
parts.append(f"❄️ {_fmt_bool(climate_on, 'On', 'Off')}")
return " • ".join(parts)
def _summary_json(vehicle, data: dict) -> dict:
"""Sanitized, machine-readable one-line summary.
Unlike `status --json`, this does NOT emit raw vehicle_data (which may include location).
"""
charge = data.get('charge_state', {})
climate = data.get('climate_state', {})
vs = data.get('vehicle_state', {})
inside_c = climate.get('inside_temp')
inside_f = _c_to_f(inside_c) if inside_c is not None else None
out = {
"vehicle": {
"display_name": vehicle.get("display_name"),
"state": vehicle.get("state"),
},
"summary": _short_status(vehicle, data),
"security": {
"locked": vs.get("locked"),
},
"battery": {
"level_percent": charge.get("battery_level"),
"range_mi": charge.get("battery_range"),
"usable_level_percent": charge.get("usable_battery_level"),
},
"charging": {
"charging_state": charge.get("charging_state"),
},
"climate": {
"inside_temp_c": inside_c,
"inside_temp_f": inside_f,
"is_climate_on": climate.get("is_climate_on"),
},
}
# Drop empty nested dicts / None values.
for k in list(out.keys()):
v = out[k]
if isinstance(v, dict):
v2 = {kk: vv for kk, vv in v.items() if vv is not None}
if v2:
out[k] = v2
else:
del out[k]
elif v is None:
del out[k]
return out
def _fmt_temp_pair(c):
if c is None:
return None
f = _c_to_f(c)
if f is None:
return None
return f"{c}°C ({f:.0f}°F)"
def _bar_to_psi(bar):
"""Convert bar to PSI.
Tesla APIs commonly return tire pressures in bar.
"""
try:
return float(bar) * 14.5037738
except Exception:
return None
def _fmt_tire_pressure(bar):
"""Format tire pressure as "X.X bar (Y psi)"."""
if bar is None:
return None
try:
b = float(bar)
except Exception:
return None
psi = _bar_to_psi(b)
if psi is None:
return None
return f"{b:.2f} bar ({psi:.0f} psi)"
def _fmt_minutes_hhmm(minutes):
"""Format minutes-from-midnight as HH:MM.
Tesla endpoints commonly represent scheduled times as minutes after midnight.
"""
try:
m = int(minutes)
except Exception:
return None
if m < 0:
return None
hh = (m // 60) % 24
mm = m % 60
return f"{hh:02d}:{mm:02d}"
def _report(vehicle, data):
"""One-screen status report (safe for chat)."""
charge = data.get('charge_state', {})
climate = data.get('climate_state', {})
vs = data.get('vehicle_state', {})
lines = []
lines.append(f"🚗 {vehicle['display_name']}")
lines.append(f"State: {vehicle.get('state')}")
locked = vs.get('locked')
if locked is not None:
lines.append(f"Locked: {_fmt_bool(locked, 'Yes', 'No')}")
sentry = vs.get('sentry_mode')
if sentry is not None:
lines.append(f"Sentry: {_fmt_bool(sentry, 'On', 'Off')}")
openings = _openings_one_line(vs)
if openings:
lines.append(f"Openings: {openings}")
batt = charge.get('battery_level')
usable = charge.get('usable_battery_level')
rng = charge.get('battery_range')
if batt is not None and rng is not None:
lines.append(f"Battery: {batt}% ({rng:.0f} mi)")
elif batt is not None:
lines.append(f"Battery: {batt}%")
# Some vehicles report usable battery level separately (helpful for health/degradation).
if usable is not None:
try:
lines.append(f"Usable battery: {int(usable)}%")
except Exception:
lines.append(f"Usable battery: {usable}%")
charging_state = charge.get('charging_state')
if charging_state is not None:
extra = []
limit = charge.get('charge_limit_soc')
if limit is not None:
extra.append(f"limit {limit}%")
if charging_state == 'Charging':
ttf = charge.get('time_to_full_charge')
if ttf is not None:
extra.append(f"{ttf:.1f}h to full")
rate = charge.get('charge_rate')
if rate is not None:
extra.append(f"{rate} mph")
suffix = f" ({', '.join(extra)})" if extra else ""
lines.append(f"Charging: {charging_state}{suffix}")
# When actively charging, show power details if available.
# This is useful to sanity-check a slow/fast charge session at a glance.
if charging_state == 'Charging':
p = charge.get('charger_power')
v = charge.get('charger_voltage')
a = charge.get('charger_actual_current')
bits = []
if p is not None:
bits.append(f"{p} kW")
if v is not None:
bits.append(f"{v}V")
if a is not None:
bits.append(f"{a}A")
if bits:
lines.append(f"Charging power: {' '.join(bits)}")
# Charge port / cable state
cpd = charge.get('charge_port_door_open')
if cpd is not None:
lines.append(f"Charge port door: {_fmt_bool(cpd, 'Open', 'Closed')}")
cable = charge.get('conn_charge_cable')
if cable is not None:
lines.append(f"Charge cable: {cable}")
sched_time = charge.get('scheduled_charging_start_time')
sched_mode = charge.get('scheduled_charging_mode')
sched_pending = charge.get('scheduled_charging_pending')
if sched_time is not None or sched_mode is not None or sched_pending is not None:
bits = []
if isinstance(sched_mode, str) and sched_mode.strip():
bits.append(sched_mode.strip())
elif sched_pending is not None:
bits.append('On' if sched_pending else 'Off')
hhmm = _fmt_minutes_hhmm(sched_time)
if hhmm:
bits.append(hhmm)
if bits:
lines.append(f"Scheduled charging: {' '.join(bits)}")
# Scheduled departure / off-peak charging (read-only)
dep_enabled = charge.get('scheduled_departure_enabled')
dep_time = charge.get('scheduled_departure_time')
precond = charge.get('preconditioning_enabled')
off_peak = charge.get('off_peak_charging_enabled')
if dep_enabled is not None or dep_time is not None or precond is not None or off_peak is not None:
bits = []
if dep_enabled is not None:
bits.append('On' if dep_enabled else 'Off')
hhmm = _fmt_minutes_hhmm(dep_time)
if hhmm:
bits.append(hhmm)
if precond is not None:
bits.append(f"precond {'On' if precond else 'Off'}")
if off_peak is not None:
bits.append(f"off-peak {'On' if off_peak else 'Off'}")
if bits:
lines.append(f"Scheduled departure: {' '.join(bits)}")
inside = _fmt_temp_pair(climate.get('inside_temp'))
outside = _fmt_temp_pair(climate.get('outside_temp'))
if inside:
lines.append(f"Inside: {inside}")
if outside:
lines.append(f"Outside: {outside}")
climate_on = climate.get('is_climate_on')
if climate_on is not None:
lines.append(f"Climate: {_fmt_bool(climate_on, 'On', 'Off')}")
heaters = _seat_heater_fields(climate)
if heaters:
lines.append(f"Seat heaters: {_seat_heaters_one_line(heaters)}")
# Tire pressures (TPMS) if available
fl = _fmt_tire_pressure(vs.get('tpms_pressure_fl'))
fr = _fmt_tire_pressure(vs.get('tpms_pressure_fr'))
rl = _fmt_tire_pressure(vs.get('tpms_pressure_rl'))
rr = _fmt_tire_pressure(vs.get('tpms_pressure_rr'))
if any([fl, fr, rl, rr]):
lines.append(
"Tires (TPMS): "
f"FL {fl or '(?)'} | FR {fr or '(?)'} | RL {rl or '(?)'} | RR {rr or '(?)'}"
)
odo = vs.get('odometer')
if odo is not None:
lines.append(f"Odometer: {odo:.0f} mi")
return "\n".join(lines)
def _report_json(vehicle, data: dict) -> dict:
"""Sanitized JSON equivalent of `_report`.
Intentionally excludes location/drive_state.
"""
charge = data.get('charge_state', {})
climate = data.get('climate_state', {})
vs = data.get('vehicle_state', {})
out = {
"vehicle": {
"display_name": vehicle.get('display_name'),
"state": vehicle.get('state'),
},
"battery": {
"level_percent": charge.get('battery_level'),
"range_mi": charge.get('battery_range'),
"usable_battery_level_percent": charge.get('usable_battery_level'),
},
"charging": {
"charging_state": charge.get('charging_state'),
"charge_limit_percent": charge.get('charge_limit_soc'),
"minutes_to_full_charge": charge.get('minutes_to_full_charge'),
"time_to_full_charge_hours": charge.get('time_to_full_charge'),
"charge_rate_mph": charge.get('charge_rate'),
"charger_power_kw": charge.get('charger_power'),
"charger_voltage_v": charge.get('charger_voltage'),
"charger_actual_current_a": charge.get('charger_actual_current'),
"charge_current_request_a": charge.get('charge_current_request'),
"charge_current_request_max_a": charge.get('charge_current_request_max'),
"charging_amps": charge.get('charging_amps'),
"charge_port_door_open": charge.get('charge_port_door_open'),
"conn_charge_cable": charge.get('conn_charge_cable'),
},
"scheduled_charging": {
"mode": charge.get('scheduled_charging_mode'),
"pending": charge.get('scheduled_charging_pending'),
"start_time_hhmm": _fmt_minutes_hhmm(charge.get('scheduled_charging_start_time')),
},
"scheduled_departure": {
"enabled": charge.get('scheduled_departure_enabled'),
"time_hhmm": _fmt_minutes_hhmm(charge.get('scheduled_departure_time')),
"preconditioning_enabled": charge.get('preconditioning_enabled'),
"off_peak_charging_enabled": charge.get('off_peak_charging_enabled'),
},
"climate": {
"inside_temp_c": climate.get('inside_temp'),
"outside_temp_c": climate.get('outside_temp'),
"is_climate_on": climate.get('is_climate_on'),
"seat_heaters": _seat_heater_fields(climate) or None,
},
"security": {
"locked": vs.get('locked'),
"sentry_mode": vs.get('sentry_mode'),
},
"openings": _openings_json(vs),
"tpms": {
"pressure_fl": vs.get('tpms_pressure_fl'),
"pressure_fr": vs.get('tpms_pressure_fr'),
"pressure_rl": vs.get('tpms_pressure_rl'),
"pressure_rr": vs.get('tpms_pressure_rr'),
},
"odometer_mi": vs.get('odometer'),
}
# Drop empty nested dicts for cleaner output.
for k in list(out.keys()):
v = out[k]
if isinstance(v, dict):
v2 = {kk: vv for kk, vv in v.items() if vv is not None}
if v2:
out[k] = v2
else:
del out[k]
elif v is None:
del out[k]
return out
def _ensure_online_or_exit(vehicle, allow_wake: bool):
if wake_vehicle(vehicle, allow_wake=allow_wake):
return
state = vehicle.get('state')
name = vehicle.get('display_name', 'Vehicle')
print(
f"ℹ️ {name} is currently '{state}'. Skipping wake because --no-wake was set.\n"
f" Re-run without --no-wake, or run: {_invocation('wake')}",
file=sys.stderr,
)
sys.exit(3)
def cmd_report(args):
"""One-screen status report."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
_ensure_online_or_exit(vehicle, allow_wake=not getattr(args, 'no_wake', False))
data = vehicle.get_vehicle_data()
if args.json:
# Default JSON output is a structured, sanitized report object.
# Use --raw-json if you explicitly want the full vehicle_data payload
# (which may include location/drive_state).
if getattr(args, "raw_json", False):
print(json.dumps(data, indent=2))
else:
print(json.dumps(_report_json(vehicle, data), indent=2))
return
print(_report(vehicle, data))
def cmd_status(args):
"""Get vehicle status."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
_ensure_online_or_exit(vehicle, allow_wake=not getattr(args, 'no_wake', False))
data = vehicle.get_vehicle_data()
# When --json is requested, print *only* JSON (no extra human text), so it can
# be reliably piped/parsed.
if args.json:
print(json.dumps(data, indent=2))
return
charge = data.get('charge_state', {})
climate = data.get('climate_state', {})
vehicle_state = data.get('vehicle_state', {})
if getattr(args, 'summary', False):
# Print a one-line summary *in addition* to the detailed view.
# (If you only want the one-liner, use the `summary` command.)
print(_short_status(vehicle, data))
print()
# Human-friendly detailed view
print(f"🚗 {vehicle['display_name']}")
print(f" State: {vehicle.get('state')}")
batt = charge.get('battery_level')
rng = charge.get('battery_range')
if batt is not None and rng is not None:
print(f" Battery: {batt}% ({rng:.0f} mi)")
elif batt is not None:
print(f" Battery: {batt}%")
charging_state = charge.get('charging_state')
if charging_state is not None:
print(f" Charging: {charging_state}")
inside_c = climate.get('inside_temp')
outside_c = climate.get('outside_temp')
if inside_c is not None:
inside_f = _c_to_f(inside_c)
if inside_f is not None:
print(f" Inside temp: {inside_c}°C ({inside_f:.0f}°F)")
if outside_c is not None:
outside_f = _c_to_f(outside_c)
if outside_f is not None:
print(f" Outside temp: {outside_c}°C ({outside_f:.0f}°F)")
climate_on = climate.get('is_climate_on')
if climate_on is not None:
print(f" Climate on: {climate_on}")
locked = vehicle_state.get('locked')
if locked is not None:
print(f" Locked: {locked}")
odo = vehicle_state.get('odometer')
if odo is not None:
print(f" Odometer: {odo:.0f} mi")
def cmd_lock(args):
"""Lock the vehicle."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
wake_vehicle(vehicle)
vehicle.command('LOCK')
print(f"🔒 {vehicle['display_name']} locked")
def cmd_unlock(args):
"""Unlock the vehicle."""
require_yes(args, 'unlock')
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
wake_vehicle(vehicle)
vehicle.command('UNLOCK')
print(f"🔓 {vehicle['display_name']} unlocked")
def cmd_climate(args):
"""Control climate.
Actions:
- status (read-only)
- on/off
- temp <value> [--celsius|--fahrenheit]
- defrost <on|off> (max defrost / preconditioning)
"""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
if args.action == 'status':
# Read-only action can skip waking the car.
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
climate = data.get('climate_state', {})
out = {
'is_climate_on': climate.get('is_climate_on'),
'inside_temp_c': climate.get('inside_temp'),
'outside_temp_c': climate.get('outside_temp'),
'driver_temp_setting_c': climate.get('driver_temp_setting'),
'passenger_temp_setting_c': climate.get('passenger_temp_setting'),
}
if args.json:
print(json.dumps(out, indent=2))
return
inside = _fmt_temp_pair(climate.get('inside_temp'))
outside = _fmt_temp_pair(climate.get('outside_temp'))
print(f"🚗 {vehicle['display_name']}")
if out.get('is_climate_on') is not None:
print(f"Climate: {_fmt_bool(out.get('is_climate_on'), 'On', 'Off')}")
if inside:
print(f"Inside: {inside}")
if outside:
print(f"Outside: {outside}")
driver = _fmt_temp_pair(climate.get('driver_temp_setting'))
passenger = _fmt_temp_pair(climate.get('passenger_temp_setting'))
if driver or passenger:
print(f"Setpoint: driver {driver or '(unknown)'} | passenger {passenger or '(unknown)'}")
return
# Mutating actions (wake is allowed)
wake_vehicle(vehicle)
if args.action == 'on':
vehicle.command('CLIMATE_ON')
print(f"❄️ {vehicle['display_name']} climate turned on")
elif args.action == 'off':
vehicle.command('CLIMATE_OFF')
print(f"🌡️ {vehicle['display_name']} climate turned off")
elif args.action == 'temp':
if args.value is None:
raise ValueError("Missing temperature value (e.g., climate temp 72 or climate temp 22 --celsius)")
value = float(args.value)
# Default is Fahrenheit unless --celsius is provided.
in_f = True
if getattr(args, "celsius", False):
in_f = False
elif getattr(args, "fahrenheit", False):
in_f = True
temp_c = (value - 32) * 5 / 9 if in_f else value
vehicle.command('CHANGE_CLIMATE_TEMPERATURE_SETTING', driver_temp=temp_c, passenger_temp=temp_c)
print(f"🌡️ {vehicle['display_name']} temperature set to {value:g}°{'F' if in_f else 'C'}")
elif args.action == 'defrost':
if args.value is None or str(args.value).strip().lower() not in ('on', 'off'):
raise ValueError("Missing defrost value. Use: climate defrost on|off")
on = str(args.value).strip().lower() == 'on'
vehicle.command('SET_PRECONDITIONING_MAX', on=on)
print(f"🧊 {vehicle['display_name']} max defrost {('enabled' if on else 'disabled')}")
else:
raise ValueError(f"Unknown action: {args.action}")
def _charge_status_json(charge: dict) -> dict:
"""Small, privacy-safe charging status object.
Intended for piping/parsing via `charge status --json`.
"""
charge = charge or {}
return {
'battery_level': charge.get('battery_level'),
'battery_range': charge.get('battery_range'),
'usable_battery_level': charge.get('usable_battery_level'),
'charging_state': charge.get('charging_state'),
'charge_limit_soc': charge.get('charge_limit_soc'),
'time_to_full_charge': charge.get('time_to_full_charge'),
'charge_rate': charge.get('charge_rate'),
'charger_power': charge.get('charger_power'),
'charger_voltage': charge.get('charger_voltage'),
'charger_actual_current': charge.get('charger_actual_current'),
'scheduled_charging_start_time': charge.get('scheduled_charging_start_time'),
'scheduled_charging_mode': charge.get('scheduled_charging_mode'),
'scheduled_charging_pending': charge.get('scheduled_charging_pending'),
'charge_port_door_open': charge.get('charge_port_door_open'),
'conn_charge_cable': charge.get('conn_charge_cable'),
}
def cmd_charge(args):
"""Control charging."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
# Read-only action can skip waking the car.
allow_wake = True
if args.action == 'status':
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
if args.action == 'status':
data = vehicle.get_vehicle_data()
charge = data['charge_state']
if args.json:
# Print *only* JSON (no extra human text) so it can be piped/parsed.
# Keep it focused to avoid leaking unrelated vehicle details.
out = _charge_status_json(charge)
# Drop nulls for cleanliness.
out = {k: v for k, v in out.items() if v is not None}
print(json.dumps(out, indent=2))
return
print(f"🔋 {vehicle['display_name']} Battery: {charge['battery_level']}%")
print(f" Range: {charge['battery_range']:.0f} mi")
usable = charge.get('usable_battery_level')
if usable is not None:
try:
print(f" Usable: {int(usable)}%")
except Exception:
print(f" Usable: {usable}%")
print(f" State: {charge['charging_state']}")
print(f" Limit: {charge['charge_limit_soc']}%")
cpd = charge.get('charge_port_door_open')
if cpd is not None:
print(f" Charge port door: {_fmt_bool(cpd, 'Open', 'Closed')}")
cable = charge.get('conn_charge_cable')
if cable is not None:
print(f" Charge cable: {cable}")
if charge['charging_state'] == 'Charging':
if charge.get('time_to_full_charge') is not None:
print(f" Time left: {charge['time_to_full_charge']:.1f} hrs")
if charge.get('charge_rate') is not None:
print(f" Rate: {charge['charge_rate']} mph")
# Power details when available
p = charge.get('charger_power')
v = charge.get('charger_voltage')
a = charge.get('charger_actual_current')
bits = []
if p is not None:
bits.append(f"{p} kW")
if v is not None:
bits.append(f"{v}V")
if a is not None:
bits.append(f"{a}A")
if bits:
print(f" Power: {' '.join(bits)}")
return
if args.action == 'start':
require_yes(args, 'charge start')
vehicle.command('START_CHARGE')
print(f"⚡ {vehicle['display_name']} charging started")
return
if args.action == 'stop':
require_yes(args, 'charge stop')
vehicle.command('STOP_CHARGE')
print(f"🛑 {vehicle['display_name']} charging stopped")
return
if args.action == 'limit':
require_yes(args, 'charge limit')
if args.value is None:
raise ValueError("Missing charge limit percent (e.g., charge limit 80)")
pct = int(args.value)
if pct < 50 or pct > 100:
raise ValueError("Invalid charge limit percent. Expected 50–100")
vehicle.command('CHANGE_CHARGE_LIMIT', percent=pct)
print(f"🎚️ {vehicle['display_name']} charge limit set to {pct}%")
return
if args.action == 'amps':
require_yes(args, 'charge amps')
if args.value is None:
raise ValueError("Missing amps value (e.g., charge amps 16)")
amps = int(args.value)
if amps < 1 or amps > 48:
# Conservative guardrail. Many cars support 5-48A depending on setup.
raise ValueError("Invalid amps. Expected 1–48")
vehicle.command('CHARGING_AMPS', charging_amps=amps)
print(f"🔌 {vehicle['display_name']} charging amps set to {amps}A")
return
raise ValueError(f"Unknown action: {args.action}")
def _parse_hhmm(value: str):
"""Parse HH:MM into minutes after midnight."""
if not isinstance(value, str) or not value.strip():
raise ValueError("Missing time. Expected HH:MM (e.g., 23:30)")
s = value.strip()
if ":" not in s:
raise ValueError("Invalid time. Expected HH:MM (e.g., 23:30)")
hh_s, mm_s = s.split(":", 1)
hh = int(hh_s)
mm = int(mm_s)
if hh < 0 or hh > 23 or mm < 0 or mm > 59:
raise ValueError("Invalid time. Expected HH:MM using 24-hour time")
return hh * 60 + mm
def cmd_scheduled_charging(args):
"""Get/set scheduled charging (requires --yes to change)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
# Read-only action can skip waking the car.
allow_wake = True
if args.action == 'status':
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
if args.action == 'status':
data = vehicle.get_vehicle_data()
charge = data.get('charge_state', {})
sched_time = charge.get('scheduled_charging_start_time')
sched_mode = charge.get('scheduled_charging_mode')
sched_pending = charge.get('scheduled_charging_pending')
if args.json:
print(json.dumps({'scheduled_charging_start_time': sched_time,
'scheduled_charging_mode': sched_mode,
'scheduled_charging_pending': sched_pending}, indent=2))
return
hhmm = _fmt_minutes_hhmm(sched_time)
mode = (sched_mode.strip() if isinstance(sched_mode, str) else None)
if not mode and sched_pending is not None:
mode = 'On' if sched_pending else 'Off'
print(f"🚗 {vehicle['display_name']}")
print(f"Scheduled charging: {mode or '(unknown)'}")
if hhmm:
print(f"Start time: {hhmm}")
return
# Mutating actions
require_yes(args, 'scheduled-charging')
if args.action == 'off':
vehicle.command('SCHEDULED_CHARGING', enable=False, time=0)
print(f"⏱️ {vehicle['display_name']} scheduled charging disabled")
return
if args.action == 'set':
minutes = _parse_hhmm(args.time)
vehicle.command('SCHEDULED_CHARGING', enable=True, time=minutes)
print(f"⏱️ {vehicle['display_name']} scheduled charging set to {_fmt_minutes_hhmm(minutes)}")
return
raise ValueError(f"Unknown action: {args.action}")
def _scheduled_departure_status_json(charge: dict) -> dict:
"""Small, privacy-safe scheduled departure status object."""
charge = charge or {}
return {
'scheduled_departure_enabled': charge.get('scheduled_departure_enabled'),
'scheduled_departure_time': charge.get('scheduled_departure_time'),
'scheduled_departure_time_hhmm': _fmt_minutes_hhmm(charge.get('scheduled_departure_time')),
'preconditioning_enabled': charge.get('preconditioning_enabled'),
'off_peak_charging_enabled': charge.get('off_peak_charging_enabled'),
}
def cmd_scheduled_departure(args):
"""Get scheduled departure / off-peak charging / preconditioning status (read-only)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
charge = data.get('charge_state', {})
out = _scheduled_departure_status_json(charge)
if getattr(args, 'json', False):
print(json.dumps(out, indent=2))
return
print(f"🚗 {vehicle['display_name']}")
if out.get('scheduled_departure_enabled') is not None:
print(f"Scheduled departure: {_fmt_bool(out.get('scheduled_departure_enabled'), 'On', 'Off')}")
else:
print("Scheduled departure: (unknown)")
hhmm = out.get('scheduled_departure_time_hhmm')
if hhmm:
print(f"Departure time: {hhmm}")
if out.get('preconditioning_enabled') is not None:
print(f"Preconditioning: {_fmt_bool(out.get('preconditioning_enabled'), 'On', 'Off')}")
if out.get('off_peak_charging_enabled') is not None:
print(f"Off-peak charging: {_fmt_bool(out.get('off_peak_charging_enabled'), 'On', 'Off')}")
def _round_coord(x, digits: int = 2):
"""Round a coordinate for safer display.
digits=2 is roughly ~1km precision (varies with latitude) and is intended
as a non-sensitive default.
We cap digits to a small range to avoid accidentally producing overly
precise coordinates.
"""
try:
d = int(digits)
except Exception:
return None
# 0..6 is still plenty for display; tighter by default.
if d < 0 or d > 6:
return None
try:
return round(float(x), d)
except Exception:
return None
def cmd_location(args):
"""Get vehicle location.
Default output is *approximate* (rounded) to reduce accidental leakage.
Use --yes for precise coordinates.
Use --digits N (0–6) to control rounding precision for approximate output.
"""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
_ensure_online_or_exit(vehicle, allow_wake=not getattr(args, 'no_wake', False))
data = vehicle.get_vehicle_data()
drive = data['drive_state']
lat, lon = drive['latitude'], drive['longitude']
if getattr(args, "yes", False):
print(f"📍 {vehicle['display_name']} Location (precise): {lat}, {lon}")
print(f" https://www.google.com/maps?q={lat},{lon}")
return
digits = getattr(args, 'digits', 2)
lat_r = _round_coord(lat, digits)
lon_r = _round_coord(lon, digits)
if lat_r is None or lon_r is None:
raise ValueError("Invalid or missing location coordinates (try --digits 0..6)")
print(f"📍 {vehicle['display_name']} Location (approx): {lat_r}, {lon_r}")
print(f" https://www.google.com/maps?q={lat_r},{lon_r}")
print(" (Use --yes for precise coordinates)")
def cmd_tires(args):
"""Show tire pressures (TPMS) (read-only)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
# Read-only action can skip waking the car.
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
vs = data.get('vehicle_state', {})
fl = _fmt_tire_pressure(vs.get('tpms_pressure_fl'))
fr = _fmt_tire_pressure(vs.get('tpms_pressure_fr'))
rl = _fmt_tire_pressure(vs.get('tpms_pressure_rl'))
rr = _fmt_tire_pressure(vs.get('tpms_pressure_rr'))
if args.json:
print(json.dumps({
'tpms_pressure_fl': vs.get('tpms_pressure_fl'),
'tpms_pressure_fr': vs.get('tpms_pressure_fr'),
'tpms_pressure_rl': vs.get('tpms_pressure_rl'),
'tpms_pressure_rr': vs.get('tpms_pressure_rr'),
}, indent=2))
return
print(f"🚗 {vehicle['display_name']}")
print("Tire pressures (TPMS):")
print(f" FL: {fl or '(unknown)'}")
print(f" FR: {fr or '(unknown)'}")
print(f" RL: {rl or '(unknown)'}")
print(f" RR: {rr or '(unknown)'}")
def _fmt_open(v):
if v is None:
return None
# Tesla often uses 0/1 ints for open states.
if isinstance(v, bool):
return 'Open' if v else 'Closed'
try:
i = int(v)
return 'Open' if i else 'Closed'
except Exception:
return None
def _openings_one_line(vs: dict) -> str:
"""Return a one-line openings summary from vehicle_state.
Returns None if no openings fields are present.
"""
out = _openings_json(vs)
if out is None:
return None
if out.get("all_closed"):
return "All closed"
if out.get("open"):
# This is a human-facing string; keep it readable.
title = {
"driver_front_door": "Driver front door",
"driver_rear_door": "Driver rear door",
"passenger_front_door": "Passenger front door",
"passenger_rear_door": "Passenger rear door",
"frunk": "Frunk",
"trunk": "Trunk",
"front_driver_window": "Front driver window",
"front_passenger_window": "Front passenger window",
"rear_driver_window": "Rear driver window",
"rear_passenger_window": "Rear passenger window",
}
labels = [title.get(x, x) for x in out["open"]]
return "Open: " + ", ".join(labels)
return None
def _openings_json(vs: dict) -> dict:
"""Sanitized openings JSON from vehicle_state.
Returns None if no openings fields are present.
"""
if not isinstance(vs, dict):
return None
fields = [
("df", "driver_front_door"),
("dr", "driver_rear_door"),
("pf", "passenger_front_door"),
("pr", "passenger_rear_door"),
("ft", "frunk"),
("rt", "trunk"),
("fd_window", "front_driver_window"),
("fp_window", "front_passenger_window"),
("rd_window", "rear_driver_window"),
("rp_window", "rear_passenger_window"),
]
any_known = False
open_items = []
for key, label in fields:
raw = vs.get(key)
if raw is None:
continue
any_known = True
if _fmt_open(raw) == 'Open':
open_items.append(label)
if not any_known:
return None
return {
"open": open_items,
"all_closed": len(open_items) == 0,
}
def cmd_openings(args):
"""Show which doors/trunks/windows are open (read-only)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
vs = data.get('vehicle_state', {})
out = {
'doors': {
'driver_front': _fmt_open(vs.get('df')),
'driver_rear': _fmt_open(vs.get('dr')),
'passenger_front': _fmt_open(vs.get('pf')),
'passenger_rear': _fmt_open(vs.get('pr')),
},
'trunks': {
'frunk': _fmt_open(vs.get('ft')),
'trunk': _fmt_open(vs.get('rt')),
},
'windows': {
'front_driver': _fmt_open(vs.get('fd_window')),
'front_passenger': _fmt_open(vs.get('fp_window')),
'rear_driver': _fmt_open(vs.get('rd_window')),
'rear_passenger': _fmt_open(vs.get('rp_window')),
},
}
# Drop unknown keys for cleaner output.
for k in list(out.keys()):
out[k] = {kk: vv for kk, vv in out[k].items() if vv is not None}
if not out[k]:
del out[k]
if args.json:
print(json.dumps(out, indent=2))
return
print(f"🚗 {vehicle['display_name']}")
if not out:
print("Openings: (unavailable)")
return
def _section(title, d):
if not d:
return
print(f"{title}:")
for kk, vv in d.items():
print(f" - {kk.replace('_',' ')}: {vv}")
_section('Doors', out.get('doors'))
_section('Trunks', out.get('trunks'))
_section('Windows', out.get('windows'))
def cmd_trunk(args):
"""Toggle frunk/trunk (requires --yes)."""
require_yes(args, 'trunk')
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
wake_vehicle(vehicle)
which = 'front' if args.which == 'frunk' else 'rear'
vehicle.command('ACTUATE_TRUNK', which_trunk=which)
label = 'Frunk' if which == 'front' else 'Trunk'
print(f"🧳 {vehicle['display_name']} {label} toggled")
def cmd_windows(args):
"""Windows: status (read-only) or vent/close (requires --yes)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
# Read-only action can skip waking the car.
if args.action == 'status':
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
vs = data.get('vehicle_state', {})
out = {
'front_driver': _fmt_open(vs.get('fd_window')),
'front_passenger': _fmt_open(vs.get('fp_window')),
'rear_driver': _fmt_open(vs.get('rd_window')),
'rear_passenger': _fmt_open(vs.get('rp_window')),
}
# Drop unknowns for cleaner output
out = {k: v for k, v in out.items() if v is not None}
if getattr(args, 'json', False):
print(json.dumps(out, indent=2))
return
print(f"🚗 {vehicle['display_name']}")
if not out:
print("Windows: (unavailable)")
return
print("Windows:")
for k, v in out.items():
print(f" - {k.replace('_',' ')}: {v}")
return
# Mutating actions
require_yes(args, 'windows')
wake_vehicle(vehicle)
# Tesla API requires lat/lon parameters; 0/0 works for this endpoint.
if args.action == 'vent':
vehicle.command('WINDOW_CONTROL', command='vent', lat=0, lon=0)
print(f"🪟 {vehicle['display_name']} windows vented")
return
if args.action == 'close':
vehicle.command('WINDOW_CONTROL', command='close', lat=0, lon=0)
print(f"🪟 {vehicle['display_name']} windows closed")
return
raise ValueError(f"Unknown action: {args.action}")
def _seat_heater_fields(climate_state: dict) -> dict:
"""Extract seat heater levels from climate_state (if present)."""
if not isinstance(climate_state, dict):
return {}
# Common Tesla API fields (may vary by model/firmware).
keys = [
"seat_heater_left", # driver
"seat_heater_right", # passenger
"seat_heater_rear_left",
"seat_heater_rear_center",
"seat_heater_rear_right",
"seat_heater_third_row_left",
"seat_heater_third_row_right",
]
out = {k: climate_state.get(k) for k in keys if k in climate_state}
# Drop unknown/nulls for clean output.
return {k: v for k, v in out.items() if v is not None}
def _seat_heaters_one_line(fields: dict) -> str:
"""Format seat heater levels in a compact one-line form.
Example: "D 3 | P 2 | RL 1".
"""
if not isinstance(fields, dict) or not fields:
return ""
labels = {
"seat_heater_left": "D",
"seat_heater_right": "P",
"seat_heater_rear_left": "RL",
"seat_heater_rear_center": "RC",
"seat_heater_rear_right": "RR",
"seat_heater_third_row_left": "3L",
"seat_heater_third_row_right": "3R",
}
parts = []
for k in [
"seat_heater_left",
"seat_heater_right",
"seat_heater_rear_left",
"seat_heater_rear_center",
"seat_heater_rear_right",
"seat_heater_third_row_left",
"seat_heater_third_row_right",
]:
if k in fields and fields.get(k) is not None:
parts.append(f"{labels.get(k, k)} {fields.get(k)}")
return " | ".join(parts)
_SEAT_NAME_TO_HEATER_ID = {
# Tesla's REMOTE_SEAT_HEATER_REQUEST uses numeric "heater" ids.
# These mappings are the common convention used by community clients.
"driver": 0,
"front-left": 0,
"front_left": 0,
"left": 0,
"passenger": 1,
"front-right": 1,
"front_right": 1,
"right": 1,
"rear-left": 2,
"rear_left": 2,
"rear-center": 3,
"rear_center": 3,
"rear-right": 4,
"rear_right": 4,
"3rd-left": 5,
"3rd_left": 5,
"third-left": 5,
"third_left": 5,
"3rd-right": 6,
"3rd_right": 6,
"third-right": 6,
"third_right": 6,
}
def _parse_seat_heater(seat: str) -> int:
"""Parse a seat name into a Tesla heater id.
Accepts friendly names (driver/passenger/rear-left/etc) or a numeric id.
"""
if seat is None:
raise ValueError("Missing seat. Example: seats set driver 3")
s = str(seat).strip().lower()
if not s:
raise ValueError("Missing seat. Example: seats set driver 3")
if s.isdigit():
hid = int(s)
if hid < 0 or hid > 6:
raise ValueError("Invalid seat heater id. Expected 0–6")
return hid
hid = _SEAT_NAME_TO_HEATER_ID.get(s)
if hid is None:
raise ValueError(
"Unknown seat. Use one of: driver, passenger, rear-left, rear-center, rear-right, 3rd-left, 3rd-right (or 0–6)"
)
return hid
def cmd_seats(args):
"""Seat heaters: status (read-only) or set (requires --yes)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
if args.action == "status":
allow_wake = not getattr(args, "no_wake", False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
data = vehicle.get_vehicle_data()
climate = data.get("climate_state", {})
out = _seat_heater_fields(climate)
if getattr(args, "json", False):
print(json.dumps(out, indent=2))
return
print(f"🚗 {vehicle['display_name']}")
if not out:
print("Seat heaters: (unavailable)")
return
print("Seat heaters (0=off .. 3=high):")
for k, v in out.items():
label = k.replace("seat_heater_", "").replace("_", " ")
print(f" - {label}: {v}")
return
if args.action == "set":
require_yes(args, "seats set")
wake_vehicle(vehicle)
heater = _parse_seat_heater(getattr(args, "seat", None))
if getattr(args, "level", None) is None:
raise ValueError("Missing level. Expected 0–3")
level = int(getattr(args, "level"))
if level < 0 or level > 3:
raise ValueError("Invalid level. Expected 0–3")
vehicle.command("REMOTE_SEAT_HEATER_REQUEST", heater=heater, level=level)
print(f"🔥 {vehicle['display_name']} seat heater {heater} set to {level}")
return
raise ValueError(f"Unknown action: {args.action}")
def cmd_sentry(args):
"""Get/set Sentry Mode (on/off requires --yes)."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
# Read-only action can skip waking the car.
allow_wake = True
if args.action == 'status':
allow_wake = not getattr(args, 'no_wake', False)
_ensure_online_or_exit(vehicle, allow_wake=allow_wake)
if args.action == 'status':
data = vehicle.get_vehicle_data()
sentry = data.get('vehicle_state', {}).get('sentry_mode')
if args.json:
print(json.dumps({'sentry_mode': sentry}, indent=2))
return
if sentry is None:
print(f"🚗 {vehicle['display_name']}\nSentry: (unknown)")
else:
print(f"🚗 {vehicle['display_name']}\nSentry: {_fmt_bool(sentry, 'On', 'Off')}")
return
# Mutating actions
require_yes(args, 'sentry')
wake_vehicle(vehicle)
if args.action == 'on':
vehicle.command('SET_SENTRY_MODE', on=True)
print(f"🛡️ {vehicle['display_name']} Sentry turned on")
return
if args.action == 'off':
vehicle.command('SET_SENTRY_MODE', on=False)
print(f"🛡️ {vehicle['display_name']} Sentry turned off")
return
raise ValueError(f"Unknown action: {args.action}")
def cmd_honk(args):
"""Honk the horn."""
require_yes(args, 'honk')
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
wake_vehicle(vehicle)
vehicle.command('HONK_HORN')
print(f"📢 {vehicle['display_name']} honked!")
def require_yes(args, action: str):
if not getattr(args, "yes", False):
print(f"❌ Refusing to run '{action}' without --yes (safety gate)", file=sys.stderr)
sys.exit(2)
def cmd_flash(args):
"""Flash the lights."""
require_yes(args, 'flash')
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
wake_vehicle(vehicle)
vehicle.command('FLASH_LIGHTS')
print(f"💡 {vehicle['display_name']} flashed lights!")
def _charge_port_status_json(vehicle, data: dict) -> dict:
"""Small, privacy-safe charge port status object."""
charge = (data or {}).get('charge_state', {})
return {
'display_name': (vehicle or {}).get('display_name'),
'state': (vehicle or {}).get('state'),
'charge_port_door_open': charge.get('charge_port_door_open'),
'charge_port_latch': charge.get('charge_port_latch'),
'conn_charge_cable': charge.get('conn_charge_cable'),
'charging_state': charge.get('charging_state'),
}
def cmd_charge_port(args):
"""Charge port operations.
- status: read-only (supports --no-wake)
- open/close: requires --yes
"""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
if args.action == 'status':
_ensure_online_or_exit(vehicle, allow_wake=not getattr(args, 'no_wake', False))
data = vehicle.get_vehicle_data()
out = _charge_port_status_json(vehicle, data)
if getattr(args, 'json', False):
# Keep this privacy-safe + stable.
print(json.dumps(out, indent=2))
return
print(f"🔌 {vehicle.get('display_name')}")
print(f" State: {vehicle.get('state')}")
if out.get('charge_port_door_open') is not None:
print(f" Port door open: {_fmt_bool(out.get('charge_port_door_open'))}")
if out.get('charge_port_latch') is not None:
print(f" Port latch: {out.get('charge_port_latch')}")
if out.get('conn_charge_cable') is not None:
print(f" Cable: {out.get('conn_charge_cable')}")
if out.get('charging_state') is not None:
print(f" Charging: {out.get('charging_state')}")
return
# open/close
require_yes(args, 'charge-port')
wake_vehicle(vehicle)
if args.action == 'open':
vehicle.command('CHARGE_PORT_DOOR_OPEN')
print(f"🔌 {vehicle['display_name']} charge port opened")
elif args.action == 'close':
vehicle.command('CHARGE_PORT_DOOR_CLOSE')
print(f"🔌 {vehicle['display_name']} charge port closed")
else:
raise ValueError(f"Unknown action: {args.action}")
def cmd_wake(args):
"""Wake up the vehicle."""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
print(f"⏳ Waking {vehicle['display_name']}...")
vehicle.sync_wake_up()
print(f"✅ {vehicle['display_name']} is awake")
def cmd_summary(args):
"""One-line status summary.
- default (human): prints a single line for chat
- with --json: prints a sanitized JSON object (privacy-safe)
- with --json --raw-json: prints raw vehicle_data (may include location)
"""
tesla = get_tesla(require_email(args))
vehicle = get_vehicle(tesla, args.car)
_ensure_online_or_exit(vehicle, allow_wake=not getattr(args, 'no_wake', False))
data = vehicle.get_vehicle_data()
if getattr(args, 'json', False):
if getattr(args, 'raw_json', False):
print(json.dumps(data, indent=2))
else:
print(json.dumps(_summary_json(vehicle, data), indent=2))
return
print(_short_status(vehicle, data))
# ----------------------------
# Mileage tracking (SQLite)
# ----------------------------
MILEAGE_DIR = Path.home() / ".my_tesla"
MILEAGE_DB_DEFAULT = MILEAGE_DIR / "mileage.sqlite"
def resolve_mileage_db_path(args=None) -> Path:
"""Resolve mileage DB path.
Priority:
1) --db (for mileage commands)
2) MY_TESLA_MILEAGE_DB env var
3) ~/.my_tesla/mileage.sqlite
"""
if args is not None and getattr(args, "db", None):
return Path(getattr(args, "db")).expanduser()
env = os.environ.get("MY_TESLA_MILEAGE_DB")
if env and env.strip():
return Path(env.strip()).expanduser()
return MILEAGE_DB_DEFAULT
def resolve_since_ts(*, since_ts: int | None = None, since_days: float | None = None) -> int | None:
"""Resolve a cutoff timestamp (UTC epoch seconds) for mileage export.
- since_ts wins if provided.
- since_days is interpreted as "now - N days".
Returns None when no cutoff is requested.
"""
if since_ts is not None:
try:
return int(since_ts)
except Exception:
raise ValueError("--since-ts must be an integer epoch timestamp (seconds)")
if since_days is None:
return None
try:
days = float(since_days)
except Exception:
raise ValueError("--since-days must be a number (e.g., 7 or 0.5)")
if days < 0:
raise ValueError("--since-days must be >= 0")
return int(time.time() - days * 86400)
def mileage_fetch_points(conn, *, since_ts: int | None = None):
"""Fetch mileage points ordered by timestamp asc, optionally filtered."""
if since_ts is None:
cur = conn.execute(
"SELECT ts_utc, vehicle_id, vehicle_name, odometer_mi, state, source, note FROM mileage_points ORDER BY ts_utc ASC"
)
return cur.fetchall()
cur = conn.execute(
"SELECT ts_utc, vehicle_id, vehicle_name, odometer_mi, state, source, note FROM mileage_points WHERE ts_utc >= ? ORDER BY ts_utc ASC",
(int(since_ts),),
)
return cur.fetchall()
def _db_connect(path: Path):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(path))
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA foreign_keys=ON;")
return conn
def mileage_init_db(path: Path):
conn = _db_connect(path)
try:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS mileage_points (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts_utc INTEGER NOT NULL,
vehicle_id TEXT,
vehicle_name TEXT,
odometer_mi REAL,
state TEXT,
source TEXT,
note TEXT
);
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_mileage_points_vehicle_ts ON mileage_points(vehicle_id, ts_utc);"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_mileage_points_ts ON mileage_points(ts_utc);"
)
conn.commit()
finally:
conn.close()
def _vehicle_identity(vehicle) -> tuple[str, str]:
vid = vehicle.get("id_s") or str(vehicle.get("vehicle_id") or "")
name = vehicle.get("display_name") or "Vehicle"
return (vid, name)
def mileage_last_success_ts(conn, vehicle_id: str) -> int | None:
cur = conn.execute(
"SELECT ts_utc FROM mileage_points WHERE vehicle_id=? AND odometer_mi IS NOT NULL ORDER BY ts_utc DESC LIMIT 1",
(vehicle_id,),
)
row = cur.fetchone()
return int(row[0]) if row else None
def mileage_insert_point(
conn,
*,
ts_utc: int,
vehicle_id: str,
vehicle_name: str,
odometer_mi: float | None,
state: str | None,
source: str,
note: str | None = None,
):
conn.execute(
"""
INSERT INTO mileage_points(ts_utc, vehicle_id, vehicle_name, odometer_mi, state, source, note)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(ts_utc, vehicle_id, vehicle_name, odometer_mi, state, source, note),
)
def _fmt_dt(ts_utc: int) -> str:
dt = datetime.fromtimestamp(ts_utc, tz=timezone.utc)
return dt.strftime("%Y-%m-%d %H:%M UTC")
def cmd_mileage(args):
"""Mileage tracking commands."""
db_path = resolve_mileage_db_path(args)
if args.action == "init":
mileage_init_db(db_path)
print(f"✅ Mileage DB ready: {db_path}")
return
# record/status/export require DB
mileage_init_db(db_path)
if args.action == "record":
tesla = get_tesla(require_email(args))
vehicles = tesla.vehicle_list()
if not vehicles:
print("❌ No vehicles found on this account", file=sys.stderr)
sys.exit(1)
ts = int(time.time())
allow_any_wake = not getattr(args, "no_wake", False)
wake_after_hours = float(getattr(args, "auto_wake_after_hours", 24.0) or 24.0)
wake_after_sec = int(wake_after_hours * 3600)
results = []
conn = _db_connect(db_path)
try:
for v in vehicles:
vid, name = _vehicle_identity(v)
state = v.get("state")
# Determine if we should allow waking based on last successful capture.
last_ts = mileage_last_success_ts(conn, vid) if vid else None
too_old = last_ts is None or (ts - last_ts) >= wake_after_sec
allow_wake = allow_any_wake and bool(too_old)
# Try to avoid waking unless threshold exceeded.
if not wake_vehicle(v, allow_wake=allow_wake):
mileage_insert_point(
conn,
ts_utc=ts,
vehicle_id=vid,
vehicle_name=name,
odometer_mi=None,
state=state,
source="skip",
note=(
f"skipped (state={state}); no_wake={getattr(args,'no_wake',False)}; "
f"last_ok={_fmt_dt(last_ts) if last_ts else 'never'}"
),
)
results.append({
"vehicle": name,
"vehicle_id": vid,
"state": state,
"recorded": False,
"reason": "asleep_or_offline",
"last_success_ts_utc": last_ts,
"woke": False,
})
continue
# Online now.
data = v.get_vehicle_data()
vs = data.get("vehicle_state", {})
odo = vs.get("odometer")
mileage_insert_point(
conn,
ts_utc=ts,
vehicle_id=vid,
vehicle_name=name,
odometer_mi=float(odo) if odo is not None else None,
state=v.get("state"),
source="vehicle_data",
note=None,
)
results.append({
"vehicle": name,
"vehicle_id": vid,
"state": v.get("state"),
"recorded": odo is not None,
"odometer_mi": float(odo) if odo is not None else None,
"woke": bool(allow_wake and state != 'online'),
})
conn.commit()
finally:
conn.close()
if args.json:
print(json.dumps({"db": str(db_path), "ts_utc": ts, "results": results}, indent=2))
return
ok = sum(1 for r in results if r.get("recorded"))
print(f"✅ Recorded mileage: {ok}/{len(results)} vehicles")
for r in results:
if r.get("recorded"):
print(f"- {r['vehicle']}: {r.get('odometer_mi'):.1f} mi")
else:
print(f"- {r['vehicle']}: skipped ({r.get('reason')})")
return
if args.action == "status":
conn = _db_connect(db_path)
try:
cur = conn.execute(
"""
SELECT vehicle_name, vehicle_id, ts_utc, odometer_mi
FROM mileage_points
WHERE odometer_mi IS NOT NULL
ORDER BY ts_utc DESC
"""
)
rows = cur.fetchall()
finally:
conn.close()
# latest per vehicle
latest = {}
for name, vid, ts, odo in rows:
if vid not in latest:
latest[vid] = {"vehicle": name, "vehicle_id": vid, "ts_utc": int(ts), "odometer_mi": float(odo)}
out = {"db": str(db_path), "vehicles": list(latest.values())}
if args.json:
print(json.dumps(out, indent=2))
return
print(f"Mileage DB: {db_path}")
if not latest:
print("No mileage points recorded yet. Run: " + _invocation("mileage record"))
return
for v in out["vehicles"]:
print(f"- {v['vehicle']}: {v['odometer_mi']:.1f} mi (last: {_fmt_dt(v['ts_utc'])})")
return
if args.action == "export":
fmt = getattr(args, "format", "csv")
since_ts = resolve_since_ts(
since_ts=getattr(args, "since_ts", None),
since_days=getattr(args, "since_days", None),
)
conn = _db_connect(db_path)
try:
rows = mileage_fetch_points(conn, since_ts=since_ts)
finally:
conn.close()
if fmt == "json":
items = [
{
"ts_utc": int(ts),
"vehicle_id": vid,
"vehicle": name,
"odometer_mi": odo,
"state": state,
"source": source,
"note": note,
}
for (ts, vid, name, odo, state, source, note) in rows
]
print(json.dumps({"db": str(db_path), "since_ts_utc": since_ts, "items": items}, indent=2))
return
# csv
import csv
w = csv.writer(sys.stdout)
w.writerow(["ts_utc", "vehicle_id", "vehicle", "odometer_mi", "state", "source", "note"])
for (ts, vid, name, odo, state, source, note) in rows:
w.writerow([int(ts), vid, name, odo, state, source, note or ""])
return
raise ValueError(f"Unknown mileage action: {args.action}")
def cmd_version(args):
"""Print the installed skill version."""
# Keep output simple for scripts.
print(read_skill_version())
def cmd_default_car(args):
"""Set or show the default car used when --car is not provided."""
if not args.name:
name = resolve_default_car_name()
if name:
print(f"Default car: {name}")
else:
print("Default car: (none)")
return
defaults = load_defaults()
defaults["default_car"] = args.name
save_defaults(defaults)
print(f"✅ Default car set to: {args.name}")
print(f"Saved to: {DEFAULTS_FILE}")
def main():
parser = argparse.ArgumentParser(description="Tesla vehicle control")
parser.add_argument("--email", "-e", help="Tesla account email")
parser.add_argument("--car", "-c", help="Vehicle name (default: first vehicle)")
parser.add_argument("--json", "-j", action="store_true", help="Output JSON")
parser.add_argument(
"--raw-json",
action="store_true",
help=(
"When used with --json on supported commands, output raw vehicle_data (may include location). "
"Default JSON output is sanitized/summary for safety."
),
)
parser.add_argument(
"--yes",
action="store_true",
help=(
"Safety confirmation for sensitive/disruptive actions "
"(unlock/charge start|stop|limit|amps/trunk/windows/seats set/honk/flash/charge-port open|close/"
"scheduled-charging set|off/sentry on|off/location precise)"
),
)
parser.add_argument(
"--debug",
action="store_true",
help="Print a full Python traceback on errors (also enabled by MY_TESLA_DEBUG=1)",
)
parser.add_argument(
"--version",
action="store_true",
help="Print skill version and exit",
)
subparsers = parser.add_subparsers(dest="command")
# Auth
subparsers.add_parser("auth", help="Authenticate with Tesla")
# List
subparsers.add_parser("list", help="List all vehicles")
# Version
subparsers.add_parser("version", help="Print skill version")
# Status
status_parser = subparsers.add_parser("status", help="Get vehicle status")
status_parser.add_argument("--summary", action="store_true", help="Also print a one-line summary")
status_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Summary (alias)
summary_parser = subparsers.add_parser("summary", help="One-line status summary")
summary_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Report (one-screen)
report_parser = subparsers.add_parser("report", help="One-screen status report")
report_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Default car
default_parser = subparsers.add_parser("default-car", help="Set/show default vehicle name")
default_parser.add_argument("name", nargs="?", help="Vehicle display name to set as default")
# Mileage tracking (odometer)
mileage_parser = subparsers.add_parser("mileage", help="Record odometer mileage to a local SQLite DB")
mileage_parser.add_argument("action", choices=["init", "record", "status", "export"], help="init|record|status|export")
mileage_parser.add_argument("--json", action="store_true", help="Output JSON (record/status only)")
mileage_parser.add_argument("--db", help="Path to the SQLite DB (default: ~/.my_tesla/mileage.sqlite)")
mileage_parser.add_argument("--no-wake", action="store_true", help="Do not wake sleeping cars")
mileage_parser.add_argument(
"--auto-wake-after-hours",
type=float,
default=24.0,
help="If a car hasn't recorded mileage in this many hours, allow waking it (default: 24)",
)
mileage_parser.add_argument("--format", choices=["csv", "json"], default="csv", help="For export: csv|json")
mileage_parser.add_argument(
"--since-ts",
type=int,
help="(export only) Only include points with ts_utc >= this epoch timestamp (seconds)",
)
mileage_parser.add_argument(
"--since-days",
type=float,
help="(export only) Only include points from the last N days (e.g., 7 or 0.5)",
)
# Lock/unlock
subparsers.add_parser("lock", help="Lock the vehicle")
subparsers.add_parser("unlock", help="Unlock the vehicle")
# Climate
climate_parser = subparsers.add_parser("climate", help="Climate control")
climate_parser.add_argument("action", choices=["status", "on", "off", "temp", "defrost"])
climate_parser.add_argument(
"value",
nargs="?",
help="For 'temp': temperature value. For 'defrost': on|off.",
)
climate_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
temp_units = climate_parser.add_mutually_exclusive_group()
temp_units.add_argument("--fahrenheit", "-f", action="store_true", help="Temperature value is in °F (default)")
temp_units.add_argument("--celsius", action="store_true", help="Temperature value is in °C")
# Charge
charge_parser = subparsers.add_parser("charge", help="Charging control")
charge_parser.add_argument("action", choices=["status", "start", "stop", "limit", "amps"])
charge_parser.add_argument(
"value",
nargs="?",
help="For 'limit': percent (e.g., 80). For 'amps': amps (e.g., 16).",
)
charge_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
# Scheduled charging
sched_parser = subparsers.add_parser("scheduled-charging", help="Get/set scheduled charging (set/off requires --yes)")
sched_parser.add_argument("action", choices=["status", "set", "off"], help="status|set|off")
sched_parser.add_argument("time", nargs="?", help="Start time for 'set' as HH:MM (24-hour)")
sched_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
# Scheduled departure (read-only)
dep_parser = subparsers.add_parser(
"scheduled-departure",
help="Scheduled departure / preconditioning / off-peak charging status (read-only)",
)
dep_parser.add_argument("action", choices=["status"], help="status")
dep_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Location
location_parser = subparsers.add_parser(
"location",
help="Get vehicle location (approx by default; use --yes for precise)",
)
location_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
location_parser.add_argument(
"--digits",
type=int,
default=2,
help="(approx output) Rounding precision for latitude/longitude (0–6). Default: 2",
)
# Tire pressures (TPMS)
tires_parser = subparsers.add_parser("tires", help="Show tire pressures (TPMS)")
tires_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Openings (doors/trunks/windows)
openings_parser = subparsers.add_parser("openings", help="Show which doors/trunks/windows are open")
openings_parser.add_argument("--no-wake", action="store_true", help="Do not wake the car (fails if asleep)")
# Trunk / frunk
trunk_parser = subparsers.add_parser("trunk", help="Toggle trunk/frunk (requires --yes)")
trunk_parser.add_argument("which", choices=["trunk", "frunk"], help="Which to actuate")
# Windows
windows_parser = subparsers.add_parser("windows", help="Windows status (read-only) or vent/close (requires --yes)")
windows_parser.add_argument("action", choices=["status", "vent", "close"], help="status|vent|close")
windows_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
windows_parser.add_argument("--json", action="store_true", help="(status only) Output JSON")
# Seat heaters
seats_parser = subparsers.add_parser("seats", help="Seat heater status (read-only) or set level (requires --yes)")
seats_parser.add_argument("action", choices=["status", "set"], help="status|set")
seats_parser.add_argument("seat", nargs="?", help="For 'set': driver|passenger|rear-left|rear-center|rear-right|3rd-left|3rd-right (or 0–6)")
seats_parser.add_argument("level", nargs="?", help="For 'set': 0–3 (0=off)")
seats_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
seats_parser.add_argument("--json", action="store_true", help="(status only) Output JSON")
# Sentry
sentry_parser = subparsers.add_parser("sentry", help="Get/set Sentry Mode (on/off requires --yes)")
sentry_parser.add_argument("action", choices=["status", "on", "off"], help="status|on|off")
sentry_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
# Honk/flash
subparsers.add_parser("honk", help="Honk the horn")
subparsers.add_parser("flash", help="Flash the lights")
# Charge port
charge_port_parser = subparsers.add_parser(
"charge-port",
help="Charge port status (read-only) or open/close (requires --yes)",
)
charge_port_parser.add_argument("action", choices=["status", "open", "close"], help="status|open|close")
charge_port_parser.add_argument("--no-wake", action="store_true", help="(status only) Do not wake the car")
# Wake
subparsers.add_parser("wake", help="Wake up the vehicle")
args = parser.parse_args()
if getattr(args, "version", False):
cmd_version(args)
return
if not getattr(args, "command", None):
parser.print_help(sys.stderr)
sys.exit(2)
commands = {
"auth": cmd_auth,
"list": cmd_list,
"version": cmd_version,
"status": cmd_status,
"summary": cmd_summary,
"report": cmd_report,
"lock": cmd_lock,
"unlock": cmd_unlock,
"climate": cmd_climate,
"charge": cmd_charge,
"scheduled-charging": cmd_scheduled_charging,
"scheduled-departure": cmd_scheduled_departure,
"location": cmd_location,
"tires": cmd_tires,
"openings": cmd_openings,
"trunk": cmd_trunk,
"windows": cmd_windows,
"seats": cmd_seats,
"sentry": cmd_sentry,
"honk": cmd_honk,
"flash": cmd_flash,
"charge-port": cmd_charge_port,
"wake": cmd_wake,
"default-car": cmd_default_car,
"mileage": cmd_mileage,
}
try:
commands[args.command](args)
except KeyboardInterrupt:
print("\n⛔ Interrupted", file=sys.stderr)
sys.exit(130)
except Exception as e:
debug = bool(getattr(args, "debug", False)) or os.environ.get("MY_TESLA_DEBUG") == "1"
if debug:
# Print both a friendly line and a full traceback.
print(f"❌ Error: {e}", file=sys.stderr)
traceback.print_exc()
else:
print(f"❌ Error: {e}", file=sys.stderr)
print(" Tip: re-run with --debug for a full traceback", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
Reference in New Issue Copy Permalink