refactor: strip redundant wrappers from presence monitor example

Author: vgrem

examples/communications/presence/teams_presence_monitor.py

@@ -1,54 +1,23 @@
 """
-Teams presence and telephony status monitor — practical real-world example.
+Teams presence and telephony status monitor.
 
-A call center / helpdesk scenario: monitor agent availability across a team,
-detect state changes, bulk-query presence, and respond programmatically.
+Covers: single/bulk presence queries, preferred presence, status messages,
+polling for state changes, and business-rules routing logic.
 
-Covers:
-  - Getting a single user's presence
-  - Getting presence for multiple users at once (bulk)
-  - Setting and clearing preferred presence
-  - Setting a status message with expiry
-  - Polling for state changes
-  - Mapping presence states to business rules
-
-Required permissions:
-  - Presence.Read (delegated)        — read own presence
-  - Presence.Read.All (application)  — read any user's presence
-  - Presence.ReadWrite (delegated)   — set own presence + status message
-
-See https://learn.microsoft.com/en-us/graph/api/resources/presence
+Required permissions: Presence.Read, Presence.Read.All, Presence.ReadWrite.
 """
 
-from __future__ import annotations
-
 from datetime import datetime, timedelta, timezone
 from time import sleep
-from typing import Dict, List, Optional
+from typing import Optional
 
 from office365.graph_client import GraphClient
 
-# ---------------------------------------------------------------------------
-# Configuration
-# ---------------------------------------------------------------------------
-
-SUPPORTED_AGENTS = [
-    "alice@company.com",
-    "bob@company.com",
-    "carol@company.com",
-]
-
-POLL_INTERVAL_SECONDS = 30
+POLL_INTERVAL = 30
 BUSY_ACTIVITIES = {"InACall", "InAConferenceCall", "InAMeeting", "Presenting"}
 
 
-# ---------------------------------------------------------------------------
-# Business logic helpers
-# ---------------------------------------------------------------------------
-
-
 def is_available(presence) -> bool:
-    """Determine if a user is considered 'available' for routing."""
     availability = presence.get_property("availability")
     activity = presence.get_property("activity")
     if not availability or not activity:
@@ -61,175 +30,73 @@ def is_available(presence) -> bool:
 
 
 def agent_status_summary(presence) -> str:
-    """Human-readable summary of an agent's current state."""
-    availability = presence.get_property("availability") or "Unknown"
-    activity = presence.get_property("activity") or "Unknown"
-    return f"availability={availability}, activity={activity}"
-
-
-# ---------------------------------------------------------------------------
-# Core operations
-# ---------------------------------------------------------------------------
-
-
-def get_presence_for_user(client: GraphClient, user_id: str):
-    """Get presence for a single user by ID or UPN."""
-    return client.communications.presences[user_id].get().execute_query()
-
-
-def get_presences_for_team(client: GraphClient, user_ids: List[str]):
-    """Get presence for multiple users in a single API call (bulk)."""
-    return client.communications.get_presences_by_user_id(user_ids).execute_query()
+    a = presence.get_property("availability") or "Unknown"
+    b = presence.get_property("activity") or "Unknown"
+    return f"availability={a}, activity={b}"
 
 
 def take_agent_offline(client: GraphClient, user_id: str):
-    """Mark an agent as DoNotDisturb + set an end-of-day status message."""
     expires = datetime.now(timezone.utc) + timedelta(hours=9)
-
     client.communications.presences[user_id].set_user_preferred_presence(
-        availability="DoNotDisturb",
-        activity="OffWork",
-        expiration_duration="PT9H",
+        availability="DoNotDisturb", activity="OffWork", expiration_duration="PT9H",
     ).execute_query()
-
     client.communications.presences[user_id].set_status_message(
-        message="Shift ended — back tomorrow",
-        expiry=expires,
+        message="Shift ended", expiry=expires,
     ).execute_query()
 
 
-def set_agent_available(client: GraphClient, user_id: str):
-    """Return an agent to available state after break."""
+def start_shift(client: GraphClient, user_id: str):
+    expires = datetime.now(timezone.utc) + timedelta(hours=9)
     client.communications.presences[user_id].set_user_preferred_presence(
-        availability="Available",
-        activity="Available",
-        expiration_duration="PT8H",
+        availability="Available", activity="Available", expiration_duration="PT9H",
+    ).execute_query()
+    client.communications.presences[user_id].set_status_message(
+        message="Online and ready", expiry=expires,
     ).execute_query()
 
 
-# ---------------------------------------------------------------------------
-# Scenario 1: snapshot — check current team status
-# ---------------------------------------------------------------------------
-
-
-def snapshot_team_status(client: GraphClient, agents: List[str]):
-    """One-shot display of every agent's current presence."""
-    print("=== Team presence snapshot ===")
-    presences = get_presences_for_team(client, agents)
-    for entry in presences:
-        print(f"  {entry.id:40s} {agent_status_summary(entry)}")
-    print()
-
-
-# ---------------------------------------------------------------------------
-# Scenario 2: monitor — poll for state changes
-# ---------------------------------------------------------------------------
-
-
-def monitor_team_status(client: GraphClient, agents: List[str], cycles: int = 5):
-    """Poll presence for a team and report state changes."""
-    print(f"=== Monitoring {len(agents)} agents (poll every {POLL_INTERVAL_SECONDS}s) ===")
+def find_available_agent(client: GraphClient, agents: list[str]) -> Optional[str]:
+    presences = client.communications.get_presences_by_user_id(agents).execute_query()
+    for e in presences:
+        if is_available(e):
+            return e.id
+    return None
 
-    previous: Dict[str, str] = {}
 
+def monitor_presence(client: GraphClient, agents: list[str], cycles: int = 3):
+    print(f"Monitoring {len(agents)} agents (poll every {POLL_INTERVAL}s)")
+    previous = {}
     for cycle in range(1, cycles + 1):
-        print(f"\n--- Cycle {cycle}/{cycles} [{datetime.now(timezone.utc).isoformat()}] ---")
-        presences = get_presences_for_team(client, agents)
-
-        for entry in presences:
-            user_id = entry.id
-            state = agent_status_summary(entry)
-            old_state = previous.get(user_id)
-
-            if old_state is None:
-                print(f"  INIT  {user_id:35s} {state}")
-            elif state != old_state:
-                print(f"  >>> {user_id:35s} {old_state} -> {state}")
-                if is_available(entry) and not is_available.__doc__:
-                    print(f"       Agent {user_id} is now available for routing")
-            else:
-                print(f"  .     {user_id:35s} {state}")
-
-            previous[user_id] = state
-
+        presences = client.communications.get_presences_by_user_id(agents).execute_query()
+        for e in presences:
+            state = agent_status_summary(e)
+            old = previous.get(e.id)
+            if old is None:
+                print(f"  INIT  {e.id}: {state}")
+            elif state != old:
+                print(f"  CHG   {e.id}: {old} -> {state}")
+            previous[e.id] = state
         if cycle < cycles:
-            sleep(POLL_INTERVAL_SECONDS)
-
-    print()
-
-
-# ---------------------------------------------------------------------------
-# Scenario 3: round-robin availability check for routing
-# ---------------------------------------------------------------------------
+            sleep(POLL_INTERVAL)
 
 
-def find_available_agent(client: GraphClient, agents: List[str]) -> Optional[str]:
-    """Return the first available agent (useful for automatic call routing)."""
-    presences = get_presences_for_team(client, agents)
-    for entry in presences:
-        if is_available(entry):
-            return entry.id
-    return None
-
-
-# ---------------------------------------------------------------------------
-# Scenario 4: set status message on shift start
-# ---------------------------------------------------------------------------
+def main():
+    client = GraphClient().with_device_flow(client_id="your_client_id")
+    agents = ["alice@company.com", "bob@company.com", "carol@company.com"]
 
+    presences = client.communications.get_presences_by_user_id(agents).execute_query()
+    print("Team presence:")
+    for e in presences:
+        print(f"  {e.id}: {agent_status_summary(e)}")
 
-def start_shift(client: GraphClient, user_id: str, agent_name: str):
-    """Begin shift: set available presence with a status message."""
-    expires = datetime.now(timezone.utc) + timedelta(hours=9)
-    client.communications.presences[user_id].set_user_preferred_presence(
-        availability="Available",
-        activity="Available",
-        expiration_duration="PT9H",
-    ).execute_query()
+    available = find_available_agent(client, agents)
+    print(f"\nRouting: {'call to ' + available if available else 'no agents — queue'}")
 
-    client.communications.presences[user_id].set_status_message(
-        message=f"{agent_name} is online — ready to take calls",
-        expiry=expires,
-    ).execute_query()
+    monitor_presence(client, agents, cycles=3)
 
+    take_agent_offline(client, agents[0])
+    print(f"\n{agents[0]} marked offline")
 
-# ---------------------------------------------------------------------------
-# Main — run all scenarios
-# ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    # Connect (choose one auth method)
-    #
-    # Option A: certificate
-    # client = GraphClient(tenant).with_client_certificate(client_id, thumbprint, key)
-    #
-    # Option B: client secret
-    # client = GraphClient(tenant).with_client_secret(client_id, client_secret)
-    #
-    # Option C: device code (interactive, works with MFA)
-    client = GraphClient().with_device_flow(client_id="your_client_id")
-    # (For brevity replace with real credentials above)
-
-    # ------------------------------------------------------------------
-    # Scenario 1 — read-only snapshot
-    # ------------------------------------------------------------------
-    snapshot_team_status(client, SUPPORTED_AGENTS)
-
-    # ------------------------------------------------------------------
-    # Scenario 2 — find someone to route a call to
-    # ------------------------------------------------------------------
-    available = find_available_agent(client, SUPPORTED_AGENTS)
-    if available:
-        print(f"Routing call to: {available}")
-    else:
-        print("No agents available — queuing call")
-
-    # ------------------------------------------------------------------
-    # Scenario 3 — monitor for state changes (run 3 cycles)
-    # ------------------------------------------------------------------
-    monitor_team_status(client, SUPPORTED_AGENTS, cycles=3)
-
-    # ------------------------------------------------------------------
-    # Scenario 4 — mark agent as off (end of shift)
-    # ------------------------------------------------------------------
-    take_agent_offline(client, SUPPORTED_AGENTS[0])
-    print(f"Agent {SUPPORTED_AGENTS[0]} marked as offline")
+    main()