Compare commits

..

1 Commits

Author SHA1 Message Date
9b651f933e Shipping v0.5.0 2020-05-23 15:07:27 +02:00
14 changed files with 294 additions and 1334 deletions

View File

@@ -1,5 +1,5 @@
# GeoRide Home assistant # Georide Home assistant
![Logo GeoRide](https://brands.home-assistant.io/georide/logo.png) ![Logo Georide](/georide-logo.png)
⚠️ This is not an official implementation ⚠️ This is not an official implementation
[![hacs_badge](https://img.shields.io/badge/HACS-Default-orange.svg?style=for-the-badge)](https://github.com/custom-components/hacs) [![hacs_badge](https://img.shields.io/badge/HACS-Default-orange.svg?style=for-the-badge)](https://github.com/custom-components/hacs)
@@ -10,86 +10,6 @@ Official GeoRide website: https://georide.fr/
## Description ## Description
This component add some sensor for GeoRide Tracker This component add some sensor for GeoRide Tracker
### What's entity is available :
Get GeoRide position
Get GeoRide lock status
Change GeoRide lock status
Get stollen status
Get crashed status
Get is owner status
Get subsription status
Get odomoter to km an m (2 entities)
Internal battery (of georide 3) (not work on GR1)
External battery (of the bike) (not work on GR1)
Fixtime (last registered positition of the georide)
### What's events are available:
you can filter by data.device_id == XX (XX is your tracker id)
you can display your tracker name by by data.device_name
event;
georide_position_event
georide_lock_event
georide_device_event
georide_alarm_event
you can filter with data.type == 'alarm_vibration' to filter by vibration
here is the alarm type available: (listen the georide_alarm_event)
alarm_vibration
alarm_exitZone
alarm_crash
alarm_crashParking
alarm_deviceOffline
alarm_deviceOnline
alarm_powerCut
alarm_powerUncut
alarm_batteryWarning
alarm_temperatureWarning
alarm_magnetOn
alarm_magnetOff
alarm_sonorAlarmOn
## Question:
### How to have the odometer in Km ? (Deprecated, now you have an entity - thx @Inervo)
Simply add a sensor like this in configuration.yaml
(Replace XX by your tracker id)
```yaml
sensor:
- platform: template # Conversion georide de m en km
sensors:
odometer_XX_km:
friendly_name: "Odometter - Km"
value_template: "{{ states.sensor.odometer_XX.state | multiply(0.001) | round(3, 'flour') }}"
unit_of_measurement: 'Km'
```
### How to use the event:
Simply made a automation like this:
```yaml
alias: '[TEST] Send notification'
description: ''
trigger:
- platform: event
event_type: georide_lock_event
condition: []
action:
- service: notify.mobile_app_oneplus_a3003
data:
message: 'The device {{ data.device_name }} have recieved a lock event'
mode: single
```
## Options ## Options

View File

@@ -1,9 +1,7 @@
""" georide custom conpennt """ """ georide custom conpennt """
from collections import defaultdict from collections import defaultdict
import asyncio
import logging import logging
from typing import Any, Mapping
from datetime import timedelta from datetime import timedelta
import math import math
import time import time
@@ -13,10 +11,11 @@ import voluptuous as vol
import jwt import jwt
from aiohttp.web import json_response from aiohttp.web import json_response
from georideapilib.objects import GeoRideAccount from georideapilib.objects import GeorideAccount
import georideapilib.api as GeoRideApi import georideapilib.api as GeorideApi
from georideapilib.socket import GeorideSocket
from georideapilib.socket import GeoRideSocket
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.const import CONF_WEBHOOK_ID from homeassistant.const import CONF_WEBHOOK_ID
@@ -25,14 +24,7 @@ import homeassistant.helpers.config_validation as cv
import homeassistant.helpers.event as ha_event import homeassistant.helpers.event as ha_event
from homeassistant.setup import async_when_setup from homeassistant.setup import async_when_setup
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .device import Device, DeviceBeacon
from .const import ( from .const import (
CONF_EMAIL, CONF_EMAIL,
CONF_PASSWORD, CONF_PASSWORD,
@@ -40,12 +32,10 @@ from .const import (
TRACKER_ID, TRACKER_ID,
TOKEN_SAFE_DAY, TOKEN_SAFE_DAY,
MIN_UNTIL_REFRESH, MIN_UNTIL_REFRESH,
DOMAIN, DOMAIN
SIREN_ACTIVATION_DELAY
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -61,11 +51,11 @@ CONFIG_SCHEMA = vol.Schema(
async def async_setup(hass, config): async def async_setup(hass, config):
"""Setup GeoRide component.""" """Setup Georide component."""
hass.data[DOMAIN] = {"config": config[DOMAIN], "devices": {}, "unsub": None} hass.data[DOMAIN] = {"config": config[DOMAIN], "devices": {}, "unsub": None}
hass.async_create_task( hass.async_create_task(
hass.config_entries.flow.async_init( hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={ context={
"source": config_entries.SOURCE_IMPORT "source": config_entries.SOURCE_IMPORT
}, },
@@ -78,25 +68,25 @@ async def async_setup(hass, config):
async def async_setup_entry(hass, entry): async def async_setup_entry(hass, entry):
"""Set up GeoRide entry.""" """Set up Georide entry."""
config = hass.data[DOMAIN]["config"] config = hass.data[DOMAIN]["config"]
email = config.get(CONF_EMAIL) or entry.data[CONF_EMAIL] email = config.get(CONF_EMAIL) or entry.data[CONF_EMAIL]
password = config.get(CONF_PASSWORD) or entry.data[CONF_PASSWORD] password = config.get(CONF_PASSWORD) or entry.data[CONF_PASSWORD]
token = config.get(CONF_TOKEN) or entry.data[CONF_TOKEN] token = config.get(CONF_TOKEN) or entry.data[CONF_TOKEN]
context = GeoRideContext( context = GeorideContext(
hass, hass,
email, email,
password, password,
token token
) )
_LOGGER.info("Context-setup and start the thread") _LOGGER.info("Context-setup and start the thread")
_LOGGER.info("Thread started") _LOGGER.info("Thread started")
hass.data[DOMAIN]["context"] = context hass.data[DOMAIN]["context"] = context
# We add trackers to the context # We add trackers to the context
await context.init_context(hass) context.refresh_trackers()
hass.async_create_task( hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "device_tracker")) hass.config_entries.async_forward_entry_setup(entry, "device_tracker"))
@@ -104,42 +94,47 @@ async def async_setup_entry(hass, entry):
hass.config_entries.async_forward_entry_setup(entry, "switch")) hass.config_entries.async_forward_entry_setup(entry, "switch"))
hass.async_create_task( hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "sensor")) hass.config_entries.async_forward_entry_setup(entry, "sensor"))
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "binary_sensor"))
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "siren"))
return True return True
async def async_unload_entry(hass, entry): async def async_unload_entry(hass, entry):
"""Unload an GeoRide config entry.""" """Unload an Georide config entry."""
await hass.config_entries.async_forward_entry_unload(entry, "device_tracker") await hass.config_entries.async_forward_entry_unload(entry, "device_tracker")
await hass.config_entries.async_forward_entry_unload(entry, "switch") await hass.config_entries.async_forward_entry_unload(entry, "switch")
await hass.config_entries.async_forward_entry_unload(entry, "sensor") await hass.config_entries.async_forward_entry_unload(entry, "sensor")
await hass.config_entries.async_forward_entry_unload(entry, "binary_sensor")
await hass.config_entries.async_forward_entry_unload(entry, "siren")
context = hass.data[DOMAIN]["context"] context = hass.data[DOMAIN]["context"]
context.socket.disconnect() context.socket.disconnect()
hass.data[DOMAIN]["unsub"]()
return True return True
def connect_socket(context):
"""subscribe to georide socket"""
_LOGGER.info("Georide socket connexion")
socket = GeorideSocket()
socket.subscribe_locked(context.on_lock_callback)
socket.subscribe_device(context.on_device_callback)
socket.subscribe_position(context.on_position_callback)
context.socket = socket
socket.init()
socket.connect(context.get_token())
class GeoRideContext: class GeorideContext:
"""Hold the current GeoRide context.""" """Hold the current Georide context."""
def __init__(self, hass, email, password, token): def __init__(self, hass, email, password, token):
"""Initialize an GeoRide context.""" """Initialize an Georide context."""
self._hass = hass self._hass = hass
self._email = email self._email = email
self._password = password self._password = password
self._georide_trackers_coordoned = [] self._georide_trackers = defaultdict(set)
self._georide_trackers_beacon_coordoned = []
self._georide_trackers_beacon = []
self._georide_trackers = []
self._token = token self._token = token
self._socket = None self._socket = None
self._thread_started = False self._thread_started = False
@@ -153,7 +148,7 @@ class GeoRideContext:
def email(self): def email(self):
""" current email """ """ current email """
return self._email return self._email
@property @property
def password(self): def password(self):
""" password """ """ password """
@@ -166,37 +161,23 @@ class GeoRideContext:
@property @property
def georide_trackers(self): def georide_trackers(self):
""" GeoRide tracker list """ """ georide tracker list """
return self._georide_trackers return self._georide_trackers
@georide_trackers.setter @georide_trackers.setter
def georide_trackers(self, trackers): def georide_trackers(self, trackers):
""" GeoRide tracker list """ """ georide tracker list """
self._georide_trackers = trackers self._georide_trackers = trackers
async def connect_socket(self): def get_token(self):
"""subscribe to GeoRide socket"""
_LOGGER.info("GeoRide socket connexion")
socket = GeoRideSocket()
socket.subscribe_locked(self.on_lock_callback)
socket.subscribe_device(self.on_device_callback)
socket.subscribe_position(self.on_position_callback)
socket.subscribe_alarm(self.on_alarm_callback)
self._socket = socket
socket.init()
self._hass.async_add_executor_job(socket.connect, await self.get_token())
async def get_token(self):
""" here we return the current valid tocken """ """ here we return the current valid tocken """
jwt_data = jwt.decode(self._token, options={"verify_signature": False}) jwt_data = jwt.decode(self._token, verify=False)
exp_timestamp = jwt_data['exp'] exp_timestamp = jwt_data['exp']
epoch = math.ceil(time.time()) epoch = math.ceil(time.time())
if (exp_timestamp - TOKEN_SAFE_DAY) < epoch: if (exp_timestamp - TOKEN_SAFE_DAY) < epoch:
_LOGGER.info("Time reached, renew token") _LOGGER.info("Time reached, renew token")
account = await self._hass.async_add_executor_job(GeoRideApi.get_authorisation_token, account = GeorideApi.get_authorisation_token(self._email, self._password)
self._email, self._password)
config = self._hass.data[DOMAIN]["config"] config = self._hass.data[DOMAIN]["config"]
config[CONF_TOKEN] = account.auth_token config[CONF_TOKEN] = account.auth_token
self._token = account.auth_token self._token = account.auth_token
@@ -205,256 +186,73 @@ class GeoRideContext:
_LOGGER.info("Token exp data: %s", exp_timestamp) _LOGGER.info("Token exp data: %s", exp_timestamp)
return self._token return self._token
async def get_tracker(self, tracker_id): def get_tracker(self, tracker_id):
""" here we return last tracker by id""" """ here we return last tracker by id"""
await self.refresh_trackers() epoch_min = math.floor(time.time()/60)
if (epoch_min % MIN_UNTIL_REFRESH) == 0:
if epoch_min != self._previous_refresh:
self._previous_refresh = epoch_min
self.refresh_trackers()
if not self._thread_started:
_LOGGER.info("Start the thread")
self._hass.async_add_executor_job(connect_socket, self)
# We refresh the tracker list each hours
self._thread_started = True
for tracker in self._georide_trackers: for tracker in self._georide_trackers:
if tracker.tracker_id == tracker_id: if tracker.tracker_id == tracker_id:
return tracker return tracker
return {} return None
async def get_tracker_beacon(self, beacon_id): def refresh_trackers(self):
""" here we return last tracker_beacon by id"""
for tracker_beacon in self._georide_trackers_beacon:
if tracker_beacon.beacon_id == beacon_id:
return tracker_beacon
return {}
async def get_tracker_beacons_by_tracker_id(self, tracker_id):
""" here we return last tracker_beacon by id"""
filtered_beacon = []
for tracker_beacon in self._georide_trackers_beacon:
if tracker_beacon.linked_tracker_id == tracker_id:
filtered_beacon.append(tracker_beacon)
return filtered_beacon
async def refresh_trackers(self):
""" here we return last tracker by id"""
_LOGGER.debug("Call refresh tracker")
epoch_min = math.floor(time.time()/60)
if epoch_min != self._previous_refresh:
self._previous_refresh = epoch_min
await self.force_refresh_trackers()
for tracker in self._georide_trackers:
if tracker.is_siren_on:
if time.time() - SIREN_ACTIVATION_DELAY > tracker.siren_last_on_date:
tracker.is_siren_on = False
async def force_refresh_trackers(self):
"""Used to refresh the tracker list""" """Used to refresh the tracker list"""
_LOGGER.info("Tracker list refresh") _LOGGER.info("Tracker list refresh")
new_georide_trackers = await self._hass.async_add_executor_job(GeoRideApi.get_trackers, self._georide_trackers = GeorideApi.get_trackers(self.get_token())
await self.get_token())
for refreshed_tracker in new_georide_trackers:
found = False
for tracker in self._georide_trackers:
if tracker.tracker_id == refreshed_tracker.tracker_id:
tracker.update_all_data(refreshed_tracker)
if tracker.version > 2:
await self.force_refresh_trackers_beacon(tracker.tracker_id)
found = True
if not found:
self._georide_trackers.append(refreshed_tracker)
if refreshed_tracker.version > 2:
await self.force_refresh_trackers_beacon(refreshed_tracker.tracker_id)
if not self._thread_started:
_LOGGER.info("Start the thread")
# We refresh the tracker list each hours
self._thread_started = True
await self.connect_socket()
async def force_refresh_trackers_beacon(self, tracker_id):
"""Used to refresh the tracker list"""
_LOGGER.info("Tracker beacon refresh")
new_georide_tracker_beacons = await self._hass.async_add_executor_job(GeoRideApi.get_tracker_beacons,
await self.get_token(), tracker_id)
for new_georide_tracker_beacon in new_georide_tracker_beacons:
found = False
for tracker_beacon in self._georide_trackers_beacon:
if tracker_beacon.tracker_id == new_georide_tracker_beacon.beacon_id:
tracker_beacon.update_all_data(new_georide_tracker_beacon)
found = True
if not found:
self._georide_trackers_beacon.append(new_georide_tracker_beacon)
if not self._thread_started:
_LOGGER.info("Start the thread")
# We refresh the tracker list each hours
self._thread_started = True
await self.connect_socket()
async def init_context(self, hass):
"""Used to refresh the tracker list"""
_LOGGER.info("Init_context")
await self.force_refresh_trackers()
update_interval = timedelta(minutes=MIN_UNTIL_REFRESH)
for tracker in self._georide_trackers:
coordinator = DataUpdateCoordinator[Mapping[str, Any]](
hass,
_LOGGER,
name=tracker.tracker_name,
update_method=self.refresh_trackers,
update_interval=update_interval
)
coordoned_tracker = {
"tracker_device": Device(tracker),
"coordinator": coordinator
}
if tracker.version > 2:
tracker_beacons = await self.get_tracker_beacons_by_tracker_id(tracker.tracker_id)
for tracker_beacon in tracker_beacons:
beacon_coordinator = DataUpdateCoordinator[Mapping[str, Any]](
hass,
_LOGGER,
name= tracker_beacon.name
)
coordoned_beacon = {
"tracker_beacon": DeviceBeacon(tracker_beacon),
"coordinator": beacon_coordinator
}
self._georide_trackers_beacon_coordoned.append(coordoned_beacon)
self._georide_trackers_coordoned.append(coordoned_tracker)
@property
def georide_trackers_coordoned(self):
"""Return coordoned trackers"""
return self._georide_trackers_coordoned
@property
def georide_trackers_beacon_coordoned(self):
"""Return coordoned trackers"""
return self._georide_trackers_beacon_coordoned
@property @property
def socket(self): def socket(self):
""" hold the GeoRide socket """ """ hold the georide socket """
return self._socket return self._socket
@socket.setter @socket.setter
def socket(self, socket): def socket(self, socket):
"""set the GeoRide socket""" """set the georide socket"""
self._socket = socket self._socket = socket
@callback @callback
def on_lock_callback(self, data): def on_lock_callback(self, data):
"""on lock callback""" """on lock callback"""
_LOGGER.info("On lock received") _LOGGER.info("On lock received")
for coordoned_tracker in self._georide_trackers_coordoned: for tracker in self._georide_trackers:
tracker_device = coordoned_tracker['tracker_device']
tracker = tracker_device.tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']: if tracker.tracker_id == data['trackerId']:
tracker.locked_latitude = data['lockedLatitude'] tracker.locked_latitude = data['lockedLatitude']
tracker.locked_longitude = data['lockedLongitude'] tracker.locked_longitude = data['lockedLongitude']
tracker.is_locked = data['isLocked'] tracker.is_locked = data['isLocked']
return
event_data = {
"device_id": tracker_device.unique_id,
"device_name": tracker_device.name,
}
self._hass.bus.async_fire(f"{DOMAIN}_lock_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback @callback
def on_device_callback(self, data): def on_device_callback(self, data):
"""on device callback""" """on device callback"""
_LOGGER.info("On device received") _LOGGER.info("On device received")
for coordoned_tracker in self._georide_trackers_coordoned: for tracker in self._georide_trackers:
tracker_device = coordoned_tracker['tracker_device']
tracker = tracker_device.tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']: if tracker.tracker_id == data['trackerId']:
tracker.status = data['status'] tracker.status = data['status']
event_data = { return
"device_id": tracker_device.unique_id,
"device_name": tracker_device.name,
}
self._hass.bus.async_fire(f"{DOMAIN}_device_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback @callback
def on_alarm_callback(self, data):
"""on device callback"""
_LOGGER.info("On alarm received")
for coordoned_tracker in self._georide_trackers_coordoned:
tracker_device = coordoned_tracker['tracker_device']
tracker = tracker_device.tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']:
if data['name'] == 'vibration':
_LOGGER.info("Vibration detected")
elif data['name'] == 'exitZone':
_LOGGER.info("Exit zone detected")
elif data['name'] == 'crash':
_LOGGER.info("Crash detected")
elif data['name'] == 'crashParking':
_LOGGER.info("Crash parking detected")
elif data['name'] == 'deviceOffline':
_LOGGER.info("Device offline detected")
elif data['name'] == 'deviceOnline':
_LOGGER.info("Device online detected")
elif data['name'] == 'powerCut':
_LOGGER.info("powerCut detected")
elif data['name'] == 'powerUncut':
_LOGGER.info("powerUncut detected")
elif data['name'] == 'batteryWarning':
_LOGGER.info("batteryWarning detected")
elif data['name'] == 'temperatureWarning':
_LOGGER.info("temperatureWarning detected")
elif data['name'] == 'magnetOn':
_LOGGER.info("magnetOn detected")
elif data['name'] == 'magnetOff':
_LOGGER.info("magnetOff detected")
elif data['name'] == 'sonorAlarmOn':
_LOGGER.info("sonorAlarmOn detected")
tracker.is_siren_on = True
else:
_LOGGER.warning("Unmanaged alarm: %s", data["name"])
event_data = {
"device_id": tracker_device.unique_id,
"device_name": tracker_device.name,
"type": f"alarm_{data['name']}"
}
self._hass.bus.async_fire(f"{DOMAIN}_alarm_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback
def on_position_callback(self, data): def on_position_callback(self, data):
"""on position callback""" """on position callback"""
_LOGGER.info("On position received") _LOGGER.info("On position received")
for coordoned_tracker in self._georide_trackers_coordoned: for tracker in self._georide_trackers:
tracker = coordoned_tracker['tracker_device'].tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']: if tracker.tracker_id == data['trackerId']:
tracker.latitude = data['latitude'] tracker.latitude = data['latitude']
tracker.longitude = data['longitude'] tracker.longitude = data['longitude']
tracker.moving = data['moving'] tracker.moving = data['moving']
tracker.speed = data['speed'] tracker.speed = data['speed']
tracker.fixtime = data['fixtime'] tracker.fixtime = data['fixtime']
return
event_data = {
"device_id": tracker_device.unique_id,
"device_name": tracker_device.name,
}
self._hass.bus.async_fire(f"{DOMAIN}_position_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break

View File

@@ -1,303 +0,0 @@
""" binary_sensor for GeoRide object """
import logging
from typing import Any, Mapping
from homeassistant.core import callback
from homeassistant.helpers.entity import DeviceInfo, EntityCategory
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.components.binary_sensor import ENTITY_ID_FORMAT
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator
)
from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device, DeviceBeacon
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613
"""Set up GeoRide tracker based off an entry."""
georide_context = hass.data[GEORIDE_DOMAIN]["context"]
entities = []
coordoned_trackers = georide_context.georide_trackers_coordoned
for coordoned_tracker in coordoned_trackers:
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
entities.append(GeoRideStolenBinarySensorEntity(coordinator, tracker_device))
entities.append(GeoRideCrashedBinarySensorEntity(coordinator, tracker_device))
entities.append(GeoRideOwnerBinarySensorEntity(coordinator, tracker_device))
entities.append(GeoRideActiveSubscriptionBinarySensorEntity(coordinator, tracker_device))
entities.append(GeoRideNetworkBinarySensorEntity(coordinator, tracker_device))
entities.append(GeoRideMovingBinarySensorEntity(coordinator, tracker_device))
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
coordoned_beacons = georide_context.georide_trackers_beacon_coordoned
for coordoned_beacon in coordoned_beacons:
tracker_beacon = coordoned_beacon['tracker_beacon']
coordinator = coordoned_beacon['coordinator']
entities.append(GeoRideBeaconUpdatedBinarySensorEntity(coordinator, tracker_beacon))
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.beacon.beacon_id] = coordinator
async_add_entities(entities, True)
return True
class GeoRideBinarySensorEntity(CoordinatorEntity, BinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self.entity_id = f"{ENTITY_ID_FORMAT.format('binary_sensor')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._is_on = False
@property
def device_info(self):
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideBeaconBinarySensorEntity(CoordinatorEntity, BinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device_beacon: DeviceBeacon):
"""Set up Georide entity."""
super().__init__(coordinator)
self._tracker_device_beacon = tracker_device_beacon
self._name = tracker_device_beacon.beacon.name
self.entity_id = f"{ENTITY_ID_FORMAT.format('binary_sensor')}.{tracker_device.beacon.beacon_id}"# pylint: disable=C0301
self._is_on = False
@property
def entity_category(self):
return None
@property
def device_info(self):
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideStolenBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('is_stolen')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def unique_id(self):
"""Return the unique ID."""
return f"is_stolen_{self._tracker_device.tracker.tracker_id}"
@property
def device_class(self):
"""Return the device class."""
return "problem"
@property
def is_on(self):
"""state value property"""
return self._tracker_device.tracker.is_stolen
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} is not stolen"
class GeoRideCrashedBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('is_crashed')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def unique_id(self):
"""Return the unique ID."""
return f"is_crashed_{self._tracker_device.tracker.tracker_id}"
@property
def device_class(self):
"""Return the device class."""
return "problem"
@property
def is_on(self):
"""state value property"""
return self._tracker_device.tracker.is_crashed
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} is not crashed"
class GeoRideActiveSubscriptionBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('is_active_subscription_')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"is_active_subscription_{self._tracker_device.tracker.tracker_id}"
@property
def is_on(self):
"""state value property"""
tracker = self._tracker_device.tracker
if tracker.is_oldsubscription:
if tracker.subscription_id is not None:
return True
return False
else:
if tracker.subscription is not None and tracker.subscription.subscription_id is not None:
return True
return False
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} has an active subscription"
class GeoRideOwnerBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('is_owner')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def unique_id(self):
"""Return the unique ID."""
return f"is_owner_{self._tracker_device.tracker.tracker_id}"
@property
def is_on(self):
"""state value property"""
if self._tracker_device.tracker.role == "owner":
return True
return False
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} is own tracker"
class GeoRideNetworkBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('have_network')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"have_network_{self._tracker_device.tracker.tracker_id}"
@property
def device_class(self):
"""Return the device class."""
return "connectivity"
@property
def is_on(self):
"""state value property"""
if self._tracker_device.tracker.status == "online":
return True
return False
@property
def name(self):
""" GeoRide name """
return f"{self._name} have network"
class GeoRideMovingBinarySensorEntity(GeoRideBinarySensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('moving')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
@property
def unique_id(self):
"""Return the unique ID."""
return f"moving_{self._tracker_device.tracker.tracker_id}"
@property
def device_class(self):
"""Return the device class."""
return "moving"
@property
def is_on(self):
"""state value property"""
return self._tracker_device.tracker.moving
@property
def name(self):
""" GeoRide name """
return f"{self._name} is moving"
class GeoRideBeaconUpdatedBinarySensorEntity(GeoRideBeaconBinarySensorEntity):
"""Represent a tracked device."""
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_beacon_device: DeviceBeacon):
"""Set up Georide entity."""
super().__init__(coordinator, tracker_beacon_device)
self.entity_id = f"{ENTITY_ID_FORMAT.format('update')}.{tracker_beacon_device.tracker_beacon.beacon_id}"# pylint: disable=C0301
@property
def unique_id(self):
"""Return the unique ID."""
return f"update_{self._tracker_beacon_device.beacon.beacon_id}"
@property
def device_class(self):
"""Return the device class."""
return "update"
@property
def is_on(self):
"""state value property"""
return not self._tracker_beacon_device.beacon.is_updated
@property
def name(self):
""" GeoRide name """
return f"{self._name} have an update"

View File

@@ -3,21 +3,21 @@
import logging import logging
from homeassistant import config_entries from homeassistant import config_entries
import voluptuous as vol import voluptuous as vol
import georideapilib.api as GeoRideApi import georideapilib.api as GeorideApi
import georideapilib.exception as GeoRideException import georideapilib.exception as GeorideException
from .const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN, DOMAIN from .const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@config_entries.HANDLERS.register("georide") @config_entries.HANDLERS.register("georide")
class GeoRideConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class GeorideConfigFlow(config_entries.ConfigFlow):
"""GeoRide config flow """ """Georide config flow """
async def async_step_user(self, user_input=None): #pylint: disable=W0613 async def async_step_user(self, user_input=None): #pylint: disable=W0613
""" handle info to help to configure GeoRide """ """ handle info to help to configure georide """
if self._async_current_entries(): if self._async_current_entries():
return self.async_abort(reason="one_instance_allowed") return self.async_abort(reason="one_instance_allowed")
@@ -54,14 +54,14 @@ class GeoRideConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
password = user_input[CONF_PASSWORD] password = user_input[CONF_PASSWORD]
try: try:
account = await self.hass.async_add_executor_job(GeoRideApi.get_authorisation_token, email, password) account = GeorideApi.get_authorisation_token(email, password)
data = { data = {
CONF_EMAIL: email, CONF_EMAIL: email,
CONF_PASSWORD: password, CONF_PASSWORD: password,
CONF_TOKEN: account.auth_token CONF_TOKEN: account.auth_token
} }
return self.async_create_entry(title=email, data=data) return self.async_create_entry(title=email, data=data)
except (GeoRideException.SeverException, GeoRideException.LoginException): except (GeorideException.SeverException, GeorideException.LoginException):
_LOGGER.error("Invalid credentials provided, config not created") _LOGGER.error("Invalid credentials provided, config not created")
errors = {"base": "faulty_credentials"} errors = {"base": "faulty_credentials"}
return self.async_show_form(step_id="georide_login", data_schema=schema, errors=errors) return self.async_show_form(step_id="georide_login", data_schema=schema, errors=errors)

View File

@@ -1,4 +1,4 @@
""" GeoRide const file """ """ Georide const file """
DOMAIN = "georide" DOMAIN = "georide"
CONF_EMAIL = "email" CONF_EMAIL = "email"
@@ -10,6 +10,4 @@ TRACKER_ID = "trackerId"
MIN_UNTIL_REFRESH = 10 MIN_UNTIL_REFRESH = 10
SIREN_ACTIVATION_DELAY = 30
TOKEN_SAFE_DAY = 432000 # five days TOKEN_SAFE_DAY = 432000 # five days

View File

@@ -1,110 +0,0 @@
"""Home Assistant representation of an GeoRide Tracker device."""
from georideapilib.objects import GeoRideTracker, GeoRideTrackerBeacon
from .const import DOMAIN as GEORIDE_DOMAIN
class Device:
"""Home Assistant representation of a GeoRide Tracker device."""
def __init__(self, tracker):
"""Initialize GeoRideTracker device."""
self._tracker: GeoRideTracker = tracker
@property
def tracker(self):
"""return the tracker"""
return self._tracker
@property
def name(self) -> str:
"""Get the name."""
return self._tracker.tracker_name
@property
def manufacturer(self) -> str:
"""Get the manufacturer."""
return "GeoRide"
@property
def model_name(self) -> str:
"""Get the model name."""
name = None
if self._tracker.version == 1:
name = "GeoRide 1"
elif self._tracker.version == 2:
name = "GeoRide 2"
elif self._tracker.version == 3:
name = "GeoRide 3"
else:
name = "Prototype / Unknown"
return name
@property
def device_info(self):
"""Return the device info."""
return {
"name": self.name,
"identifiers": self.unique_id,
"manufacturer": "GeoRide",
"model": self.model_name,
"suggested_area": "Garage"
}
@property
def unique_id(self) -> str:
"""Get the unique id."""
return {(GEORIDE_DOMAIN, self._tracker.tracker_id)}
def __str__(self) -> str:
"""Get string representation."""
return f"GeoRide Device: {self.name}::{self.model_name}::{self.unique_id}"
class DeviceBeacon:
"""Home Assistant representation of a GeoRide Tracker device."""
def __init__(self, beacon):
"""Initialize GeoRideTracker device."""
self._beacon: GeoRideTrackerBeacon = beacon
@property
def beacon(self):
"""return the tracker beacon"""
return self._beacon
@property
def name(self) -> str:
"""Get the name."""
return self._beacon.name
@property
def manufacturer(self) -> str:
"""Get the manufacturer."""
return "GeoRide"
@property
def model_name(self) -> str:
"""Get the model name."""
name = "GeoRide Beacon"
return name
@property
def device_info(self):
"""Return the device info."""
return {
"name": self.name,
"identifiers": self.unique_id,
"manufacturer": "GeoRide",
"model": self.model_name,
"suggested_area": "Garage"
}
@property
def unique_id(self) -> str:
"""Get the unique id."""
return {(GEORIDE_DOMAIN, self._beacon.beacon_id)}
def __str__(self) -> str:
"""Get string representation."""
return f"GeoRide Device: {self.name}::{self.model_name}::{self.unique_id}"

View File

@@ -1,17 +1,12 @@
""" device tracker for GeoRide object """ """ device tracker for Georide object """
import logging import logging
from typing import Any, Mapping
from homeassistant.components.device_tracker.const import DOMAIN, SOURCE_TYPE_GPS from homeassistant.components.device_tracker.const import DOMAIN, SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity from homeassistant.components.device_tracker.config_entry import TrackerEntity
from homeassistant.helpers.update_coordinator import ( import georideapilib.api as GeorideApi
CoordinatorEntity,
DataUpdateCoordinator,
)
from .device import Device
from .const import DOMAIN as GEORIDE_DOMAIN from .const import DOMAIN as GEORIDE_DOMAIN
@@ -19,67 +14,73 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613 async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613
"""Set up Georide tracker based off an entry.""" """Set up Georide tracker based off an entry."""
georide_context = hass.data[GEORIDE_DOMAIN]["context"] georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.georide_trackers_coordoned
if georide_context.get_token() is None:
return False
entities = [] _LOGGER.info('Current georide token: %s', georide_context.get_token())
for coordoned_tracker in coordoned_trackers:
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
entity = GeoRideTrackerEntity(coordinator, tracker_device, hass)
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
entities.append(entity)
async_add_entities(entities)
trackers = GeorideApi.get_trackers(georide_context.get_token())
tracker_entities = []
for tracker in trackers:
entity = GeorideTrackerEntity(tracker.tracker_id, georide_context.get_token,
georide_context.get_tracker, tracker)
hass.data[GEORIDE_DOMAIN]["devices"][tracker.tracker_id] = entity
tracker_entities.append(entity)
async_add_entities(tracker_entities)
return True return True
class GeoRideTrackerEntity(CoordinatorEntity, TrackerEntity): class GeorideTrackerEntity(TrackerEntity):
"""Represent a tracked device.""" """Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]], def __init__(self, tracker_id, get_token_callback, get_tracker_callback, tracker):
tracker_device: Device, hass): """Set up Georide entity."""
"""Set up GeoRide entity.""" self._tracker_id = tracker_id
super().__init__(coordinator) self._get_token_callback = get_token_callback
self._name = tracker_device.tracker.tracker_name self._get_tracker_callback = get_tracker_callback
self._tracker_device = tracker_device self._name = tracker.tracker_name
self.entity_id = DOMAIN + ".{}".format(tracker_device.tracker.tracker_id) self._data = tracker or {}
self._hass = hass self.entity_id = DOMAIN + ".{}".format(tracker_id)
@property
def entity_category(self):
return None
@property @property
def unique_id(self): def unique_id(self):
"""Return the unique ID.""" """Return the unique ID."""
return f"georide_tracker_{self._tracker_device.tracker.tracker_id}" return self._tracker_id
@property @property
def name(self): def name(self):
""" GeoRide odometer name """ return self._name
return f"{self._name} position"
@property @property
def latitude(self): def latitude(self):
"""Return latitude value of the device.""" """Return latitude value of the device."""
if self._tracker_device.tracker.latitude: if self._data.latitude:
return self._tracker_device.tracker.latitude return self._data.latitude
return None return None
@property @property
def longitude(self): def longitude(self):
"""Return longitude value of the device.""" """Return longitude value of the device."""
if self._tracker_device.tracker.longitude: if self._data.longitude:
return self._tracker_device.tracker.longitude return self._data.longitude
return None return None
@property @property
def source_type(self): def source_type(self):
"""Return the source type, eg gps or router, of the device.""" """Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS return SOURCE_TYPE_GPS
@property @property
def location_accuracy(self): def location_accuracy(self):
""" return the gps accuracy of georide (could not be aquired, then 10) """ """ return the gps accuracy of georide (could not be aquired, then 10) """
@@ -87,10 +88,38 @@ class GeoRideTrackerEntity(CoordinatorEntity, TrackerEntity):
@property @property
def icon(self): def icon(self):
"""return the entity icon"""
return "mdi:map-marker" return "mdi:map-marker"
@property @property
def device_info(self): def device_info(self):
"""Return the device info.""" """Return the device info."""
return self._tracker_device.device_info return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker_id)},
"manufacturer": "GeoRide",
"odometer": "{} km".format(self._data.odometer)
}
@property
def get_tracker_callback(self):
""" get tracker callaback"""
return self._get_tracker_callback
@property
def get_token_callback(self):
""" get token callaback"""
return self._get_token_callback
@property
def should_poll(self):
"""No polling needed."""
return True
def update(self):
""" update the current tracker"""
_LOGGER.info('update')
self._data = self._get_tracker_callback(self._tracker_id)
self._name = self._data.tracker_name

View File

@@ -1,15 +1,12 @@
{ {
"domain": "georide", "domain": "georide",
"name": "GeoRide", "name": "Georide",
"config_flow": true, "config_flow": true,
"documentation": "https://github.com/ptimatth/GeorideHA", "documentation": "https://github.com/ptimatth/GeorideHA",
"issue_tracker": "https://github.com/ptimatth/GeorideHA/issues",
"iot_class": "cloud_polling",
"requirements": [ "requirements": [
"georideapilib>=0.8.2", "georideapilib>=0.4.4",
"pyjwt==2.1.0" "pyjwt>=1.7.1"
], ],
"dependencies": [], "dependencies": [],
"codeowners": ["ptimatth"], "codeowners": ["ptimatth"]
"version": "0.9.0"
} }

View File

@@ -1,329 +1,102 @@
""" odometter sensor for GeoRide object """ """ odometter sensor for Georide object """
import logging import logging
from typing import Any, Mapping
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.helpers.entity import DeviceInfo, EntityCategory from homeassistant.components.switch import SwitchEntity
from homeassistant.components.sensor import SensorEntity from homeassistant.components.switch import ENTITY_ID_FORMAT
from homeassistant.components.sensor import ENTITY_ID_FORMAT
from homeassistant.helpers.update_coordinator import ( import georideapilib.api as GeorideApi
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import DOMAIN as GEORIDE_DOMAIN from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device, DeviceBeacon
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613 async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613
"""Set up GeoRide tracker based off an entry.""" """Set up Georide tracker based off an entry."""
georide_context = hass.data[GEORIDE_DOMAIN]["context"] georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.georide_trackers_coordoned
entities = [] if georide_context.get_token() is None:
for coordoned_tracker in coordoned_trackers: return False
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
entities.append(GeoRideOdometerSensorEntity(coordinator, tracker_device, hass))
entities.append(GeoRideOdometerKmSensorEntity(coordinator, tracker_device, hass))
entities.append(GeoRideFixtimeSensorEntity(coordinator, tracker_device))
if tracker_device.tracker.version > 2:
entities.append(GeoRideInternalBatterySensorEntity(coordinator, tracker_device))
entities.append(GeoRideExternalBatterySensorEntity(coordinator, tracker_device))
coordoned_beacons = georide_context.georide_trackers_beacon_coordoned
for coordoned_beacon in coordoned_beacons:
tracker_beacon = coordoned_beacon['tracker_beacon']
coordinator = coordoned_beacon['coordinator']
entities.append(GeoRideBeaconBatterySensorEntity(coordinator, tracker_beacon))
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.beacon.beacon_id] = coordinator
async_add_entities(entities) trackers = GeorideApi.get_trackers(georide_context.get_token())
odometer_switch_entities = []
for tracker in trackers:
entity = GeorideOdometerSensorEntity(tracker.tracker_id, georide_context.get_token,
georide_context.get_tracker, data=tracker)
hass.data[GEORIDE_DOMAIN]["devices"][tracker.tracker_id] = entity
odometer_switch_entities.append(entity)
async_add_entities(odometer_switch_entities)
return True return True
class GeoRideOdometerSensorEntity(CoordinatorEntity, SensorEntity): class GeorideOdometerSensorEntity(SwitchEntity):
"""Represent a tracked device.""" """Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]], def __init__(self, tracker_id, get_token_callback, get_tracker_callback, data):
tracker_device:Device, hass): """Set up Georide entity."""
"""Set up GeoRide entity.""" self._tracker_id = tracker_id
super().__init__(coordinator) self._data = data or {}
self._tracker_device = tracker_device self._get_token_callback = get_token_callback
self._name = tracker_device.tracker.tracker_name self._get_tracker_callback = get_tracker_callback
self._name = data.tracker_name
self._unit_of_measurement = "m" self._unit_of_measurement = "m"
self.entity_id = f"{ENTITY_ID_FORMAT.format('odometer')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self.entity_id = ENTITY_ID_FORMAT.format("odometer") + "." + str(tracker_id)
self._state = 0 self._state = 0
self._hass = hass
@property
def entity_category(self): def update(self):
return None """ update the current tracker"""
_LOGGER.info('update')
self._data = self._get_tracker_callback(self._tracker_id)
self._name = self._data.tracker_name
self._state = self._data.odometer
@property @property
def unique_id(self): def unique_id(self):
"""Return the unique ID.""" """Return the unique ID."""
return f"odometer_{self._tracker_device.tracker.tracker_id}" return self._tracker_id
@property
def state(self):
"""state property"""
odometer = self._tracker_device.tracker.odometer
return odometer
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
return self._unit_of_measurement
@property @property
def name(self): def name(self):
""" GeoRide odometer name """ """ Georide switch name """
return f"{self._name} odometer" return self._name
@property
def state(self):
return self._state
@property
def unit_of_measurement(self):
return self._unit_of_measurement
@property
def get_token_callback(self):
""" Georide switch token callback method """
return self._get_token_callback
@property
def get_tracker_callback(self):
""" Georide switch token callback method """
return self._get_tracker_callback
@property @property
def icon(self): def icon(self):
"""icon getter"""
return "mdi:counter" return "mdi:counter"
@property @property
def device_info(self) -> DeviceInfo: def device_info(self):
"""Return the device info.""" """Return the device info."""
return self._tracker_device.device_info return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker_id)},
"manufacturer": "GeoRide"
}
class GeoRideOdometerKmSensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device, hass):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self._unit_of_measurement = "km"
self.entity_id = f"{ENTITY_ID_FORMAT.format('odometer_km')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._state = 0
self._hass = hass
@property
def unique_id(self):
"""Return the unique ID."""
return f"odometer_km_{self._tracker_device.tracker.tracker_id}"
@property
def state(self):
"""state property"""
odometer = self._tracker_device.tracker.odometer // 1000
return odometer
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
return self._unit_of_measurement
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} odometer km"
@property
def icon(self):
"""icon getter"""
return "mdi:counter"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideInternalBatterySensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self._unit_of_measurement = "V"
self.entity_id = f"{ENTITY_ID_FORMAT.format('internal_battery_voltage')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._state = 0
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"internal_battery_voltage_{self._tracker_device.tracker.tracker_id}"
@property
def state(self):
"""state property"""
return self._tracker_device.tracker.internal_battery_voltage
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
return self._unit_of_measurement
@property
def name(self):
""" GeoRide internal_battery_voltage name """
return f"{self._name} internal_battery_voltage"
@property
def icon(self):
"""icon getter"""
return "mdi:battery"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideExternalBatterySensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self._unit_of_measurement = "V"
self.entity_id = f"{ENTITY_ID_FORMAT.format('external_battery_voltage')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._state = 0
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"external_battery_voltage_{self._tracker_device.tracker.tracker_id}"
@property
def state(self):
"""state property"""
return self._tracker_device.tracker.external_battery_voltage
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
return self._unit_of_measurement
@property
def name(self):
""" GeoRide internal_battery_voltage name """
return f"{self._name} external_battery_voltage"
@property
def icon(self):
"""icon getter"""
return "mdi:battery"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideFixtimeSensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self.entity_id = f"{ENTITY_ID_FORMAT.format('fixtime')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._state = 0
self._device_class = "timestamp"
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"fixtime_{self._tracker_device.tracker.tracker_id}"
@property
def state(self):
"""state property"""
return self._tracker_device.tracker.fixtime
@property
def name(self):
""" GeoRide fixtime name """
return f"{self._name} last fixed position"
@property
def icon(self):
"""icon getter"""
return "mdi:map-clock"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._tracker_device.device_info
class GeoRideBeaconBatterySensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_beacon:DeviceBeacon):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self._unit_of_measurement = "%"
self.entity_id = f"{ENTITY_ID_FORMAT.format('battery_percent')}.{tracker_beacon.beacon.beacon_id}"# pylint: disable=C0301
self._state = 0
@property
def entity_category(self):
return EntityCategory.DIAGNOSTIC
@property
def unique_id(self):
"""Return the unique ID."""
return f"battery_percent_{self._tracker_device.beacon.beacon_id}"
@property
def state(self):
"""state property"""
return self._tracker_device.beacon.battery_level
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
return self._unit_of_measurement
@property
def name(self):
""" GeoRide internal_battery_voltage name """
return f"{self._name} battery percent"
@property
def icon(self):
"""icon getter"""
return "mdi:battery"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return self._tracker_device.device_info

View File

@@ -1,96 +0,0 @@
""" device tracker for GeoRide object """
import logging
from typing import Any, Mapping
from homeassistant.components.siren import SirenEntity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
import georideapilib.api as GeoRideApi
from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device
ENTITY_ID_FORMAT = GEORIDE_DOMAIN + ".{}"
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613
"""Set up GeoRide tracker based off an entry."""
georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.georide_trackers_coordoned
entities = []
for coordoned_tracker in coordoned_trackers:
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
if tracker_device.tracker.version > 2:
entities.append(GeoRideSirenEntity(coordinator, tracker_device, hass))
async_add_entities(entities)
return True
class GeoRideSirenEntity(CoordinatorEntity, SirenEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device, hass):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self.entity_id = f"{ENTITY_ID_FORMAT.format('eco_mode')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._hass = hass
@property
def entity_category(self):
return None
async def async_turn_on(self, **kwargs):
""" lock the GeoRide tracker """
_LOGGER.info('async_turn_on eco %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.change_tracker_siren_state,
token, self._tracker_device.tracker.tracker_id, True)
if success:
self._tracker_device.tracker.is_siren_on = True
async def async_turn_off(self, **kwargs):
""" unlock the GeoRide tracker """
_LOGGER.info('async_turn_off eco %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.change_tracker_siren_state,
token, self._tracker_device.tracker.tracker_id, False)
if success:
self._tracker_device.tracker.is_siren_on = False
@property
def unique_id(self):
"""Return the unique ID."""
return f"siren_{self._tracker_device.tracker.tracker_id}"
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} siren"
@property
def is_on(self):
""" GeoRide switch status """
return self._tracker_device.tracker.is_siren_on
@property
def device_info(self):
"""Return the device info."""
return self._tracker_device.device_info

View File

@@ -4,7 +4,7 @@
"step": { "step": {
"georide_login": { "georide_login": {
"title": "Set up GeoRide", "title": "Set up GeoRide",
"description": "Sign-in to GeoRide", "description": "Sign-in to georide",
"data": { "data": {
"email": "email", "email": "email",
"password": "password" "password": "password"

View File

@@ -1,175 +1,129 @@
""" device tracker for GeoRide object """ """ device tracker for Georide object """
import logging import logging
from homeassistant.core import callback
from typing import Any, Mapping
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.components.switch import ENTITY_ID_FORMAT from homeassistant.components.switch import ENTITY_ID_FORMAT
from homeassistant.helpers.update_coordinator import ( import georideapilib.api as GeorideApi
CoordinatorEntity,
DataUpdateCoordinator,
)
import georideapilib.api as GeoRideApi
from .const import DOMAIN as GEORIDE_DOMAIN from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613 async def async_setup_entry(hass, config_entry, async_add_entities): # pylint: disable=W0613
"""Set up GeoRide tracker based off an entry.""" """Set up Georide tracker based off an entry."""
georide_context = hass.data[GEORIDE_DOMAIN]["context"] georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.georide_trackers_coordoned
entities = [] if georide_context.get_token() is None:
for coordoned_tracker in coordoned_trackers: return False
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
entities.append(GeoRideLockSwitchEntity(coordinator, tracker_device, hass))
if tracker_device.tracker.version > 2:
entities.append(GeoRideEcoModeSwitchEntity(coordinator, tracker_device, hass))
async_add_entities(entities)
_LOGGER.info('Current georide token: %s', georide_context.get_token())
trackers = GeorideApi.get_trackers(georide_context.get_token())
lock_switch_entities = []
for tracker in trackers:
entity = GeorideLockSwitchEntity(tracker.tracker_id, georide_context.get_token,
georide_context.get_tracker, data=tracker)
hass.data[GEORIDE_DOMAIN]["devices"][tracker.tracker_id] = entity
lock_switch_entities.append(entity)
async_add_entities(lock_switch_entities)
return True return True
class GeoRideLockSwitchEntity(CoordinatorEntity, SwitchEntity):
class GeorideLockSwitchEntity(SwitchEntity):
"""Represent a tracked device.""" """Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]], def __init__(self, tracker_id, get_token_callback, get_tracker_callback, data):
tracker_device:Device, hass): """Set up Georide entity."""
"""Set up GeoRide entity.""" self._tracker_id = tracker_id
super().__init__(coordinator) self._data = data or {}
self._tracker_device = tracker_device self._get_token_callback = get_token_callback
self._name = tracker_device.tracker.tracker_name self._get_tracker_callback = get_tracker_callback
self.entity_id = f"{ENTITY_ID_FORMAT.format('lock')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301 self._name = data.tracker_name
self._hass = hass self._is_on = data.is_locked
self.entity_id = ENTITY_ID_FORMAT.format("lock") +"." + str(tracker_id)
self._state = {}
@property
def entity_category(self):
return None
async def async_turn_on(self, **kwargs): def turn_on(self, **kwargs):
""" lock the GeoRide tracker """ """ lock the georide tracker """
_LOGGER.info('async_turn_on %s', kwargs) _LOGGER.info('async_turn_on %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"] success = GeorideApi.lock_tracker(self._get_token_callback(), self._tracker_id)
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.lock_tracker,
token, self._tracker_device.tracker.tracker_id)
if success: if success:
self._tracker_device.tracker.is_locked = True self._data.is_locked = True
self._is_on = True
async def async_turn_off(self, **kwargs):
""" unlock the GeoRide tracker """ def turn_off(self, **kwargs):
""" unlock the georide tracker """
_LOGGER.info('async_turn_off %s', kwargs) _LOGGER.info('async_turn_off %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"] success = GeorideApi.unlock_tracker(self._get_token_callback(), self._tracker_id)
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.unlock_tracker,
token, self._tracker_device.tracker.tracker_id)
if success: if success:
self._tracker_device.tracker.is_locked = False self._data.is_locked = False
self._is_on = False
async def async_toggle(self, **kwargs): async def async_toggle(self, **kwargs):
""" toggle lock the georide tracker """ """ toggle lock the georide tracker """
_LOGGER.info('async_toggle %s', kwargs) _LOGGER.info('async_toggle %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"] result = GeorideApi.toogle_lock_tracker(self._get_token_callback(),
token = await georide_context.get_token() self._tracker_id)
result = await self._hass.async_add_executor_job(GeoRideApi.toogle_lock_tracker, self._data.is_locked = result
token, self._tracker_device.tracker.tracker_id) self._is_on = result
self._tracker_device.tracker.is_locked = result
def update(self):
""" update the current tracker"""
_LOGGER.info('update')
self._data = self._get_tracker_callback(self._tracker_id)
self._name = self._data.tracker_name
self._is_on = self._data.is_locked
@property @property
def unique_id(self): def unique_id(self):
"""Return the unique ID.""" """Return the unique ID."""
return f"lock_{self._tracker_device.tracker.tracker_id}" return self._tracker_id
@property @property
def name(self): def name(self):
""" GeoRide odometer name """ """ Georide switch name """
return f"{self._name} lock" return self._name
@property @property
def is_on(self): def is_on(self):
""" GeoRide switch status """ """ Georide switch status """
return self._tracker_device.tracker.is_locked return self._is_on
@property
def get_token_callback(self):
""" Georide switch token callback method """
return self._get_token_callback
@property
def get_tracker_callback(self):
""" Georide switch token callback method """
return self._get_tracker_callback
@property @property
def icon(self): def icon(self):
"""return the entity icon""" if self._is_on:
if self._tracker_device.tracker.is_locked:
return "mdi:lock" return "mdi:lock"
return "mdi:lock-open" return "mdi:lock-open"
@property @property
def device_info(self): def device_info(self):
"""Return the device info.""" """Return the device info."""
return self._tracker_device.device_info return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker_id)},
"manufacturer": "GeoRide"
}
class GeoRideEcoModeSwitchEntity(CoordinatorEntity, SwitchEntity):
"""Represent a tracked device."""
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device, hass):
"""Set up GeoRide entity."""
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self.entity_id = f"{ENTITY_ID_FORMAT.format('eco_mode')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._hass = hass
@property
def entity_category(self):
return None
async def async_turn_on(self, **kwargs):
""" lock the GeoRide tracker """
_LOGGER.info('async_turn_on eco %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.change_tracker_eco_mode_state,
token, self._tracker_device.tracker.tracker_id, True)
if success:
self._tracker_device.tracker.is_in_eco = True
async def async_turn_off(self, **kwargs):
""" unlock the GeoRide tracker """
_LOGGER.info('async_turn_off eco %s', kwargs)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
token = await georide_context.get_token()
success = await self._hass.async_add_executor_job(GeoRideApi.change_tracker_eco_mode_state,
token, self._tracker_device.tracker.tracker_id, False)
if success:
self._tracker_device.tracker.is_in_eco = False
@property
def unique_id(self):
"""Return the unique ID."""
return f"eco_mode_{self._tracker_device.tracker.tracker_id}"
@property
def name(self):
""" GeoRide odometer name """
return f"{self._name} eco mode"
@property
def is_on(self):
""" GeoRide switch status """
return self._tracker_device.tracker.is_in_eco
@property
def icon(self):
"""return the entity icon"""
if self._tracker_device.tracker.is_in_eco:
return "mdi:battery-heart-variant"
return "mdi:battery"
@property
def device_info(self):
"""Return the device info."""
return self._tracker_device.device_info

View File

@@ -4,7 +4,7 @@
"step": { "step": {
"georide_login": { "georide_login": {
"title": "Set up GeoRide", "title": "Set up GeoRide",
"description": "Sign-in to GeoRide", "description": "Sign-in to georide",
"data": { "data": {
"email": "email", "email": "email",
"password": "password" "password": "password"

View File

@@ -1,8 +1,8 @@
{ {
"name": "GeoRide integration", "name": "Georide integration",
"content_in_root": false, "content_in_root": false,
"render_readme": true, "render_readme": true,
"domains": ["devices_tracker", "sensor"], "domains": ["devices_tracker", "sensor"],
"country": ["FR"], "country": ["FR"],
"homeassistant": "2022.2.0" "homeassistant": "0.110.0"
} }