diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index b217e3c..470857e 100755 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -14,19 +14,19 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v3 + uses: actions/checkout@v4 - name: Prepare id: prepare run: | DOCKER_IMAGE=portableprogrammer/status-light - DOCKER_PLATFORMS=linux/amd64,linux/arm/v6,linux/arm/v7,linux/arm64 + DOCKER_PLATFORMS=linux/amd64,linux/arm/v7,linux/arm64 VERSION=edge if [[ $GITHUB_REF == refs/tags/* ]]; then VERSION=${GITHUB_REF#refs/tags/v} fi - + TAGS="--tag ${DOCKER_IMAGE}:${VERSION}" if [[ $VERSION =~ ^[0-9]{1,3}\.[0-9]{1,3}(\.[0-9]{1,3})?$ ]]; then TAGS="$TAGS --tag ${DOCKER_IMAGE}:latest" @@ -37,10 +37,13 @@ jobs: echo "buildx_args=--platform ${DOCKER_PLATFORMS} --build-arg VERSION=${VERSION} --build-arg BUILD_DATE=$(date -u +'%Y-%m-%dT%H:%M:%SZ') --build-arg VCS_REF=${GITHUB_SHA::8} ${TAGS} --file ./Dockerfiles/Dockerfile ./" >> "$GITHUB_ENV" - name: Set up QEMU - uses: docker/setup-qemu-action@v2 + uses: docker/setup-qemu-action@v3 - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 + uses: docker/setup-buildx-action@v3 + with: + # Enable Docker layer caching + buildkitd-flags: --debug - name: Docker Buildx (build) run: | diff --git a/.gitignore b/.gitignore index e5cc0d9..9f6b4c6 100755 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ __pycache__ *.pyc .vscode/ .devcontainer/ +.claude/ build/ dist/ status-light/__pycache__ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..4471672 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,105 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Status-Light monitors user status across collaboration platforms (Webex, Slack) and calendars (Office 365, Google Calendar) to display the "busiest" status on a Tuya RGB LED smart bulb. The application runs as a continuous polling loop inside a Docker container. + +## Running the Application + +**Local execution:** +```bash +python -u status-light/status-light.py +``` + +**Docker (preferred):** +```bash +docker run -e SOURCES=webex,office365 -e WEBEX_PERSONID=... portableprogrammer/status-light +``` + +**Build Docker image locally:** +```bash +docker build -f Dockerfiles/Dockerfile -t status-light . +``` + +There is no formal test suite, linter configuration, or build system. Development testing is done manually through Docker. + +## Architecture + +### Core Application Flow + +`status-light/status-light.py` is the entry point containing the `StatusLight` class which: +1. Validates environment variables and initializes configured sources +2. Runs a continuous polling loop (configurable via `SLEEP_SECONDS`) +3. Queries each enabled source for current status +4. Applies status precedence rules to determine the "busiest" status +5. Updates the Tuya smart light color accordingly +6. Handles graceful shutdown via signal handlers (SIGHUP, SIGINT, SIGQUIT, SIGTERM) + +### Status Sources + +**Collaboration sources** (`sources/collaboration/`): +- `webex.py` - Webex Teams presence API +- `slack.py` - Slack presence with custom status emoji/text parsing + +**Calendar sources** (`sources/calendar/`): +- `office365.py` - Microsoft Graph API free/busy +- `google.py` - Google Calendar API free/busy +- `ics.py` - ICS/iCalendar file support (RFC 5545 compliant, uses `icalendar` + `recurring-ical-events` for proper timezone and recurring event handling) + +### Status Precedence + +Status determination follows a strict hierarchy: +1. **Source priority:** Collaboration always wins over Calendar (except UNKNOWN/OFF) +2. **Within collaboration:** Webex > Slack +3. **Within calendar:** Office 365 > Google > ICS +4. **Status priority:** Busy > Tentative/Scheduled > Available > Off + +### Key Utilities + +- `utility/enum.py` - Status enumerations (CollaborationStatus, CalendarStatus, Color) +- `utility/env.py` - Environment variable parsing and validation +- `utility/util.py` - Helper functions including `get_env_or_secret()` for Docker secrets +- `utility/precedence.py` - Status precedence selection logic (selects winning status from multiple collaboration and calendar sources) + +### Output Targets + +All targets implement the `LightTarget` abstract interface defined in `targets/base.py`: + +- `targets/base.py` - Abstract base class defining the light target interface (`set_color()`, `turn_off()`) +- `targets/tuya.py` - Tuya smart bulbs (locked to COLOR mode, handles retry logic and DPS format conversion) +- `targets/virtual.py` - Virtual light for testing (logs status changes, no hardware required) + +## Configuration + +All configuration is via environment variables (no config files). Key categories: + +- **Sources:** `SOURCES` (comma-separated: webex, slack, office365, google, ics) +- **Target:** `TARGET` (tuya or virtual; default: tuya) +- **Authentication:** Platform-specific tokens/IDs (e.g., `WEBEX_PERSONID`, `SLACK_BOT_TOKEN`, `O365_APPID`, `ICS_URL`) +- **Colors:** `AVAILABLE_COLOR`, `SCHEDULED_COLOR`, `BUSY_COLOR` (predefined names or 24-bit hex) +- **Light:** `LIGHT_BRIGHTNESS` (0-100 percentage; default: 50; auto-detects legacy 0-255 format) +- **Device:** `TUYA_DEVICE` (JSON with protocol, deviceid, ip, localkey; required only when TARGET=tuya) +- **Behavior:** `SLEEP_SECONDS`, `CALENDAR_LOOKAHEAD`, `LOGLEVEL`, `ACTIVE_DAYS`, `ACTIVE_HOURS_*` + +Secrets support `*_FILE` variants for Docker secrets integration. + +## CI/CD + +GitHub Actions (`.github/workflows/docker-image.yml`) builds multi-platform Docker images: +- Platforms: linux/amd64, linux/arm/v6, linux/arm/v7, linux/arm64 +- Triggers: PRs to main (build only), pushes with `v*` tags (build + push to Docker Hub) +- Published to: `portableprogrammer/status-light` + +## Documentation Files + +Keep these files in sync with code changes: +- `README.md` - User-facing documentation for environment variables, setup, and usage +- `CLAUDE.md` - Developer guidance for Claude Code (this file) + +Update when: +- New or modified environment variables +- New dependencies in `requirements.txt` +- Architecture changes (new sources, targets, utilities) +- Configuration options diff --git a/Dockerfiles/Dockerfile b/Dockerfiles/Dockerfile index 336442f..bbdbcd2 100755 --- a/Dockerfiles/Dockerfile +++ b/Dockerfiles/Dockerfile @@ -7,10 +7,10 @@ WORKDIR /usr/src/status-light # Install dependencies COPY ./requirements.txt ./ RUN apt-get update \ - && apt-get install -y gcc \ + && apt-get install -y gcc libffi-dev \ && rm -rf /var/lib/apt/lists/* \ && pip install -r requirements.txt \ - && apt-get purge -y --auto-remove gcc + && apt-get purge -y --auto-remove gcc libffi-dev # Copy the project COPY ./status-light . diff --git a/README.md b/README.md index d666918..cb8610a 100755 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ services: image: portableprogrammer/status-light:latest environment: - "SOURCES=Webex,Office365" + - "TARGET=tuya" - "AVAILABLE_COLOR=green" - "SCHEDULED_COLOR=orange" - "BUSY_COLOR=red" @@ -48,7 +49,7 @@ services: - "BUSY_STATUS=call,donotdisturb,meeting,presenting,pending" - "OFF_STATUS=inactive,outofoffice,free,unknown" - 'TUYA_DEVICE={ "protocol": "3.3", "deviceid": "xxx", "ip": "yyy", "localkey": "zzz" }' - - "TUYA_BRIGHTNESS=128" + - "LIGHT_BRIGHTNESS=50" - "WEBEX_PERSONID=xxx" - "WEBEX_BOTID=xxx" - "O365_APPID=xxx" @@ -56,6 +57,9 @@ services: - "O365_TOKENSTORE=/data" - "GOOGLE_TOKENSTORE=/data" - "GOOGLE_CREDENTIALSTORE=/data" + - "ICS_URL=https://example.com/calendar.ics" + - "ICS_CACHESTORE=/data" + - "ICS_CACHELIFETIME=30" - "SLACK_USER_ID=xxx" - "SLACK_BOT_TOKEN=xxx" - "SLACK_CUSTOM_AVAILABLE_STATUS=''" @@ -125,12 +129,25 @@ secrets: - `slack` - `office365` - `google` + - `ics` - Default value: `webex,office365` If specificed, requires at least one of the available options. This will control which services Status-Light uses to determine overall availability status. --- +### `TARGET` + +- *Optional* +- Available values: + - `tuya` - Physical Tuya smart bulb (requires `TUYA_DEVICE`) + - `virtual` - Virtual light that logs status changes (for testing) +- Default value: `tuya` + +Specifies the output target for status display. Use `virtual` for testing without hardware. + +--- + ### **Statuses** - *Optional* @@ -155,6 +172,10 @@ If specificed, requires at least one of the available options. This will control - Google - `free` - `busy` + - ICS + - `free` + - `tentative` + - `busy` #### `AVAILABLE_STATUS` @@ -204,14 +225,17 @@ if webexStatus == const.Status.unknown or webexStatus in offStatus: # Fall through to Slack currentStatus = slackStatus -if (currentStatus in availableStatus or currentStatus in offStatus) - and (officeStatus not in offStatus or googleStatus not in offStatus): +if (currentStatus in availableStatus or currentStatus in offStatus) + and (officeStatus not in offStatus or googleStatus not in offStatus + or icsStatus not in offStatus): - # Office 365 currently takes precedence over Google + # Office 365 currently takes precedence over Google, Google over ICS if (officeStatus != const.Status.unknown): currentStatus = officeStatus - else: + elif (googleStatus != const.Status.unknown): currentStatus = googleStatus + else: + currentStatus = icsStatus if currentStatus in availableStatus: # Get availableColor @@ -304,11 +328,13 @@ In the example above, the Slack custom status would match (since it is a case-in ### **Tuya** +**Note:** Tuya configuration is only required when [`TARGET`](#target) is set to `tuya` (the default). If using `TARGET=virtual`, you can skip this section. + #### `TUYA_DEVICE` -- *Required* +- *Required if [`TARGET`](#target) is `tuya`* -Status-Light requires a [Tuya](https://www.tuya.com/) device, which are white-boxed and sold under many brand names. For example, the Tuya light working in the current environment is an [Above Lights](http://alabovelights.com/) [Smart Bulb 9W, model AL1](http://alabovelights.com/pd.jsp?id=17). +Status-Light supports [Tuya](https://www.tuya.com/) devices, which are white-boxed and sold under many brand names. For example, the Tuya light working in the current environment is an [Above Lights](http://alabovelights.com/) [Smart Bulb 9W, model AL1](http://alabovelights.com/pd.jsp?id=17). Status-Light uses the [tuyaface](https://github.com/TradeFace/tuyaface/) module for Tuya communication. @@ -326,13 +352,19 @@ Example `TUYA_DEVICE` value: **Note:** Status-Light will accept an FQDN instead of IP, as long as the name can be resolved. Tuya devices will typically register themselves with the last 6 digits of the device ID, for example `ESP_xxxxxx.local`. -#### `TUYA_BRIGHTNESS` +#### `LIGHT_BRIGHTNESS` - *Optional* -- Acceptable range: `32`-`255` -- Default value: `128` +- Acceptable range: `0`-`100` (percentage) +- Default value: `50` + +Set the brightness of your RGB light as a percentage (0-100%). Status-Light defaults to 50% brightness. -Set the brightness of your Tuya light. This is an 8-bit `integer` corresponding to a percentage from 0%-100% (though Tuya lights typically don't accept a brightness value below `32`). Status-Light defaults to 50% brightness, `128`. +**Legacy Format Auto-Detection:** For backward compatibility, values above 100 (or in the range 32-100) are automatically detected as the legacy 0-255 format and converted to percentage. For example, `LIGHT_BRIGHTNESS=128` is auto-detected and converted to 50%. To avoid confusion, use percentage values (0-100) for new configurations. + +**Note:** The legacy format was specific to Tuya's device scale. The new percentage format works with all light targets and is more intuitive. + +**Backward Compatibility:** `TUYA_BRIGHTNESS` is still supported as an alias for `LIGHT_BRIGHTNESS` but is deprecated. Use `LIGHT_BRIGHTNESS` for new configurations. --- @@ -452,6 +484,70 @@ Since Google has [deprecated](https://developers.googleblog.com/2022/02/making-o --- +### **ICS** + +**Note:** See [`CALENDAR_LOOKAHEAD`](#calendar_lookahead) to configure lookahead timing for Calendar sources. + +Status-Light uses the [icalendar](https://github.com/collective/icalendar) and [recurring-ical-events](https://github.com/niccokunzmann/python-recurring-ical-events) libraries to parse ICS files. These libraries correctly handle recurring events and cross-timezone event matching (e.g., a Pacific time event will be correctly detected when running in Mountain time). + +Status-Light's ICS source implements **RFC 5545 compliant** status detection based on the `TRANSP` (transparency) and `STATUS` properties, **plus Microsoft CDO extensions** for enhanced Office 365/Outlook compatibility: + +**Standard RFC 5545 Properties:** + +| Event Properties | Status-Light Status | +|-----------------|---------------------| +| `TRANSP=TRANSPARENT` | `free` (event doesn't block time) | +| `STATUS=CANCELLED` | `free` (event was cancelled) | +| `STATUS=TENTATIVE` | `tentative` (maps to BUSY-TENTATIVE in RFC terms) | +| `STATUS=CONFIRMED` or unset | `busy` (default blocking event) | + +**Microsoft CDO Extensions** (Office 365/Outlook ICS exports): + +When present, the `X-MICROSOFT-CDO-BUSYSTATUS` property takes precedence and provides more granular status information: + +| CDO BUSYSTATUS Property | Status-Light Status | +|------------------------|---------------------| +| `FREE` or `0` | `free` | +| `TENTATIVE` or `1` | `tentative` | +| `BUSY` or `2` | `busy` | +| `OOF` or `3` | `outofoffice` (away/out of office) | +| `WORKINGELSEWHERE` or `4` | `workingelsewhere` (remote work) | + +**Status Precedence:** When multiple events exist in the lookahead window, the "busiest" status wins: `busy` > `outofoffice` > `workingelsewhere` > `tentative` > `free`. + +#### `ICS_URL` + +- *Required if `ics` is present in [`SOURCES`](#sources)* + +The URL to an ICS (iCalendar) file. This can be any publicly accessible URL that returns a valid `.ics` file, such as: +- A shared Google Calendar ICS link +- An Office 365 published calendar +- Any other iCalendar-compatible calendar export + +Status-Light will fetch this file periodically (controlled by `ICS_CACHELIFETIME`) and check for events within the [`CALENDAR_LOOKAHEAD`](#calendar_lookahead) window. + +**Docker Secrets:** This variable can instead be specified in a secrets file, using the `ICS_URL_FILE` variable. + +#### `ICS_CACHESTORE` + +- *Optional, only valid if `ics` is present in [`SOURCES`](#sources)* +- Acceptable value: Any writable location on disk, e.g. `/path/to/cache/` +- Default value: `~` + +Defines a writable location on disk where the cached ICS file is stored. + +**Note:** This path is directory only. Status-Light will persist a file named `status-light-ics-cache.ics` within the directory supplied. + +#### `ICS_CACHELIFETIME` + +- *Optional, only valid if `ics` is present in [`SOURCES`](#sources)* +- Acceptable range: `5`-`60` +- Default value: `30` + +Set the number of minutes the cached ICS file remains valid before being re-fetched from the URL. A lower value means more frequent updates but more network requests. + +--- + ### **Active Times** If you prefer to leave Status-Light running all the time (e.g. headless in a Docker container), you may wish to disable status polling during off hours. diff --git a/requirements.txt b/requirements.txt index 4e7e9d3..3a71b0b 100755 --- a/requirements.txt +++ b/requirements.txt @@ -5,4 +5,6 @@ O365 google-api-python-client google-auth-httplib2 google-auth-oauthlib -slack-sdk \ No newline at end of file +slack-sdk +icalendar +recurring-ical-events \ No newline at end of file diff --git a/status-light/sources/calendar/google.py b/status-light/sources/calendar/google.py index f606f40..e2f5c17 100755 --- a/status-light/sources/calendar/google.py +++ b/status-light/sources/calendar/google.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Google Calendar Source diff --git a/status-light/sources/calendar/ics.py b/status-light/sources/calendar/ics.py new file mode 100644 index 0000000..677ddb6 --- /dev/null +++ b/status-light/sources/calendar/ics.py @@ -0,0 +1,301 @@ +"""Status-Light +(c) 2020-2026 Nick Warner +https://github.com/portableprogrammer/Status-Light/ + +ICS Calendar Source +""" + +# Standard imports +import os +from datetime import datetime, timedelta, timezone +import logging +import urllib.request + +# 3rd-party imports +import icalendar +import recurring_ical_events + +# Project imports +from utility import enum + +logger = logging.getLogger(__name__) + + +class Ics: + """Handles ICS Calendar""" + + url: str = '' + + cacheStore: str = '~' + cacheFile: str = 'status-light-ics-cache.ics' + cacheLifetime: int = 30 + + # 81 - Make calendar lookahead configurable + lookahead: int = 5 + + # Cached calendar object + _calendar: icalendar.Calendar | None = None + + def _get_cache_path(self) -> str: + """Returns the full path to the cache file.""" + return os.path.normpath(os.path.join( + os.path.expanduser(self.cacheStore), + self.cacheFile + )) + + def _should_refresh_cache(self) -> bool: + """Checks if the cache file is stale or missing. + + Returns True if the cache should be refreshed.""" + cache_path = self._get_cache_path() + + # If the file doesn't exist, we need to fetch + if not os.path.exists(cache_path): + logger.debug('Cache file does not exist: %s', cache_path) + return True + + # Check if the file is older than the cache lifetime + try: + file_mtime = datetime.fromtimestamp( + os.stat(cache_path).st_mtime, tz=timezone.utc) + cache_expiry = datetime.now(timezone.utc) - timedelta(minutes=self.cacheLifetime) + + if file_mtime < cache_expiry: + logger.debug('Cache file is stale (mtime: %s, expiry: %s)', + file_mtime, cache_expiry) + return True + + logger.debug('Cache file is still valid (mtime: %s)', file_mtime) + return False + except OSError as ex: + logger.warning('Error checking cache file: %s', ex) + return True + + def _fetch_and_cache(self) -> bool: + """Downloads the ICS file and saves it to the cache. + + Returns True on success, False on failure.""" + cache_path = self._get_cache_path() + + try: + logger.debug('Fetching ICS from URL: %s', self.url) + + # Create cache directory if it doesn't exist + cache_dir = os.path.dirname(cache_path) + if cache_dir and not os.path.exists(cache_dir): + os.makedirs(cache_dir) + + # Download the ICS file + with urllib.request.urlopen(self.url, timeout=30) as response: + ics_content = response.read() + + # Write to cache file + with open(cache_path, 'wb') as cache_file: + cache_file.write(ics_content) + + # Invalidate cached calendar object + self._calendar = None + + logger.debug('Successfully cached ICS file to: %s', cache_path) + return True + + except Exception as ex: # pylint: disable=broad-except + logger.warning('Error fetching ICS file: %s', ex) + logger.exception(ex) + return False + + def _load_calendar(self) -> icalendar.Calendar | None: + """Loads and parses the cached ICS file. + + Returns the calendar object or None on error.""" + if self._calendar is not None: + return self._calendar + + cache_path = self._get_cache_path() + try: + with open(cache_path, 'rb') as f: + self._calendar = icalendar.Calendar.from_ical(f.read()) + return self._calendar + except Exception as ex: # pylint: disable=broad-except + logger.warning('Error loading ICS file: %s', ex) + return None + + def _get_event_status(self, event: icalendar.Event) -> enum.Status: + """Determines the Status-Light status for a single iCal event. + + RFC 5545 compliant mapping based on TRANSP and STATUS properties: + - TRANSP=TRANSPARENT → FREE (event doesn't block time) + - STATUS=CANCELLED → FREE (event was cancelled) + - STATUS=TENTATIVE → TENTATIVE (maps to BUSY-TENTATIVE in RFC terms) + - STATUS=CONFIRMED or None → BUSY (default blocking event) + + Additionally handles Microsoft CDO extensions (common in Office 365/Outlook): + - X-MICROSOFT-CDO-BUSYSTATUS=FREE (0) → FREE + - X-MICROSOFT-CDO-BUSYSTATUS=TENTATIVE (1) → TENTATIVE + - X-MICROSOFT-CDO-BUSYSTATUS=BUSY (2) → BUSY + - X-MICROSOFT-CDO-BUSYSTATUS=OOF (3) → OUTOFOFFICE + - X-MICROSOFT-CDO-BUSYSTATUS=WORKINGELSEWHERE (4) → WORKINGELSEWHERE + """ + summary = str(event.get('SUMMARY', 'Untitled')) + + # Transparent events don't block time (e.g., all-day reminders) + transp = str(event.get('TRANSP', 'OPAQUE')).upper() + if transp == 'TRANSPARENT': + logger.debug('Event "%s" is transparent, treating as FREE', summary) + return enum.Status.FREE + + # Check the STATUS property + status = event.get('STATUS') + status_str = str(status).upper() if status else None + + if status_str == 'CANCELLED': + logger.debug('Event "%s" is cancelled, treating as FREE', summary) + return enum.Status.FREE + + if status_str == 'TENTATIVE': + logger.debug('Event "%s" is tentative, treating as TENTATIVE', summary) + return enum.Status.TENTATIVE + + # Check for Microsoft CDO busy status extension (common in Office 365/Outlook ICS exports) + # This property provides more granular status information than standard RFC 5545 + busystatus = event.get('X-MICROSOFT-CDO-BUSYSTATUS') + if busystatus: + busystatus_str = str(busystatus).upper() + + # Map numeric and string values to Status enums + # Office 365/Outlook uses both numeric codes and string values + if busystatus_str in ('0', 'FREE'): + logger.debug('Event "%s" has CDO busystatus FREE, treating as FREE', summary) + return enum.Status.FREE + elif busystatus_str in ('1', 'TENTATIVE'): + logger.debug('Event "%s" has CDO busystatus TENTATIVE, treating as TENTATIVE', + summary) + return enum.Status.TENTATIVE + elif busystatus_str in ('3', 'OOF', 'OUT-OF-OFFICE'): + logger.debug('Event "%s" has CDO busystatus OOF, treating as OUTOFOFFICE', summary) + return enum.Status.OUTOFOFFICE + elif busystatus_str in ('4', 'WORKINGELSEWHERE', 'WORKING-ELSEWHERE'): + logger.debug('Event "%s" has CDO busystatus WORKINGELSEWHERE, treating as WORKINGELSEWHERE', + summary) + return enum.Status.WORKINGELSEWHERE + elif busystatus_str in ('2', 'BUSY'): + logger.debug('Event "%s" has CDO busystatus BUSY, treating as BUSY', summary) + return enum.Status.BUSY + + # CONFIRMED or no status = BUSY (default per RFC 5545) + logger.debug('Event "%s" is confirmed/opaque, treating as BUSY', summary) + return enum.Status.BUSY + + def get_current_status(self) -> enum.Status: + """Retrieves the ICS calendar status within the lookahead period. + + RFC 5545 compliant behavior with Microsoft CDO extensions: + - Returns BUSY if any confirmed opaque events exist + - Returns OUTOFOFFICE if any OOF events exist + - Returns WORKINGELSEWHERE if any working-elsewhere events exist + - Returns TENTATIVE if only tentative events exist + - Returns FREE if no blocking events exist + - Returns UNKNOWN on error + + Status precedence: BUSY > OUTOFOFFICE > WORKINGELSEWHERE > TENTATIVE > FREE + """ + try: + # Refresh cache if needed + if self._should_refresh_cache(): + if not self._fetch_and_cache(): + # If fetch fails but we have an old cache, try to use it + cache_path = self._get_cache_path() + if not os.path.exists(cache_path): + logger.warning('No cached ICS file available') + return enum.Status.UNKNOWN + logger.warning('Using stale cache file due to fetch failure') + + # Load the calendar + calendar = self._load_calendar() + if calendar is None: + return enum.Status.UNKNOWN + + # Get events within the lookahead window using timezone-aware datetimes + # recurring-ical-events handles timezone conversion correctly + local_tz = datetime.now().astimezone().tzinfo + start_time = datetime.now(local_tz) + end_time = start_time + timedelta(minutes=self.lookahead) + + logger.debug('Checking for events between %s and %s', + start_time.strftime('%I:%M %p %Z'), + end_time.strftime('%I:%M %p %Z')) + + # Use recurring-ical-events to expand recurring events and filter by time + calendar_events = recurring_ical_events.of(calendar).between(start_time, end_time) + + if not calendar_events: + logger.debug('No events in lookahead window') + return enum.Status.FREE + + logger.debug('Found %d event(s) in lookahead window', len(calendar_events)) + + # Determine the "busiest" status from all events + # Precedence: BUSY > OUTOFOFFICE > WORKINGELSEWHERE > TENTATIVE > FREE + has_outofoffice = False + has_workingelsewhere = False + has_tentative = False + + for event in calendar_events: + dtstart = event.get('DTSTART') + dtend = event.get('DTEND') + summary = str(event.get('SUMMARY', 'Untitled')) + transp = str(event.get('TRANSP', 'OPAQUE')) + status = str(event.get('STATUS', '')) + + # Convert to local timezone for logging + start_dt = dtstart.dt if dtstart else None + end_dt = dtend.dt if dtend else start_dt + if hasattr(start_dt, 'astimezone'): + start_local = start_dt.astimezone(local_tz) + end_local = end_dt.astimezone(local_tz) if hasattr(end_dt, 'astimezone') else end_dt + else: + start_local = start_dt + end_local = end_dt + + logger.debug('Event: %s (%s - %s, transp=%s, status=%s)', + summary, + start_local.strftime('%I:%M %p') if start_local else '?', + end_local.strftime('%I:%M %p') if end_local else '?', + transp, status) + + event_status = self._get_event_status(event) + + if event_status == enum.Status.BUSY: + # BUSY takes highest precedence, return immediately + return enum.Status.BUSY + + if event_status == enum.Status.OUTOFOFFICE: + has_outofoffice = True + + if event_status == enum.Status.WORKINGELSEWHERE: + has_workingelsewhere = True + + if event_status == enum.Status.TENTATIVE: + has_tentative = True + # FREE events don't affect the result + + # Return the highest precedence status found + # OUTOFOFFICE > WORKINGELSEWHERE > TENTATIVE > FREE + if has_outofoffice: + return enum.Status.OUTOFOFFICE + + if has_workingelsewhere: + return enum.Status.WORKINGELSEWHERE + + if has_tentative: + return enum.Status.TENTATIVE + + # All events were transparent or cancelled + return enum.Status.FREE + + except (SystemExit, KeyboardInterrupt): + return enum.Status.UNKNOWN + except Exception as ex: # pylint: disable=broad-except + logger.warning('Exception while getting ICS status: %s', ex) + logger.exception(ex) + return enum.Status.UNKNOWN diff --git a/status-light/sources/calendar/office365.py b/status-light/sources/calendar/office365.py index fc78fe0..ba6f4fb 100755 --- a/status-light/sources/calendar/office365.py +++ b/status-light/sources/calendar/office365.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Office 365 Source diff --git a/status-light/sources/collaboration/slack.py b/status-light/sources/collaboration/slack.py index a2136f8..d3aab11 100755 --- a/status-light/sources/collaboration/slack.py +++ b/status-light/sources/collaboration/slack.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Slack Source diff --git a/status-light/sources/collaboration/webex.py b/status-light/sources/collaboration/webex.py index dfad9a2..62b3033 100755 --- a/status-light/sources/collaboration/webex.py +++ b/status-light/sources/collaboration/webex.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Webex Teams Source diff --git a/status-light/status-light.py b/status-light/status-light.py index 9095ed1..fe10b9b 100755 --- a/status-light/status-light.py +++ b/status-light/status-light.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Main Application Entry Point @@ -17,11 +17,11 @@ # Project imports # 47 - Add Google support -from sources.calendar import google, office365 +from sources.calendar import google, ics, office365 # 48 - Add Slack support from sources.collaboration import slack, webex -from targets import tuya -from utility import enum, env, util +from targets import tuya, virtual +from utility import enum, env, util, precedence class StatusLight: @@ -40,9 +40,10 @@ class StatusLight: slack_api: slack.SlackAPI = slack.SlackAPI() office_api: office365.OfficeAPI = office365.OfficeAPI() google_api: google.GoogleCalendarAPI = google.GoogleCalendarAPI() + ics_api: ics.Ics = ics.Ics() # Target Properties - light: tuya.TuyaLight = tuya.TuyaLight() + light: tuya.TuyaLight | virtual.VirtualLight = tuya.TuyaLight() def init(self): """Initializes all class and environment variables.""" @@ -56,7 +57,7 @@ def init(self): # Validate environment variables in a structured way if False in [self.local_env.get_sources(), - self.local_env.get_tuya(), + self.local_env.get_target(), self.local_env.get_colors(), self.local_env.get_status(), self.local_env.get_active_time(), @@ -134,15 +135,32 @@ def init(self): 'Requested Google, but could not find all environment variables!') sys.exit(1) - # Tuya - self.light.device = self.local_env.tuya_device - self.logger.debug('Retrieved TUYA_DEVICE variable: %s', self.light.device) - tuya_status = self.light.get_status() - self.logger.debug('Found initial Tuya status: %s', tuya_status) - if not tuya_status: - self.logger.error( - 'Could not connect to Tuya device!') - sys.exit(1) + # ICS Calendar support + if enum.StatusSource.ICS in self.local_env.selected_sources: + if self.local_env.get_ics(): + self.logger.info('Requested ICS') + self.ics_api.url = self.local_env.ics_url + self.ics_api.cacheStore = self.local_env.ics_cache_store + self.ics_api.cacheLifetime = self.local_env.ics_cache_lifetime + # 81 - Make calendar lookahead configurable + self.ics_api.lookahead = self.local_env.calendar_lookahead + else: + self.logger.error( + 'Requested ICS, but could not find all environment variables!') + sys.exit(1) + + # Target initialization + if self.local_env.target == 'virtual': + self.logger.info('Using virtual light target') + self.light = virtual.VirtualLight() + else: + # Tuya target (requires TUYA_DEVICE) + if not self.local_env.get_tuya(): + self.logger.error( + 'TUYA_DEVICE is required when TARGET=tuya!') + sys.exit(1) + self.light.device = self.local_env.tuya_device + self.logger.debug('Retrieved TUYA_DEVICE variable: %s', self.light.device) def run(self): """Runs the main loop of the application""" @@ -166,6 +184,7 @@ def run(self): slack_status = enum.Status.UNKNOWN office_status = enum.Status.UNKNOWN google_status = enum.Status.UNKNOWN + ics_status = enum.Status.UNKNOWN logger_format = " {}: {} |" logger_string = "" @@ -198,39 +217,38 @@ def run(self): logger_format.format(enum.StatusSource.GOOGLE.name.capitalize(), google_status.name.lower()) + # ICS Status (based on calendar) + if enum.StatusSource.ICS in self.local_env.selected_sources: + ics_status = self.ics_api.get_current_status() + logger_string += \ + logger_format.format(enum.StatusSource.ICS.name.capitalize(), + ics_status.name.lower()) + # 74: Log enums as names, not values 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 - # Compare statii and pick a winner - # Collaboration status always wins except in specific scenarios - # Webex currently takes precendence over Slack - self.current_status = webex_status - if (webex_status == enum.Status.UNKNOWN or - webex_status in self.local_env.off_status): - # 74: Log enums as names, not values - self.logger.debug('Using slack_status: %s', - slack_status.name.lower()) - # Fall through to Slack - self.current_status = slack_status - - if (self.current_status in self.local_env.available_status or - self.current_status in self.local_env.off_status) \ - and (office_status not in self.local_env.off_status - or google_status not in self.local_env.off_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: - self.logger.debug('Using office_status: %s', - office_status.name.lower()) - self.current_status = office_status - else: - self.logger.debug('Using google_status: %s', - google_status.name.lower()) - self.current_status = google_status + # Build input dictionaries from retrieved statuses + collaboration_statuses = {} + if enum.StatusSource.WEBEX in self.local_env.selected_sources: + collaboration_statuses[enum.StatusSource.WEBEX] = webex_status + if enum.StatusSource.SLACK in self.local_env.selected_sources: + collaboration_statuses[enum.StatusSource.SLACK] = slack_status + + calendar_statuses = {} + if enum.StatusSource.OFFICE365 in self.local_env.selected_sources: + calendar_statuses[enum.StatusSource.OFFICE365] = office_status + if enum.StatusSource.GOOGLE in self.local_env.selected_sources: + calendar_statuses[enum.StatusSource.GOOGLE] = google_status + if enum.StatusSource.ICS in self.local_env.selected_sources: + calendar_statuses[enum.StatusSource.ICS] = ics_status + + # Call precedence module to select winning status + self.current_status, winning_source = precedence.select_status( + collaboration_statuses, + calendar_statuses, + self.local_env.off_status, + self.local_env.available_status + ) status_changed = False if self.last_status != self.current_status: @@ -259,13 +277,13 @@ def run(self): if not already_handled_inactive_hours: self.logger.info( 'Outside of active hours, transitioning to off') - last_transition_result = self.light.off() + last_transition_result = self.light.turn_off() self.last_status = enum.Status.UNKNOWN already_handled_inactive_hours = True # 40: If the last transition failed, try again if not last_transition_result: - last_transition_result = self.light.off() + last_transition_result = self.light.turn_off() # Sleep for a few seconds time.sleep(self.local_env.sleep_seconds) @@ -278,7 +296,7 @@ def run(self): self.logger.exception(ex) self.logger.debug('Turning light off') - self.light.off() + self.light.turn_off() def _transition_status(self) -> bool: """Internal Helper Method to determine the correct color for the light @@ -299,18 +317,17 @@ def _transition_status(self) -> bool: color = self.local_env.busy_color if color is not None: - return_value = self.light.set_status( - enum.TuyaMode.COLOR, color, self.local_env.tuya_brightness) + return_value = self.light.set_color(color, self.local_env.light_brightness) # OffStatus has the lowest priority, so only check it if none of the others are valid elif self.current_status in self.local_env.off_status: - return_value = self.light.off() + return_value = self.light.turn_off() # In the case that we made it here without a valid state, # just turn the light off and warn about it # 74: Log enums as names, not values else: self.logger.warning('Called with an invalid status: %s', self.current_status.name.lower()) - return_value = self.light.off() + return_value = self.light.turn_off() return return_value diff --git a/status-light/targets/base.py b/status-light/targets/base.py new file mode 100644 index 0000000..5e8f47a --- /dev/null +++ b/status-light/targets/base.py @@ -0,0 +1,79 @@ +"""Status-Light +(c) 2020-2026 Nick Warner +https://github.com/portableprogrammer/Status-Light/ + +Abstract Light Target Base Class +""" + +# Standard imports +import re +from abc import ABC, abstractmethod + + +class LightTarget(ABC): + """Abstract base class for light targets. + + Defines the interface that all light targets must implement. + Status-Light uses a simple model: set a color with brightness, or turn off. + """ + + @abstractmethod + def set_color(self, color: str, brightness: int) -> bool: + """Set light to specified color and brightness. + + Args: + color: 6-character hex RGB color (e.g., 'ff0000' for red) + brightness: Brightness percentage 0-100 (0 effectively off, but + use turn_off() for clarity) + + Returns: + True if successful, False otherwise + + Note: + Implementations should convert the percentage brightness to their device-specific + format (e.g., Tuya uses 0-255, Hue uses 0-254, LIFX uses 0.0-1.0). + """ + pass + + @abstractmethod + def turn_off(self) -> bool: + """Turn the light completely off. + + Returns: + True if successful, False otherwise + + Note: + This is preferred over set_color() with brightness=0 for clarity and + to allow implementations to optimize for the off state. + """ + pass + + def validate_color(self, color: str) -> bool: + """Validate color format (6-character hex RGB). + + Args: + color: String to validate as hex color + + Returns: + True if color is valid 6-character hex, False otherwise + + Note: + Implementations can override this for custom validation or + to accept additional color formats. + """ + return bool(re.match(r'^[0-9A-Fa-f]{6}$', color)) + + def validate_brightness(self, brightness: int) -> bool: + """Validate brightness percentage (0-100). + + Args: + brightness: Integer percentage value to validate + + Returns: + True if brightness is in valid range, False otherwise + + Note: + Base class expects percentage (0-100). Implementations convert this to + device-specific ranges. Override this method if you need different validation. + """ + return 0 <= brightness <= 100 diff --git a/status-light/targets/tuya.py b/status-light/targets/tuya.py index 94f4f08..696a82e 100755 --- a/status-light/targets/tuya.py +++ b/status-light/targets/tuya.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Tuya Target @@ -13,62 +13,91 @@ import tuyaface # Project imports -from utility import enum +from targets.base import LightTarget logger = logging.getLogger(__name__) -class TuyaLight: - """Represents a Tuya device, utilizing `tuyaface` for connections.""" - device: dict +class TuyaLight(LightTarget): + """Represents a Tuya device, utilizing `tuyaface` for connections. - def on(self): # pylint: disable=invalid-name - """Turns on the light.""" - return self._set_status({'1': True}) + Implements the LightTarget interface for Tuya-compatible RGB bulbs. + Always uses COLOR mode internally (not white temperature mode). + """ + device: dict - def off(self): - """Turns off the light.""" - return self._set_status({'1': False}) + def set_color(self, color: str, brightness: int) -> bool: + """Set light to specified color and brightness. - def get_status(self): - """Retrieves the current status of the Tuya device. - May return an error if the device is unavailable.""" - return tuyaface.status(self.device) + Args: + color: 6-character hex RGB color (e.g., 'ff0000' for red) + brightness: Brightness percentage 0-100 - def set_status(self, mode: enum.TuyaMode = enum.TuyaMode.WHITE, - color: str = enum.Color.WHITE.value, brightness: int = 128) -> bool: - """Sends a full command to the Tuya device + Returns: + True if successful, False otherwise - `mode`: `white`, `colour` - `color`: A hex-formatted RGB color (e.g. `rrggbb`) - `brightness`: An `int` between 0 and 255 inclusive, though the lowest usable threshold is 32 + Note: + Tuya devices require specific DPS (Data Point) format. This method + automatically converts the percentage brightness to Tuya's 0-255 scale, + converts the hex color to Tuya's expected format, and locks the device + to COLOR mode. """ - # DPS: + # Validate inputs using base class helpers + if not self.validate_color(color): + logger.warning('Invalid color format: %s (expected 6-char hex)', color) + return False + + if not self.validate_brightness(brightness): + logger.warning('Invalid brightness: %s (expected 0-100 percentage)', brightness) + return False + + # Convert percentage (0-100) to Tuya's device scale (0-255) + # Note: Tuya devices have a minimum usable threshold around 32-64 + device_brightness = int(brightness * 255 / 100) + + # Tuya DPS (Data Points): # 1: Power, bool # 2: Mode, 'white' or 'colour' - # 3: Brightness, int 0-255, but the lowest usable threshold is between 32 and 64 - # 5: Color, 'rrggbb0000ffff' + # 3: Brightness, int 0-255 (lowest usable threshold is 32-64) + # 5: Color, 'rrggbb0000ffff' (RGB + HSV suffix required by device) - # Ensure the color has all the bits we care about + # Convert simple hex RGB to Tuya's DPS format if not color.endswith('0000ffff'): color = color + '0000ffff' - # Light must be on (so we'll do that first) - # Brightness should be next - # Color must be before Mode - + # Command order matters: Power → Brightness → Color → Mode dps: dict = { - '1': True, - '3': brightness, - '5': color, - '2': mode.value + '1': True, # Power on + '3': device_brightness, # Set brightness (converted to 0-255) + '5': color, # Set color (with Tuya suffix) + '2': 'colour' # Lock to COLOR mode (not white temperature) } - return_vaue = self._set_status(dps) - return return_vaue + return self._set_dps(dps) + + def turn_off(self) -> bool: + """Turn the light completely off. - def _set_status(self, dps: dict, retry: int = 5): - """Internal Helper Method for `set_status`""" + Returns: + True if successful, False otherwise + """ + return self._set_dps({'1': False}) + + def _set_dps(self, dps: dict, retry: int = 5) -> bool: + """Internal helper method for sending DPS commands to Tuya device. + + Args: + dps: Dictionary of Data Point values to set + retry: Number of retry attempts on failure (default: 5) + + Returns: + True if successful, False otherwise + + Note: + Implements retry logic with 1-second delays between attempts. + Clears the tuyaface connection cache after each attempt to avoid + broken pipe errors. + """ # We sometimes get a connection reset, or other errors, so let's retry after a second count = 0 status = False diff --git a/status-light/targets/virtual.py b/status-light/targets/virtual.py new file mode 100644 index 0000000..524d6b9 --- /dev/null +++ b/status-light/targets/virtual.py @@ -0,0 +1,58 @@ +"""Status-Light +(c) 2020-2026 Nick Warner +https://github.com/portableprogrammer/Status-Light/ + +Virtual Light Target for testing +""" + +import logging + +from targets.base import LightTarget + +logger = logging.getLogger(__name__) + + +class VirtualLight(LightTarget): + """Virtual light target that logs status changes for testing. + + Implements the LightTarget interface without requiring any physical hardware. + Useful for development, testing, and demonstrations. + """ + _power: bool = False + _color: str = 'ffffff' + _brightness: int = 128 + + def set_color(self, color: str, brightness: int) -> bool: + """Set virtual light to specified color and brightness. + + Args: + color: 6-character hex RGB color (e.g., 'ff0000' for red) + brightness: Brightness level 0-255 + + Returns: + Always returns True (virtual light cannot fail) + """ + # Validate inputs using base class helpers + if not self.validate_color(color): + logger.warning('[Virtual] Invalid color format: %s (expected 6-char hex)', color) + return False + + if not self.validate_brightness(brightness): + logger.warning('[Virtual] Invalid brightness: %s (expected 0-255)', brightness) + return False + + self._power = True + self._color = color + self._brightness = brightness + logger.info('[Virtual] set_color(color=%s, brightness=%s)', color, brightness) + return True + + def turn_off(self) -> bool: + """Turn the virtual light off. + + Returns: + Always returns True (virtual light cannot fail) + """ + self._power = False + logger.info('[Virtual] turn_off()') + return True diff --git a/status-light/utility/enum.py b/status-light/utility/enum.py index 334be44..125d987 100755 --- a/status-light/utility/enum.py +++ b/status-light/utility/enum.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Enumeration Definition @@ -19,6 +19,8 @@ class StatusSource(enum.IntEnum): GOOGLE = enum.auto() # 48 - Add Slack suport SLACK = enum.auto() + # ICS Calendar support + ICS = enum.auto() @classmethod def _missing_(cls, value): diff --git a/status-light/utility/env.py b/status-light/utility/env.py index 4492af5..22e20e9 100755 --- a/status-light/utility/env.py +++ b/status-light/utility/env.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Environment Variable Management @@ -21,7 +21,7 @@ class Environment: """Represents a structured set of environment variables passed to the application.""" tuya_device: dict = {} - tuya_brightness: int = 128 + light_brightness: int = 50 # Percentage (0-100) # 32 - SOURCES variable default is wrong # 32 Recurred; _parseSource expects a string, not a list. Convert the list to a string. @@ -51,6 +51,14 @@ class Environment: google_credential_store: str = './utility/api/calendar/google' google_token_store: str = '~' + # ICS Calendar support + ics_url: str = '' + ics_cache_store: str = '~' + ics_cache_lifetime: int = 30 + + # Target selection + target: str = 'tuya' + # 38 - Working Elsewhere isn't handled off_status: list[enum.Status] = [enum.Status.INACTIVE, enum.Status.OUTOFOFFICE, @@ -122,12 +130,24 @@ def get_tuya(self) -> bool: return_value = False # 41: Replace decorator with utility function - self.tuya_brightness = util.try_parse_int(os.environ.get('TUYA_BRIGHTNESS', ''), - self.tuya_brightness) - # 34 - Better environment variable errors - # TUYA_BRIGHTNESS should be within the range 32..255 - if self.tuya_brightness < 32 or self.tuya_brightness > 255: - logger.warning('TUYA_BRIGHTNESS must be between 32 and 255!') + # Support both LIGHT_BRIGHTNESS and TUYA_BRIGHTNESS (backward compatibility) + brightness_value = os.environ.get('LIGHT_BRIGHTNESS', + os.environ.get('TUYA_BRIGHTNESS', '')) + raw_brightness = util.try_parse_int(brightness_value, self.light_brightness) + + # Auto-detect legacy 0-255 format and convert to percentage + # Simple heuristic: Values > 100 are legacy format (percentage is 0-100) + if raw_brightness > 100: + self.light_brightness = int(raw_brightness * 100 / 255) + logger.info('LIGHT_BRIGHTNESS=%d detected as legacy 0-255 format, ' + 'converted to %d%% (use 0-100 for percentages)', + raw_brightness, self.light_brightness) + else: + self.light_brightness = raw_brightness + + # Validate brightness is in percentage range (0-100) + if self.light_brightness < 0 or self.light_brightness > 100: + logger.warning('LIGHT_BRIGHTNESS must be between 0 and 100 (percentage)!') return_value = False return return_value @@ -181,6 +201,28 @@ def get_google(self) -> bool: self.google_token_store) return ('' not in [self.google_credential_store, self.google_token_store]) + def get_ics(self) -> bool: + """Retrieves and validates the `ICS_*` variables.""" + # ICS_URL is required and may contain secrets + self.ics_url = util.get_env_or_secret('ICS_URL', '') + self.ics_cache_store = os.environ.get('ICS_CACHESTORE', self.ics_cache_store) + self.ics_cache_lifetime = util.try_parse_int( + os.environ.get('ICS_CACHELIFETIME', ''), + self.ics_cache_lifetime) + # Validate cache lifetime is within 5-60 minutes + if self.ics_cache_lifetime < 5 or self.ics_cache_lifetime > 60: + logger.warning('ICS_CACHELIFETIME must be between 5 and 60 minutes!') + self.ics_cache_lifetime = 30 + return self.ics_url != '' + + def get_target(self) -> bool: + """Retrieves and validates the `TARGET` variable.""" + self.target = os.environ.get('TARGET', self.target).lower() + if self.target not in ('tuya', 'virtual'): + logger.warning('TARGET must be "tuya" or "virtual"!') + return False + return True + def get_colors(self) -> bool: """Retrieves and validates the `*_COLOR` variables.""" self.available_color = util.parse_color(os.environ.get('AVAILABLE_COLOR', ''), diff --git a/status-light/utility/precedence.py b/status-light/utility/precedence.py new file mode 100644 index 0000000..6f2231c --- /dev/null +++ b/status-light/utility/precedence.py @@ -0,0 +1,144 @@ +"""Status-Light +(c) 2020-2026 Nick Warner +https://github.com/portableprogrammer/Status-Light/ + +Status Precedence Selection Logic +""" + +# Standard imports +import logging + +# Project imports +from utility import enum + +logger: logging.Logger = logging.getLogger(__name__) + + +def select_status( + collaboration_statuses: dict[enum.StatusSource, enum.Status], + calendar_statuses: dict[enum.StatusSource, enum.Status], + off_status: list[enum.Status], + available_status: list[enum.Status] +) -> tuple[enum.Status, enum.StatusSource]: + """Selects the winning status from multiple collaboration and calendar sources. + + Applies a hierarchical precedence algorithm: + 1. Collaboration sources (Webex, Slack) take precedence over Calendar sources + 2. Exception: Calendar overrides when collaboration is UNKNOWN or in off_status + AND at least one calendar source has events (not in off_status) + 3. Within Collaboration: Webex > Slack + 4. Within Calendar: Office365 > Google > ICS + + Args: + collaboration_statuses: Dictionary mapping StatusSource to Status for + collaboration sources (e.g., {StatusSource.WEBEX: Status.MEETING}) + calendar_statuses: Dictionary mapping StatusSource to Status for + calendar sources (e.g., {StatusSource.OFFICE365: Status.BUSY}) + off_status: List of Status values considered "off" (e.g., [Status.INACTIVE]) + available_status: List of Status values considered "available" + (e.g., [Status.ACTIVE, Status.FREE]) + + Returns: + Tuple of (winning_status, winning_source) where: + - winning_status: The selected Status enum value + - winning_source: The StatusSource enum indicating which source won + Returns (Status.UNKNOWN, StatusSource.UNKNOWN) if no valid status found + + Examples: + >>> # Webex takes precedence over Slack + >>> select_status( + ... {StatusSource.WEBEX: Status.MEETING, StatusSource.SLACK: Status.ACTIVE}, + ... {}, + ... [Status.INACTIVE], + ... [Status.ACTIVE] + ... ) + (Status.MEETING, StatusSource.WEBEX) + + >>> # Calendar overrides when collaboration is available and calendar is busy + >>> select_status( + ... {StatusSource.WEBEX: Status.ACTIVE}, + ... {StatusSource.OFFICE365: Status.BUSY}, + ... [Status.INACTIVE], + ... [Status.ACTIVE] + ... ) + (Status.BUSY, StatusSource.OFFICE365) + """ + # Log input sources for debugging + if collaboration_statuses: + collab_log = ', '.join(f'{src.name.lower()}={status.name.lower()}' + for src, status in collaboration_statuses.items()) + logger.debug('Collaboration sources: %s', collab_log) + else: + logger.debug('Collaboration sources: none') + + if calendar_statuses: + calendar_log = ', '.join(f'{src.name.lower()}={status.name.lower()}' + for src, status in calendar_statuses.items()) + logger.debug('Calendar sources: %s', calendar_log) + else: + logger.debug('Calendar sources: none') + + # Phase 1: Select collaboration status + # Check Webex first (highest priority) + current_status = collaboration_statuses.get(enum.StatusSource.WEBEX, enum.Status.UNKNOWN) + winning_source = enum.StatusSource.WEBEX + + # If Webex is UNKNOWN or in off_status, fallback to Slack + if (current_status == enum.Status.UNKNOWN or current_status in off_status): + slack_status = collaboration_statuses.get(enum.StatusSource.SLACK, enum.Status.UNKNOWN) + if slack_status != enum.Status.UNKNOWN: + logger.debug('Using slack_status: %s', slack_status.name.lower()) + current_status = slack_status + winning_source = enum.StatusSource.SLACK + elif current_status != enum.Status.UNKNOWN: + # Webex returned an off_status value, using it + logger.debug('Using webex_status: %s (off_status)', current_status.name.lower()) + else: + # Webex has a valid collaboration status, using it + logger.debug('Using webex_status: %s', current_status.name.lower()) + + # Phase 2: Check calendar override condition + # Calendar overrides if: + # - Current status is in available_status OR off_status + # - At least one calendar source is NOT in off_status (has events) + if (current_status in available_status or current_status in off_status): + # Check if any calendar source has events + office_status = calendar_statuses.get(enum.StatusSource.OFFICE365, enum.Status.UNKNOWN) + google_status = calendar_statuses.get(enum.StatusSource.GOOGLE, enum.Status.UNKNOWN) + ics_status = calendar_statuses.get(enum.StatusSource.ICS, enum.Status.UNKNOWN) + + if (office_status not in off_status or + google_status not in off_status or + ics_status not in off_status): + + logger.debug('Using calendar-based status (overriding collaboration status: %s)', + current_status.name.lower()) + + # Phase 3: Select calendar status in priority order + # Office365 > Google > ICS + if office_status != enum.Status.UNKNOWN: + logger.debug('Using office_status: %s', office_status.name.lower()) + current_status = office_status + winning_source = enum.StatusSource.OFFICE365 + elif google_status != enum.Status.UNKNOWN: + logger.debug('Using google_status: %s', google_status.name.lower()) + current_status = google_status + winning_source = enum.StatusSource.GOOGLE + else: + # Use ICS as final fallback even if UNKNOWN + logger.debug('Using ics_status: %s', ics_status.name.lower()) + current_status = ics_status + winning_source = enum.StatusSource.ICS + else: + logger.debug('Calendar sources all off, keeping collaboration status: %s', + current_status.name.lower()) + else: + logger.debug('Collaboration status %s not overrideable, keeping it', + current_status.name.lower()) + + # Log final decision + logger.debug('Selected status: %s from source: %s', + current_status.name.lower(), winning_source.name.lower()) + + # Return the winning status and its source + return (current_status, winning_source) diff --git a/status-light/utility/util.py b/status-light/utility/util.py index a577944..cf7dd65 100755 --- a/status-light/utility/util.py +++ b/status-light/utility/util.py @@ -1,5 +1,5 @@ """Status-Light -(c) 2020-2023 Nick Warner +(c) 2020-2026 Nick Warner https://github.com/portableprogrammer/Status-Light/ Utility Functions