Shipping v0.6.0

master 0.6.0
Matthieu DUVAL 4 years ago
commit b84a4cfd7f

@ -1,7 +1,9 @@
""" georide custom conpennt """
from collections import defaultdict
import asyncio
import logging
from typing import Any, Mapping
from datetime import timedelta
import math
import time
@ -11,10 +13,10 @@ import voluptuous as vol
import jwt
from aiohttp.web import json_response
from georideapilib.objects import GeorideAccount as GeoRideAccount
from georideapilib.objects import GeoRideAccount
import georideapilib.api as GeoRideApi
from georideapilib.socket import GeorideSocket as GeoRideSocket
from georideapilib.socket import GeoRideSocket
from homeassistant import config_entries
from homeassistant.const import CONF_WEBHOOK_ID
@ -23,7 +25,14 @@ import homeassistant.helpers.config_validation as cv
import homeassistant.helpers.event as ha_event
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
from .const import (
CONF_EMAIL,
CONF_PASSWORD,
@ -35,6 +44,7 @@ from .const import (
)
_LOGGER = logging.getLogger(__name__)
@ -54,7 +64,7 @@ async def async_setup(hass, config):
hass.data[DOMAIN] = {"config": config[DOMAIN], "devices": {}, "unsub": None}
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
DOMAIN,
context={
"source": config_entries.SOURCE_IMPORT
},
@ -78,14 +88,14 @@ async def async_setup_entry(hass, entry):
password,
token
)
_LOGGER.info("Context-setup and start the thread")
_LOGGER.info("Thread started")
hass.data[DOMAIN]["context"] = context
# We add trackers to the context
context.refresh_trackers()
await context.init_context(hass)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "device_tracker"))
@ -93,7 +103,8 @@ async def async_setup_entry(hass, entry):
hass.config_entries.async_forward_entry_setup(entry, "switch"))
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "sensor"))
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "binary_sensor"))
return True
@ -103,26 +114,14 @@ async def async_unload_entry(hass, entry):
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, "sensor")
await hass.config_entries.async_forward_entry_unload(entry, "binary_sensor")
context = hass.data[DOMAIN]["context"]
context.socket.disconnect()
hass.data[DOMAIN]["unsub"]()
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:
@ -133,7 +132,8 @@ class GeoRideContext:
self._hass = hass
self._email = email
self._password = password
self._georide_trackers = defaultdict(set)
self._georide_trackers_coordoned = []
self._georide_trackers = []
self._token = token
self._socket = None
self._thread_started = False
@ -147,7 +147,7 @@ class GeoRideContext:
def email(self):
""" current email """
return self._email
@property
def password(self):
""" password """
@ -166,9 +166,23 @@ class GeoRideContext:
@georide_trackers.setter
def georide_trackers(self, trackers):
""" GeoRide tracker list """
self._georide_trackers = trackers
self._georide_trackers = trackers
def get_token(self):
async def connect_socket(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 """
jwt_data = jwt.decode(self._token, verify=False)
exp_timestamp = jwt_data['exp']
@ -176,7 +190,8 @@ class GeoRideContext:
epoch = math.ceil(time.time())
if (exp_timestamp - TOKEN_SAFE_DAY) < epoch:
_LOGGER.info("Time reached, renew token")
account = GeoRideApi.get_authorisation_token(self._email, self._password)
account = await self._hass.async_add_executor_job(GeoRideApi.get_authorisation_token,
self._email, self._password)
config = self._hass.data[DOMAIN]["config"]
config[CONF_TOKEN] = account.auth_token
self._token = account.auth_token
@ -185,36 +200,77 @@ class GeoRideContext:
_LOGGER.info("Token exp data: %s", exp_timestamp)
return self._token
def get_tracker(self, tracker_id):
async def get_tracker(self, tracker_id):
""" here we return last tracker by id"""
await self.refresh_trackers()
for tracker in self._georide_trackers:
if tracker.tracker_id == tracker_id:
return tracker
return {}
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 % MIN_UNTIL_REFRESH) == 0:
if epoch_min != self._previous_refresh:
self._previous_refresh = epoch_min
self.refresh_trackers()
#if (epoch_min % MIN_UNTIL_REFRESH) == 0:
if epoch_min != self._previous_refresh:
self._previous_refresh = epoch_min
await self.force_refresh_trackers()
#else:
# _LOGGER.debug("We wil dont refresh the tracker list")
async def force_refresh_trackers(self):
"""Used to refresh the tracker list"""
_LOGGER.info("Tracker list refresh")
new_georide_trackers = await self._hass.async_add_executor_job(GeoRideApi.get_trackers,
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)
found = True
if not found:
self._georide_trackers.append(refreshed_tracker)
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
await self.connect_socket()
for tracker in self._georide_trackers:
if tracker.tracker_id == tracker_id:
return tracker
return None
def refresh_trackers(self):
async def init_context(self, hass):
"""Used to refresh the tracker list"""
_LOGGER.info("Tracker list refresh")
self._georide_trackers = GeoRideApi.get_trackers(self.get_token())
_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
)
self._georide_trackers_coordoned.append({
"tracker_device": Device(tracker),
"coordinator": coordinator
})
def get_coordoned_trackers(self):
"""Return coordoned trackers"""
return self._georide_trackers_coordoned
@property
def socket(self):
""" hold the GeoRide socket """
return self._socket
@socket.setter
def socket(self, socket):
"""set the GeoRide socket"""
@ -224,34 +280,103 @@ class GeoRideContext:
def on_lock_callback(self, data):
"""on lock callback"""
_LOGGER.info("On lock received")
for tracker in self._georide_trackers:
for coordoned_tracker in self._georide_trackers_coordoned:
tracker = coordoned_tracker['tracker_device'].tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']:
tracker.locked_latitude = data['lockedLatitude']
tracker.locked_longitude = data['lockedLongitude']
tracker.is_locked = data['isLocked']
return
event_data = {
"device_id": tracker.tracker_id,
"type": "on_lock",
}
self._hass.bus.async_fire(f"{DOMAIN}_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback
def on_device_callback(self, data):
"""on device callback"""
_LOGGER.info("On device received")
for tracker in self._georide_trackers:
for coordoned_tracker in self._georide_trackers_coordoned:
tracker = coordoned_tracker['tracker_device'].tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']:
tracker.status = data['status']
return
event_data = {
"device_id": tracker.tracker_id,
"type": "on_device",
}
self._hass.bus.async_fire(f"{DOMAIN}_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback
def on_alarm_callback(self, data):
"""on device callback"""
_LOGGER.info("On alarm received")
for coordoned_tracker in self._georide_trackers_coordoned:
tracker = coordoned_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")
else:
_LOGGER.warning("Unamanged alarm: %s", data.name)
event_data = {
"device_id": tracker.tracker_id,
"type": f"alarm_{data.name}",
}
self._hass.bus.async_fire(f"{DOMAIN}_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break
@callback
def on_position_callback(self, data):
"""on position callback"""
_LOGGER.info("On position received")
for tracker in self._georide_trackers:
for coordoned_tracker in self._georide_trackers_coordoned:
tracker = coordoned_tracker['tracker_device'].tracker
coordinator = coordoned_tracker['coordinator']
if tracker.tracker_id == data['trackerId']:
tracker.latitude = data['latitude']
tracker.longitude = data['longitude']
tracker.moving = data['moving']
tracker.speed = data['speed']
tracker.fixtime = data['fixtime']
return
event_data = {
"device_id": tracker.tracker_id,
"type": "position",
}
self._hass.bus.async_fire(f"{DOMAIN}_event", event_data)
asyncio.run_coroutine_threadsafe(
coordinator.async_request_refresh(), self._hass.loop
).result()
break

@ -0,0 +1,157 @@
""" binary_sensor for GeoRide object """
import logging
from typing import Any, Mapping
from homeassistant.core import callback
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
_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.get_coordoned_trackers()
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))
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_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 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 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 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 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 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 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"""
if self._tracker_device.tracker.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"

@ -7,13 +7,13 @@ import georideapilib.api as GeoRideApi
import georideapilib.exception as GeoRideException
from .const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN
from .const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN, DOMAIN
_LOGGER = logging.getLogger(__name__)
@config_entries.HANDLERS.register("georide")
class GeoRideConfigFlow(config_entries.ConfigFlow):
class GeoRideConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""GeoRide config flow """
async def async_step_user(self, user_input=None): #pylint: disable=W0613
@ -54,7 +54,7 @@ class GeoRideConfigFlow(config_entries.ConfigFlow):
password = user_input[CONF_PASSWORD]
try:
account = GeoRideApi.get_authorisation_token(email, password)
account = await self.hass.async_add_executor_job(GeoRideApi.get_authorisation_token, email, password)
data = {
CONF_EMAIL: email,
CONF_PASSWORD: password,

@ -0,0 +1,57 @@
"""Home Assistant representation of an GeoRide Tracker device."""
import georideapilib.objects as GeoRideTracker
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 = "GeoRide 1"
if self._tracker.is_old_tracker:
name = "Prototype / GeoRide 1"
elif self._tracker.is_second_gen:
name = "GeoRide 2 / GeoRide 3"
return name
@property
def device_info(self):
"""Return the device info."""
return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker.tracker_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"

@ -1,12 +1,17 @@
""" device tracker for GeoRide object """
import logging
from typing import Any, Mapping
from homeassistant.components.device_tracker.const import DOMAIN, SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity
import georideapilib.api as GeoRideApi
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .device import Device
from .const import DOMAIN as GEORIDE_DOMAIN
@ -14,73 +19,63 @@ _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"]
if georide_context.get_token() is None:
return False
_LOGGER.debug('Current GeoRide token: %s', georide_context.get_token())
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)
coordoned_trackers = georide_context.get_coordoned_trackers()
entities = []
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)
hass.data[GEORIDE_DOMAIN]["devices"][tracker.tracker_id] = entity
tracker_entities.append(entity)
async_add_entities(tracker_entities)
async_add_entities(entities)
return True
class GeoRideTrackerEntity(TrackerEntity):
class GeoRideTrackerEntity(CoordinatorEntity, TrackerEntity):
"""Represent a tracked device."""
def __init__(self, tracker_id, get_token_callback, get_tracker_callback, tracker):
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device: Device, hass):
"""Set up GeoRide entity."""
self._tracker_id = tracker_id
self._get_token_callback = get_token_callback
self._get_tracker_callback = get_tracker_callback
self._name = tracker.tracker_name
self._data = tracker or {}
self.entity_id = DOMAIN + ".{}".format(tracker_id)
super().__init__(coordinator)
self._name = tracker_device.tracker.tracker_name
self._tracker_device = tracker_device
self.entity_id = DOMAIN + ".{}".format(tracker_device.tracker.tracker_id)
self._hass = hass
@property
def unique_id(self):
"""Return the unique ID."""
return self._tracker_id
return f"georide_tracker_{self._tracker_device.tracker.tracker_id}"
@property
def name(self):
return self._name
""" GeoRide odometer name """
return f"{self._name} position"
@property
def latitude(self):
"""Return latitude value of the device."""
if self._data.latitude:
return self._data.latitude
if self._tracker_device.tracker.latitude:
return self._tracker_device.tracker.latitude
return None
@property
def longitude(self):
"""Return longitude value of the device."""
if self._data.longitude:
return self._data.longitude
if self._tracker_device.tracker.longitude:
return self._tracker_device.tracker.longitude
return None
@property
def source_type(self):
"""Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS
@property
def location_accuracy(self):
""" return the gps accuracy of georide (could not be aquired, then 10) """
@ -90,37 +85,8 @@ class GeoRideTrackerEntity(TrackerEntity):
def icon(self):
"""return the entity icon"""
return "mdi:map-marker"
@property
def device_info(self):
"""Return the 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
return self._tracker_device.device_info

@ -4,9 +4,10 @@
"config_flow": true,
"documentation": "https://github.com/ptimatth/GeorideHA",
"requirements": [
"georideapilib>=0.4.4",
"georideapilib>=0.6.0",
"pyjwt>=1.7.1"
],
"dependencies": [],
"codeowners": ["ptimatth"]
"codeowners": ["ptimatth"],
"version": "0.6.0"
}

@ -1,14 +1,18 @@
""" odometter sensor for GeoRide object """
import logging
from typing import Any, Mapping
from homeassistant.core import callback
from homeassistant.components.switch import SwitchEntity
from homeassistant.components.switch import ENTITY_ID_FORMAT
import georideapilib.api as GeoRideApi
from homeassistant.components.sensor import SensorEntity
from homeassistant.components.sensor import ENTITY_ID_FORMAT
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device
_LOGGER = logging.getLogger(__name__)
@ -16,87 +20,62 @@ _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"]
if georide_context.get_token() is None:
return False
georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.get_coordoned_trackers()
trackers = GeoRideApi.get_trackers(georide_context.get_token())
entities = []
for coordoned_tracker in coordoned_trackers:
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
entity = GeoRideOdometerSensorEntity(coordinator, tracker_device, hass)
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
entities.append(entity)
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)
async_add_entities(entities)
return True
class GeoRideOdometerSensorEntity(SwitchEntity):
class GeoRideOdometerSensorEntity(CoordinatorEntity, SensorEntity):
"""Represent a tracked device."""
def __init__(self, tracker_id, get_token_callback, get_tracker_callback, data):
"""Set up Georide entity."""
self._tracker_id = tracker_id
self._data = data or {}
self._get_token_callback = get_token_callback
self._get_tracker_callback = get_tracker_callback
self._name = data.tracker_name
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 = "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
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._state = self._data.odometer
self._hass = hass
@property
def unique_id(self):
"""Return the unique ID."""
return self._tracker_id
@property
def name(self):
""" GeoRide odometer name """
return self._name
return f"odometer_{self._tracker_device.tracker.tracker_id}"
@property
def state(self):
return self._state
"""state property"""
return self._tracker_device.tracker.odometer
@property
def unit_of_measurement(self):
"""unit of mesurment property"""
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
def name(self):
""" GeoRide odometer name """
return f"{self._name} odometer"
@property
def icon(self):
"""icon getter"""
return "mdi:counter"
@property
def device_info(self):
"""Return the device info."""
return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker_id)},
"manufacturer": "GeoRide"
}
return self._tracker_device.device_info

@ -2,129 +2,106 @@
import logging
from homeassistant.core import callback
from typing import Any, Mapping
from homeassistant.components.switch import SwitchEntity
from homeassistant.components.switch import ENTITY_ID_FORMAT
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
import georideapilib.api as GeoRideApi
from .const import DOMAIN as GEORIDE_DOMAIN
from .device import Device
_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"]
if georide_context.get_token() is None:
return False
_LOGGER.info('Current georide token: %s', georide_context.get_token())
trackers = GeoRideApi.get_trackers(georide_context.get_token())
georide_context = hass.data[GEORIDE_DOMAIN]["context"]
coordoned_trackers = georide_context.get_coordoned_trackers()
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
for coordoned_tracker in coordoned_trackers:
tracker_device = coordoned_tracker['tracker_device']
coordinator = coordoned_tracker['coordinator']
entity = GeoRideLockSwitchEntity(coordinator, tracker_device, hass)
hass.data[GEORIDE_DOMAIN]["devices"][tracker_device.tracker.tracker_id] = coordinator
lock_switch_entities.append(entity)
async_add_entities(lock_switch_entities)
return True
class GeoRideLockSwitchEntity(SwitchEntity):
class GeoRideLockSwitchEntity(CoordinatorEntity, SwitchEntity):
"""Represent a tracked device."""
def __init__(self, tracker_id, get_token_callback, get_tracker_callback, data):
def __init__(self, coordinator: DataUpdateCoordinator[Mapping[str, Any]],
tracker_device:Device, hass):
"""Set up GeoRide entity."""
self._tracker_id = tracker_id
self._data = data or {}
self._get_token_callback = get_token_callback
self._get_tracker_callback = get_tracker_callback
self._name = data.tracker_name
self._is_on = data.is_locked
self.entity_id = ENTITY_ID_FORMAT.format("lock") +"." + str(tracker_id)
self._state = {}
super().__init__(coordinator)
self._tracker_device = tracker_device
self._name = tracker_device.tracker.tracker_name
self.entity_id = f"{ENTITY_ID_FORMAT.format('lock')}.{tracker_device.tracker.tracker_id}"# pylint: disable=C0301
self._hass = hass
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
""" lock the GeoRide tracker """
_LOGGER.info('async_turn_on %s', kwargs)
success = GeoRideApi.lock_tracker(self._get_token_callback(), self._tracker_id)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
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:
self._data.is_locked = True
self._is_on = True
def turn_off(self, **kwargs):
self._tracker.is_locked = True
async def async_turn_off(self, **kwargs):
""" unlock the GeoRide tracker """
_LOGGER.info('async_turn_off %s', kwargs)
success = GeoRideApi.unlock_tracker(self._get_token_callback(), self._tracker_id)
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
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:
self._data.is_locked = False
self._is_on = False
self._tracker.is_locked = False
async def async_toggle(self, **kwargs):
""" toggle lock the georide tracker """
_LOGGER.info('async_toggle %s', kwargs)
result = GeoRideApi.toogle_lock_tracker(self._get_token_callback(),
self._tracker_id)
self._data.is_locked = result
self._is_on = 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
georide_context = self._hass.data[GEORIDE_DOMAIN]["context"]
token = await georide_context.get_token()
result = await self._hass.async_add_executor_job(GeoRideApi.toogle_lock_tracker,
token, self._tracker_device.tracker.tracker_id)
self._tracker.is_locked = result
@property
def unique_id(self):
"""Return the unique ID."""
return self._tracker_id
return f"lock_{self._tracker_device.tracker.tracker_id}"
@property
def name(self):
""" GeoRide switch name """
return self._name
""" GeoRide odometer name """
return f"{self._name} lock"
@property
def is_on(self):
""" GeoRide switch status """
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
return self._tracker_device.tracker.is_locked
@property
def icon(self):
"""return the entity icon"""
if self._is_on:
if self._tracker_device.tracker.is_locked:
return "mdi:lock"
return "mdi:lock-open"
@property
def device_info(self):
"""Return the device info."""
return {
"name": self.name,
"identifiers": {(GEORIDE_DOMAIN, self._tracker_id)},
"manufacturer": "GeoRide"
}
return self._tracker_device.device_info

@ -4,5 +4,5 @@
"render_readme": true,
"domains": ["devices_tracker", "sensor"],
"country": ["FR"],
"homeassistant": "0.110.0"
"homeassistant": "2021.4.0"
}
Loading…
Cancel
Save