mirror of
https://github.com/RichieCahill/dotfiles.git
synced 2026-04-17 04:58:19 -04:00
updated _format_location to use van van_last_known_longitude and van_last_known_latitude
This commit is contained in:
@@ -11,8 +11,8 @@ if TYPE_CHECKING:
|
|||||||
from python.signal_bot.signal_client import SignalClient
|
from python.signal_bot.signal_client import SignalClient
|
||||||
|
|
||||||
|
|
||||||
def _get_location_payload(ha_url: str, ha_token: str, entity_id: str) -> dict[str, Any]:
|
def _get_entity_state(ha_url: str, ha_token: str, entity_id: str) -> dict[str, Any]:
|
||||||
"""Fetch location entity state from Home Assistant."""
|
"""Fetch an entity's state from Home Assistant."""
|
||||||
response = requests.get(
|
response = requests.get(
|
||||||
f"{ha_url}/api/states/{entity_id}",
|
f"{ha_url}/api/states/{entity_id}",
|
||||||
headers={"Authorization": f"Bearer {ha_token}"},
|
headers={"Authorization": f"Bearer {ha_token}"},
|
||||||
@@ -22,35 +22,12 @@ def _get_location_payload(ha_url: str, ha_token: str, entity_id: str) -> dict[st
|
|||||||
return response.json()
|
return response.json()
|
||||||
|
|
||||||
|
|
||||||
def _format_location(payload: dict[str, Any]) -> str:
|
def _format_location(latitude: str, longitude: str) -> str:
|
||||||
"""Render a friendly location response."""
|
"""Render a friendly location response."""
|
||||||
attributes = payload.get("attributes", {})
|
return (
|
||||||
latitude = attributes.get("latitude")
|
f"Van location: {latitude}, {longitude}\n"
|
||||||
longitude = attributes.get("longitude")
|
f"https://maps.google.com/?q={latitude},{longitude}"
|
||||||
|
)
|
||||||
if latitude is None or longitude is None:
|
|
||||||
state = payload.get("state", "unknown")
|
|
||||||
if "," not in state:
|
|
||||||
return "Van location is unavailable in Home Assistant right now."
|
|
||||||
latitude_text, longitude_text = [part.strip() for part in state.split(",", maxsplit=1)]
|
|
||||||
else:
|
|
||||||
latitude_text = str(latitude)
|
|
||||||
longitude_text = str(longitude)
|
|
||||||
|
|
||||||
lines = [
|
|
||||||
f"Van location: {latitude_text}, {longitude_text}",
|
|
||||||
f"https://maps.google.com/?q={latitude_text},{longitude_text}",
|
|
||||||
]
|
|
||||||
|
|
||||||
speed = attributes.get("speed")
|
|
||||||
if speed not in (None, "", "unknown", "unavailable"):
|
|
||||||
lines.append(f"Speed: {speed}")
|
|
||||||
|
|
||||||
last_updated = attributes.get("last_updated")
|
|
||||||
if last_updated:
|
|
||||||
lines.append(f"Updated: {last_updated}")
|
|
||||||
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
|
|
||||||
def handle_location_request(
|
def handle_location_request(
|
||||||
@@ -58,7 +35,6 @@ def handle_location_request(
|
|||||||
signal: SignalClient,
|
signal: SignalClient,
|
||||||
ha_url: str | None,
|
ha_url: str | None,
|
||||||
ha_token: str | None,
|
ha_token: str | None,
|
||||||
ha_location_entity: str,
|
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Reply with van location from Home Assistant."""
|
"""Reply with van location from Home Assistant."""
|
||||||
if ha_url is None or ha_token is None:
|
if ha_url is None or ha_token is None:
|
||||||
@@ -66,9 +42,17 @@ def handle_location_request(
|
|||||||
return
|
return
|
||||||
|
|
||||||
try:
|
try:
|
||||||
payload = _get_location_payload(ha_url, ha_token, ha_location_entity)
|
lat_payload = _get_entity_state(ha_url, ha_token, "sensor.van_last_known_latitude")
|
||||||
|
lon_payload = _get_entity_state(ha_url, ha_token, "sensor.van_last_known_longitude")
|
||||||
except requests.RequestException:
|
except requests.RequestException:
|
||||||
signal.reply(message, "Couldn't fetch van location from Home Assistant right now.")
|
signal.reply(message, "Couldn't fetch van location from Home Assistant right now.")
|
||||||
return
|
return
|
||||||
|
|
||||||
signal.reply(message, _format_location(payload))
|
latitude = lat_payload.get("state", "")
|
||||||
|
longitude = lon_payload.get("state", "")
|
||||||
|
|
||||||
|
if not latitude or not longitude or latitude == "unavailable" or longitude == "unavailable":
|
||||||
|
signal.reply(message, "Van location is unavailable in Home Assistant right now.")
|
||||||
|
return
|
||||||
|
|
||||||
|
signal.reply(message, _format_location(latitude, longitude))
|
||||||
|
|||||||
@@ -87,9 +87,7 @@ class Bot:
|
|||||||
|
|
||||||
def _location(self, message: SignalMessage, _cmd: str) -> None:
|
def _location(self, message: SignalMessage, _cmd: str) -> None:
|
||||||
"""Reply with current van location."""
|
"""Reply with current van location."""
|
||||||
handle_location_request(
|
handle_location_request(message, self.signal, self.config.ha_url, self.config.ha_token)
|
||||||
message, self.signal, self.config.ha_url, self.config.ha_token, self.config.ha_location_entity
|
|
||||||
)
|
|
||||||
|
|
||||||
# -- dispatch -------------------------------------------------------------
|
# -- dispatch -------------------------------------------------------------
|
||||||
|
|
||||||
@@ -210,7 +208,6 @@ def main(
|
|||||||
inventory_api_url=inventory_api_url,
|
inventory_api_url=inventory_api_url,
|
||||||
ha_url=getenv("HA_URL"),
|
ha_url=getenv("HA_URL"),
|
||||||
ha_token=getenv("HA_TOKEN"),
|
ha_token=getenv("HA_TOKEN"),
|
||||||
ha_location_entity=getenv("HA_LOCATION_ENTITY", "sensor.gps_location"),
|
|
||||||
engine=engine,
|
engine=engine,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -90,7 +90,6 @@ class BotConfig(BaseModel):
|
|||||||
inventory_api_url: str
|
inventory_api_url: str
|
||||||
ha_url: str | None = None
|
ha_url: str | None = None
|
||||||
ha_token: str | None = None
|
ha_token: str | None = None
|
||||||
ha_location_entity: str = "sensor.gps_location"
|
|
||||||
engine: Engine
|
engine: Engine
|
||||||
reconnect_delay: int = 5
|
reconnect_delay: int = 5
|
||||||
max_reconnect_delay: int = 300
|
max_reconnect_delay: int = 300
|
||||||
|
|||||||
Reference in New Issue
Block a user