Skip to content

LOGLEVEL and SLACK_CUSTOM_AVAILABLE_STATUS fixes, minor linting. #85

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PortableProgrammer merged 2 commits into from
Jun 30, 2023
Merged

LOGLEVEL and SLACK_CUSTOM_AVAILABLE_STATUS fixes, minor linting. #85

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
78f53f3
Handle `LOGLEVEL` correctly globally.
PortableProgrammer Jun 30, 2023
e45a521
Slack: Fix `SLACK_CUSTOM_AVAILABLE_STATUS` default.
PortableProgrammer Jun 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 18 additions & 8 deletions status-light/sources/collaboration/slack.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ def get_user_presence(self) -> enum.Status:
'Slack Exception while getting user presence: %s', ex.response['error'])
logger.exception(ex)
return_value = enum.Status.UNKNOWN
except Exception as ex: # pylint: disable=broad-except
except Exception as ex: # pylint: disable=broad-except
logger.warning(
'Exception while getting Slack user presence: %s', ex)
logger.exception(ex)
Expand All @@ -80,7 +80,7 @@ def _get_user_info(self, client: WebClient) -> dict | None:
'Slack Exception while getting user info: %s', ex.response['error'])
logger.exception(ex)
return None
except Exception as ex: # pylint: disable=broad-except
except Exception as ex: # pylint: disable=broad-except
logger.warning('Exception while getting Slack user info: %s', ex)
logger.exception(ex)
return None
Expand All @@ -103,25 +103,35 @@ def _parse_custom_status(self, client: WebClient,

# For each of the Slack custom statuses, check them in reverse precedence order
# Off, Available, Scheduled, Busy
if self.custom_off_status and custom_status.startswith(tuple(self.custom_off_status)):
if len(self.custom_off_status) > 0 and \
custom_status.startswith(tuple(self.custom_off_status)):
logger.debug(
'Custom status matched custom_off_status: %s', custom_status)
return_value = self.custom_off_status_map

if self.custom_available_status and \
if len(self.custom_available_status) > 0 and \
custom_status.startswith(tuple(self.custom_available_status)):
logger.debug(
'Custom status matched custom_available_status: %s', custom_status)
return_value = self.custom_available_status_map

if self.custom_scheduled_status and \
if len(self.custom_scheduled_status) > 0 and \
custom_status.startswith(tuple(self.custom_scheduled_status)):
logger.debug(
'Custom status matched custom_scheduled_status: %s', custom_status)
return_value = self.custom_scheduled_status_map

if self.custom_busy_status and \
if len(self.custom_busy_status) > 0 and \
custom_status.startswith(tuple(self.custom_busy_status)):
logger.debug(
'Custom status matched custom_busy_status: %s', custom_status)
return_value = self.custom_busy_status_map

# Check for Huddle and Call
if user_info['profile']['huddle_state'] == 'in_a_huddle' or \
user_info['profile']['status_emoji'] == ':slack_call:':

logger.debug('Custom status indicates Huddle (%s) or Call (%s)',
user_info['profile']['huddle_state'], custom_status)
return_value = enum.Status.CALL

except (SystemExit, KeyboardInterrupt):
Expand All @@ -131,7 +141,7 @@ def _parse_custom_status(self, client: WebClient,
'Slack Exception while parsing custom status: %s', ex.response['error'])
logger.exception(ex)
return_value = enum.Status.UNKNOWN
except Exception as ex: # pylint: disable=broad-except
except Exception as ex: # pylint: disable=broad-except
logger.warning(
'Exception while parsing Slack custom status: %s', ex)
logger.exception(ex)
Expand Down
76 changes: 41 additions & 35 deletions status-light/status-light.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,9 @@

class StatusLight:
"""Provides a structured entry point for the Status-Light application"""
# Instance Logger
logger: logging.Logger = logging.getLogger('status-light')

# Instance Properties
local_env = env.Environment()
current_status: enum.Status = enum.Status.UNKNOWN
Expand Down Expand Up @@ -62,27 +65,30 @@ def init(self):
self.local_env.get_log_level()]:

# We failed to gather some environment variables
logger.error('Failed to find all environment variables!')
self.logger.error('Failed to find all environment variables!')
sys.exit(1)

# 23 - Make logging level configurable
logger.info('Setting log level to %s', self.local_env.log_level.name)
logger.setLevel(self.local_env.log_level.value)
self.logger.info('Setting log level to %s', self.local_env.log_level.name)
# Reset the root logger config to our epxected logging level
logging.basicConfig(format='%(asctime)s %(name)s.%(funcName)s %(levelname)s: %(message)s',
datefmt='[%Y-%m-%d %H:%M:%S]', level=self.local_env.log_level.value, force=True)
self.logger.setLevel(self.local_env.log_level.value)

# Depending on the selected sources, get the environment
if enum.StatusSource.WEBEX in self.local_env.selected_sources:
if self.local_env.get_webex():
logger.info('Requested Webex')
self.logger.info('Requested Webex')
self.webex_api.bot_id = self.local_env.webex_bot_id
self.webex_api.person_id = self.local_env.webex_person_id
else:
logger.error(
self.logger.error(
'Requested Webex, but could not find all environment variables!')
sys.exit(1)

if enum.StatusSource.SLACK in self.local_env.selected_sources:
if self.local_env.get_slack():
logger.info('Requested Slack')
self.logger.info('Requested Slack')
self.slack_api.user_id = self.local_env.slack_user_id
self.slack_api.bot_token = self.local_env.slack_bot_token
# 66 - Support Slack custom statuses
Expand All @@ -97,44 +103,44 @@ def init(self):
self.slack_api.custom_scheduled_status_map = self.local_env.scheduled_status[
0]
else:
logger.error(
self.logger.error(
'Requested Slack, but could not find all environment variables!')
sys.exit(1)

if enum.StatusSource.OFFICE365 in self.local_env.selected_sources:
if self.local_env.get_office():
logger.info('Requested Office 365')
self.logger.info('Requested Office 365')
self.office_api.appID = self.local_env.office_app_id
self.office_api.appSecret = self.local_env.office_app_secret
self.office_api.tokenStore = self.local_env.office_token_store
# 81 - Make calendar lookahead configurable
self.office_api.lookahead = self.local_env.calendar_lookahead
self.office_api.authenticate()
else:
logger.error(
self.logger.error(
'Requested Office 365, but could not find all environment variables!')
sys.exit(1)

# 47 - Add Google support
if enum.StatusSource.GOOGLE in self.local_env.selected_sources:
if self.local_env.get_google():
logger.info('Requested Google')
self.logger.info('Requested Google')
self.google_api.credentialStore = self.local_env.google_credential_store
self.google_api.tokenStore = self.local_env.google_token_store
# 81 - Make calendar lookahead configurable
self.google_api.lookahead = self.local_env.calendar_lookahead
else:
logger.error(
self.logger.error(
'Requested Google, but could not find all environment variables!')
sys.exit(1)

# Tuya
self.light.device = self.local_env.tuya_device
logger.debug('Retrieved TUYA_DEVICE variable: %s', self.light.device)
self.logger.debug('Retrieved TUYA_DEVICE variable: %s', self.light.device)
tuya_status = self.light.get_status()
logger.debug('Found initial Tuya status: %s', tuya_status)
self.logger.debug('Found initial Tuya status: %s', tuya_status)
if not tuya_status:
logger.error(
self.logger.error(
'Could not connect to Tuya device!')
sys.exit(1)

Expand All @@ -151,7 +157,7 @@ def run(self):
self.local_env.active_hours_start,
self.local_env.active_hours_end):

logger.debug('Within Active Hours, polling')
self.logger.debug('Within Active Hours, polling')

# Reset the "outside of active hours" handler
already_handled_inactive_hours = False
Expand Down Expand Up @@ -193,7 +199,7 @@ def run(self):
google_status.name.lower())

# 74: Log enums as names, not values
logger.debug(logger_string.lstrip().rstrip(' |'))
self.logger.debug(logger_string.lstrip().rstrip(' |'))

# TODO: Now that we have more than one calendar-based status source,
# build a real precedence module for these
Expand All @@ -204,7 +210,7 @@ def run(self):
if (webex_status == enum.Status.UNKNOWN or
webex_status in self.local_env.off_status):
# 74: Log enums as names, not values
logger.debug('Using slack_status: %s',
self.logger.debug('Using slack_status: %s',
slack_status.name.lower())
# Fall through to Slack
self.current_status = slack_status
Expand All @@ -214,15 +220,15 @@ def run(self):
and (office_status not in self.local_env.off_status
or google_status not in self.local_env.off_status):

logger.debug('Using calendar-based status')
self.logger.debug('Using calendar-based status')
# Office should take precedence over Google for now
# 74: Log enums as names, not values
if office_status != enum.Status.UNKNOWN:
logger.debug('Using office_status: %s',
self.logger.debug('Using office_status: %s',
office_status.name.lower())
self.current_status = office_status
else:
logger.debug('Using google_status: %s',
self.logger.debug('Using google_status: %s',
google_status.name.lower())
self.current_status = google_status

Expand All @@ -232,26 +238,26 @@ def run(self):
status_changed = True

if status_changed:
logger.info('Found new status: %s',
self.logger.info('Found new status: %s',
self.current_status.name.lower())

if not last_transition_result:
logger.warning(
self.logger.warning(
'Last attempt to set status failed. Retrying.')

# If status changed this loop
# 40: or the last transition failed,
if status_changed or not last_transition_result:
# 74: Log enums as names, not values
logger.info('Transitioning to %s',
self.logger.info('Transitioning to %s',
self.current_status.name.lower())
last_transition_result = self._transition_status()

else:
logger.debug('Outside Active Hours, pausing')
self.logger.debug('Outside Active Hours, pausing')

if not already_handled_inactive_hours:
logger.info(
self.logger.info(
'Outside of active hours, transitioning to off')
last_transition_result = self.light.off()
self.last_status = enum.Status.UNKNOWN
Expand All @@ -264,14 +270,14 @@ def run(self):
# Sleep for a few seconds
time.sleep(self.local_env.sleep_seconds)
except (SystemExit, KeyboardInterrupt) as ex:
logger.info('%s received; shutting down...',
self.logger.info('%s received; shutting down...',
ex.__class__.__name__)
self.should_continue = False
except Exception as ex: # pylint: disable=broad-except
logger.warning('Exception during main loop: %s', ex)
logger.exception(ex)
self.logger.warning('Exception during main loop: %s', ex)
self.logger.exception(ex)

logger.debug('Turning light off')
self.logger.debug('Turning light off')
self.light.off()

def _transition_status(self) -> bool:
Expand Down Expand Up @@ -302,7 +308,7 @@ def _transition_status(self) -> bool:
# just turn the light off and warn about it
# 74: Log enums as names, not values
else:
logger.warning('Called with an invalid status: %s',
self.logger.warning('Called with an invalid status: %s',
self.current_status.name.lower())
return_value = self.light.off()

Expand All @@ -316,7 +322,7 @@ def _transition_status(self) -> bool:
# Default to INFO level until we load the environment
logging.basicConfig(format='%(asctime)s %(name)s.%(funcName)s %(levelname)s: %(message)s',
datefmt='[%Y-%m-%d %H:%M:%S]', level=logging.INFO)
logger: logging.Logger = logging.getLogger('status-light')
global_logger: logging.Logger = logging.getLogger('status-light')


def receive_signal(signal_number, frame): # pylint: disable=unused-argument
Expand All @@ -331,9 +337,9 @@ def receive_signal(signal_number, frame): # pylint: disable=unused-argument
try:
signal_name = signal.Signals(signal_number).name
except ValueError as value_ex:
logger.warning(
global_logger.warning(
'Exception encountered converting %s to signal.Signals: %s', signal_number, value_ex)
logger.warning('Signal received: %s', signal_name)
global_logger.warning('Signal received: %s', signal_name)
status_light.should_continue = False


Expand All @@ -342,10 +348,10 @@ def receive_signal(signal_number, frame): # pylint: disable=unused-argument

def main():
"""Provides the entry point for the application"""
logger.info('Startup')
global_logger.info('Startup')
status_light.init()
status_light.run()
logger.info('Shutdown')
global_logger.info('Shutdown')


if __name__ == '__main__':
Expand Down
2 changes: 1 addition & 1 deletion status-light/utility/env.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ class Environment:
# 66 - Add Slack custom status support
slack_off_status: list[str] = [
':no_entry: Out of Office', ':airplane:', ':palm_tree: Vacationing']
slack_available_status: list[str] = ['']
slack_available_status: list[str] = []
slack_scheduled_status: list[str] = [':spiral_calendar_pad: In a meeting']
slack_busy_status: list[str] = [
':no_entry_sign:', ':no_entry: Do not Disturb']
Expand Down
Loading