jcgoette / empower_me_homeassistant Goto Github PK
View Code? Open in Web Editor NEWThis custom integration provides sensors for ourclublogin.com data points.
License: MIT License
This custom integration provides sensors for ourclublogin.com data points.
License: MIT License
translations
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from .const import ATTR_TITLE, DOMAIN
# TODO: better validation of data
# TODO: translations
class OurClubLoginConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, user_input):
if user_input is not None:
return self.async_create_entry(title=ATTR_TITLE, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ID): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}
),
)
63a3084554a68943838932099a6dd5134b0b3b52
OOP this
"""Platform for ourclublogin.com sensor integration."""
import logging
import re
from datetime import datetime, timedelta, timezone
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
import requests
import voluptuous as vol
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.entity import Entity
from .const import ATTR_CHECK_IN_DATE, ATTR_DATETIME_FORMAT, ATTR_ICON, ATTR_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=15)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup ourclublogin.com sensors from a config entry created in the integrations UI."""
config = hass.data[DOMAIN][config_entry.entry_id]
club_id = config[CONF_ID]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
our_club_login_data = OurClubLoginData(club_id, username, password)
async_add_entities(
[OurClubLoginSensor(our_club_login_data)], update_before_add=True
)
class OurClubLoginSensor(Entity):
"""Representation of a ourclublogin.com Sensor."""
def __init__(self, our_club_login_data):
"""Initialize the ourclublogin.com sensor."""
self._state = None
self._our_club_login_data = our_club_login_data
self._data = self._our_club_login_data.data
@property
def name(self):
"""Return the name of the ourclublogin.com sensor."""
return f'Last {self._data.get("ClubName")} Check In'
@property
def state(self):
"""Return the state of the ourclublogin.com sensor."""
return self._data.get("CheckInDate")
@property
def icon(self):
"""Return the icon to use in ourclublogin.com frontend."""
return ATTR_ICON
def update(self):
"""Update data from ourclublogin.com for the sensor."""
self._our_club_login_data.update()
self._data = self._our_club_login_data.data
class OurClubLoginData:
"""Coordinate retrieving and updating data from ourclublogin.com."""
def __init__(self, club_id, username, password):
"""Initialize the OurClubLoginData object."""
self._club_id = club_id
self._username = username
self._password = password
self._session = requests.Session()
self.data = None
# TODO: OOP this
def our_club_login_query(self):
"""Query ourclublogin.com for data."""
params = (("ReturnUrl", f"^%^{self._club_id}"),)
response = self._session.get(f"{ATTR_URL}/Account/Login", params=params)
if response.ok:
_LOGGER.info("Connected to ourclublogin.com")
else:
_LOGGER.critical("Could not connect to ourclublogin.com")
csrf_hidden_token = re.search(
'"\_\_RequestVerificationToken".+?value\="(.+?)"', str(response.content)
)
if csrf_hidden_token:
csrf_hidden_token = csrf_hidden_token[1]
_LOGGER.info("Hidden CSRF token acquired")
else:
_LOGGER.critical("Could not acquire hidden CSRF token")
data = {
"__RequestVerificationToken": csrf_hidden_token,
"Username": self._username,
"Password": self._password,
}
response = self._session.post(f"{ATTR_URL}/Account/Login", data=data)
if response.ok:
_LOGGER.info("Authenticated with ourclublogin.com")
else:
_LOGGER.critical("Could not authenticate with ourclublogin.com")
datetime_local_now = datetime.now().astimezone()
datetime_local_new_year = datetime(datetime_local_now.year, 1, 1)
datetime_utc_now = datetime_local_now.astimezone(timezone.utc)
datetime_utc_new_year = datetime_local_new_year.astimezone(timezone.utc)
data = {
"startDate": datetime_utc_new_year.strftime(ATTR_DATETIME_FORMAT),
"endDate": datetime_utc_now.strftime(ATTR_DATETIME_FORMAT),
}
response = self._session.post(
f"{ATTR_URL}/Checkin/GetCustomerVisits", data=data
)
if response.ok:
_LOGGER.info("Posted to GetCustomerVisits endpoint")
else:
_LOGGER.critical("Could not post to GetCustomerVisits endpoint")
customer_visits = response.json()
customer_visits.sort(key=lambda item: item[ATTR_CHECK_IN_DATE]) # ensure sorted
last_customer_visit = customer_visits[-1:][0]
# TODO: add other data points
self.data = last_customer_visit
def update(self):
"""Update data from ourclublogin.com via our_club_login_query."""
return self.our_club_login_query()
de93d90701a7e1d69af01ad3b518d15f82bd4d67
add other data points
"""Platform for ourclublogin.com sensor integration."""
import logging
import re
from datetime import datetime, timedelta, timezone
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
import requests
import voluptuous as vol
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.entity import Entity
from .const import ATTR_CHECK_IN_DATE, ATTR_DATETIME_FORMAT, ATTR_ICON, ATTR_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=15)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Setup ourclublogin.com sensors from a config entry created in the integrations UI."""
config = hass.data[DOMAIN][config_entry.entry_id]
club_id = config[CONF_ID]
username = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
our_club_login_data = OurClubLoginData(club_id, username, password)
async_add_entities(
[OurClubLoginSensor(our_club_login_data)], update_before_add=True
)
class OurClubLoginSensor(Entity):
"""Representation of a ourclublogin.com Sensor."""
def __init__(self, our_club_login_data):
"""Initialize the ourclublogin.com sensor."""
self._state = None
self._our_club_login_data = our_club_login_data
self._data = self._our_club_login_data.data
@property
def name(self):
"""Return the name of the ourclublogin.com sensor."""
return f'Last {self._data.get("ClubName")} Check In'
@property
def state(self):
"""Return the state of the ourclublogin.com sensor."""
return self._data.get("CheckInDate")
@property
def icon(self):
"""Return the icon to use in ourclublogin.com frontend."""
return ATTR_ICON
def update(self):
"""Update data from ourclublogin.com for the sensor."""
self._our_club_login_data.update()
self._data = self._our_club_login_data.data
class OurClubLoginData:
"""Coordinate retrieving and updating data from ourclublogin.com."""
def __init__(self, club_id, username, password):
"""Initialize the OurClubLoginData object."""
self._club_id = club_id
self._username = username
self._password = password
self._session = requests.Session()
self.data = None
# TODO: OOP this
def our_club_login_query(self):
"""Query ourclublogin.com for data."""
params = (("ReturnUrl", f"^%^{self._club_id}"),)
response = self._session.get(f"{ATTR_URL}/Account/Login", params=params)
if response.ok:
_LOGGER.info("Connected to ourclublogin.com")
else:
_LOGGER.critical("Could not connect to ourclublogin.com")
csrf_hidden_token = re.search(
'"\_\_RequestVerificationToken".+?value\="(.+?)"', str(response.content)
)
if csrf_hidden_token:
csrf_hidden_token = csrf_hidden_token[1]
_LOGGER.info("Hidden CSRF token acquired")
else:
_LOGGER.critical("Could not acquire hidden CSRF token")
data = {
"__RequestVerificationToken": csrf_hidden_token,
"Username": self._username,
"Password": self._password,
}
response = self._session.post(f"{ATTR_URL}/Account/Login", data=data)
if response.ok:
_LOGGER.info("Authenticated with ourclublogin.com")
else:
_LOGGER.critical("Could not authenticate with ourclublogin.com")
datetime_local_now = datetime.now().astimezone()
datetime_local_new_year = datetime(datetime_local_now.year, 1, 1)
datetime_utc_now = datetime_local_now.astimezone(timezone.utc)
datetime_utc_new_year = datetime_local_new_year.astimezone(timezone.utc)
data = {
"startDate": datetime_utc_new_year.strftime(ATTR_DATETIME_FORMAT),
"endDate": datetime_utc_now.strftime(ATTR_DATETIME_FORMAT),
}
response = self._session.post(
f"{ATTR_URL}/Checkin/GetCustomerVisits", data=data
)
if response.ok:
_LOGGER.info("Posted to GetCustomerVisits endpoint")
else:
_LOGGER.critical("Could not post to GetCustomerVisits endpoint")
customer_visits = response.json()
customer_visits.sort(key=lambda item: item[ATTR_CHECK_IN_DATE]) # ensure sorted
last_customer_visit = customer_visits[-1:][0]
# TODO: add other data points
self.data = last_customer_visit
def update(self):
"""Update data from ourclublogin.com via our_club_login_query."""
return self.our_club_login_query()
56c73b0aeff96d38eaaf3faf61f044a9196bdf71
better validation of data
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
from .const import ATTR_TITLE, DOMAIN
# TODO: better validation of data
# TODO: translations
class OurClubLoginConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, user_input):
if user_input is not None:
return self.async_create_entry(title=ATTR_TITLE, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ID): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}
),
)
bd9ca5743b07870500bdf6b2d8d0911e2021eb00
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.