dubey/datacat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add activity history, insights, and cat usage events

Browse Source 
- Add activity history collection for cat usage events
- Add insight metrics (total cycles, average cycles, cycles today)
- Add cat usage events with weight for LR4
- Add intermediate state events (cycle started, cycle complete, etc.)
- Add pet visits_24h metric from weight history
- Add new LR4 metrics: lifetime_cycles, dfi_trigger_count, cat_detected
- Add configurable activity lookback window to prevent duplicate events
- Add API call delays to avoid rate limiting
- Update default poll interval to 5 minutes
- Update documentation with new metrics and events
This commit is contained in:
Tanishq Dubey
2025-12-16 15:37:34 -05:00
parent bbce85fa31
commit f126202410
6 changed files with 421 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.env.example
View File

@@ -11,10 +11,16 @@ DATADOG_APP_KEY=your-datadog-app-key
DATADOG_SITE=datadoghq.com
# Collector settings
DATACAT_POLL_INTERVAL=120
DATACAT_POLL_INTERVAL=300
DATACAT_INCLUDE_PETS=true
DATACAT_EMIT_EVENTS=true
DATACAT_METRIC_PREFIX=litterrobot
# Activity and insights collection
DATACAT_COLLECT_ACTIVITY=true
DATACAT_COLLECT_INSIGHTS=true
DATACAT_API_DELAY=1.0
DATACAT_ACTIVITY_LOOKBACK=10
# Optional: Path to config file (if using YAML config instead of env vars)
# DATACAT_CONFIG_FILE=/path/to/config.yaml

97
README.md
View File

@@ -12,10 +12,12 @@ Datacat connects to the Whisker API (used by Litter-Robot and Feeder-Robot devic
## Features
- Collects metrics from all connected robots every 2 minutes (configurable)
- Collects metrics from all connected robots every 5 minutes (configurable)
- Fetches activity history to detect cat usage events
- Collects insight data (cycle counts, daily averages)
- Submits metrics to Datadog via DogStatsD
- Emits Datadog events for state changes (drawer full, offline, errors)
- Supports pet profile metrics (weight, health status)
- Emits Datadog events for state changes (cat usage, drawer full, offline, errors)
- Supports pet profile metrics (weight, health status, visit count)
- Dry-run mode for testing without submitting metrics
- Configuration via environment variables or YAML file
@@ -69,9 +71,13 @@ Datacat can be configured via environment variables, a YAML config file, or both
| `DATADOG_API_KEY` | Datadog API key | Yes |
| `DATADOG_APP_KEY` | Datadog application key | No |
| `DATADOG_SITE` | Datadog site (default: `datadoghq.com`) | No |
| `DATACAT_POLL_INTERVAL` | Polling interval in seconds (default: `120`) | No |
| `DATACAT_POLL_INTERVAL` | Polling interval in seconds (default: `300`) | No |
| `DATACAT_INCLUDE_PETS` | Include pet metrics (default: `true`) | No |
| `DATACAT_EMIT_EVENTS` | Emit Datadog events (default: `true`) | No |
| `DATACAT_COLLECT_ACTIVITY` | Fetch activity history (default: `true`) | No |
| `DATACAT_COLLECT_INSIGHTS` | Fetch insight data (default: `true`) | No |
| `DATACAT_API_DELAY` | Delay between API calls in seconds (default: `1.0`) | No |
| `DATACAT_ACTIVITY_LOOKBACK` | Only emit events for activities within N minutes (default: `10`) | No |
| `DATACAT_METRIC_PREFIX` | Metric prefix (default: `litterrobot`) | No |
| `DATACAT_CONFIG_FILE` | Path to YAML config file | No |
@@ -89,9 +95,13 @@ datadog:
metric_prefix: "litterrobot"
collector:
poll_interval_seconds: 120
poll_interval_seconds: 300
include_pets: true
emit_events: true
collect_activity_history: true
collect_insights: true
api_call_delay_seconds: 1.0
activity_lookback_minutes: 10
```
## Usage
@@ -130,16 +140,30 @@ datacat -v
| `litterrobot.panel_lock_enabled` | Gauge | Panel lock status |
| `litterrobot.power_status` | Gauge | 2=AC, 1=DC, 0=NC |
### Insight Metrics (from Activity History)
| Metric | Type | Description |
|--------|------|-------------|
| `litterrobot.insight.total_cycles` | Gauge | Total cycles in last 30 days |
| `litterrobot.insight.average_cycles` | Gauge | Average daily cycles |
| `litterrobot.insight.cycles_today` | Gauge | Cycles completed today |
### Litter Robot 4 Additional Metrics
| Metric | Type | Description |
|--------|------|-------------|
| `litterrobot.litter_level` | Gauge | Litter level (0-100%) |
| `litterrobot.pet_weight` | Gauge | Last recorded weight (lbs) |
| `litterrobot.scoops_saved` | Gauge | Environmental counter |
| `litterrobot.wifi_rssi` | Gauge | WiFi signal strength |
| `litterrobot.litter_level_calculated` | Gauge | Calculated litter level from ToF sensor |
| `litterrobot.pet_weight` | Gauge | Last recorded cat weight (lbs) from robot scale |
| `litterrobot.scoops_saved` | Gauge | Environmental savings counter |
| `litterrobot.wifi_rssi` | Gauge | WiFi signal strength (dBm) |
| `litterrobot.night_light_brightness` | Gauge | Brightness level |
| `litterrobot.hopper_enabled` | Gauge | Hopper status |
| `litterrobot.lifetime_cycles` | Count | Total lifetime clean cycles |
| `litterrobot.odometer_power_cycles` | Count | Total power cycles |
| `litterrobot.odometer_empty_cycles` | Count | Total empty cycles |
| `litterrobot.dfi_trigger_count` | Count | Drawer full indicator triggers |
| `litterrobot.cat_detected` | Gauge | 1 if cat currently detected |
### Feeder Robot Metrics
@@ -153,10 +177,13 @@ datacat -v
| Metric | Type | Description |
|--------|------|-------------|
| `litterrobot.pet.weight` | Gauge | Pet weight (lbs) |
| `litterrobot.pet.weight` | Gauge | Pet weight from profile (lbs) |
| `litterrobot.pet.estimated_weight` | Gauge | User-entered estimated weight |
| `litterrobot.pet.last_weight_reading` | Gauge | Last sensor weight reading |
| `litterrobot.pet.is_healthy` | Gauge | Health status |
| `litterrobot.pet.is_active` | Gauge | Active status |
| `litterrobot.pet.age` | Gauge | Pet age |
| `litterrobot.pet.visits_24h` | Gauge | Litter box visits in last 24 hours |
### Tags
@@ -168,14 +195,56 @@ All metrics include the following tags:
- `robot_model` - Robot model (Litter-Robot 3, Litter-Robot 4, Feeder-Robot)
- `status` - Current robot status (for litter robots)
Pet metrics include:
- `pet_id` - Unique pet identifier
- `pet_name` - Pet name
- `pet_type` - Pet type (cat/dog)
## Events
Datacat emits Datadog events for important state changes:
Datacat emits Datadog events for important state changes. Events are sourced from both real-time state changes and activity history polling.
- **Robot goes offline/online** - Warning/Success event
- **Waste drawer full** - Warning event
- **Robot errors** (sensor faults, pinch detect, etc.) - Error event
- **Food level low** (Feeder Robot, <20%) - Warning event
### Cat Usage Events
| Event | Alert Type | Description |
|-------|------------|-------------|
| Cat used litter box | Info | Cat detected and weighed (LR4 includes weight) |
| Cat detected | Info | Cat is using the litter box |
### Cycle Events
| Event | Alert Type | Description |
|-------|------------|-------------|
| Clean cycle started | Info | Cleaning cycle has begun |
| Clean cycle complete | Info | Cleaning cycle finished |
| Litter dispensed | Info | Litter hopper dispensed (LR4 with hopper) |
### Warning Events
| Event | Alert Type | Description |
|-------|------------|-------------|
| Drawer full | Warning | Waste drawer needs emptying |
| Drawer almost full | Warning | 1-2 cycles remaining |
| Robot went offline | Warning | Lost connection to robot |
| Bonnet removed | Warning | Bonnet has been removed |
| On battery backup | Warning | Running on DC power |
| Pinch detected | Warning | Pinch was detected |
| Food level low | Warning | Feeder below 20% |
### Error Events
| Event | Alert Type | Description |
|-------|------------|-------------|
| Cat sensor fault | Error | Cat sensor has a fault |
| Over torque fault | Error | Motor over torque detected |
| Dump position fault | Error | Dump position fault |
| Home position fault | Error | Home position fault |
### Success Events
| Event | Alert Type | Description |
|-------|------------|-------------|
| Robot back online | Success | Robot reconnected |
## Development

6
config.example.yaml
View File

@@ -15,6 +15,10 @@ datadog:
# Collector settings
collector:
poll_interval_seconds: 120 # How often to collect metrics (default: 2 minutes)
poll_interval_seconds: 300 # How often to collect metrics (default: 5 minutes)
include_pets: true # Include pet profile metrics
emit_events: true # Emit Datadog events for state changes
collect_activity_history: true # Fetch activity history for cat usage events
collect_insights: true # Fetch insight data (cycle counts, averages)
api_call_delay_seconds: 1.0 # Delay between API calls to avoid rate limits
activity_lookback_minutes: 10 # Only emit events for activities within this window

310
src/datacat/collector.py
View File

@@ -1,10 +1,12 @@
"""Collector module for fetching data from Whisker/Litter Robot API."""
import asyncio
import logging
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from typing import Any
from pylitterbot import Account
from pylitterbot.enums import LitterBoxStatus
from pylitterbot.robot import Robot
from pylitterbot.robot.feederrobot import FeederRobot
from pylitterbot.robot.litterrobot import LitterRobot
@@ -22,6 +24,37 @@ from datacat.models import (
logger = logging.getLogger(__name__)
# Activity type mappings for LR4
LR4_ACTIVITY_MAP: dict[str, tuple[str, str, str]] = {
# value: (title_suffix, description, alert_type)
"catWeight": ("Cat used litter box", "Cat detected and weighed", "info"),
"robotCycleStatusDump": ("Clean cycle started", "Cleaning cycle has begun", "info"),
"robotCycleStatusIdle": ("Clean cycle complete", "Cleaning cycle finished", "info"),
"DFIFullFlagOn": ("Drawer full", "Waste drawer is full", "warning"),
"bonnetRemovedYes": ("Bonnet removed", "Bonnet has been removed", "warning"),
"catDetectStuckLaser": ("Cat sensor fault", "Cat sensor laser is stuck", "error"),
"litterHopperDispensed": ("Litter dispensed", "Litter hopper dispensed litter", "info"),
"powerTypeDC": ("On battery backup", "Robot switched to battery power", "warning"),
}
# LR3 status codes that indicate activity events
LR3_STATUS_EVENTS: dict[str, tuple[str, str, str]] = {
"CCP": ("Clean cycle started", "Cleaning cycle in progress", "info"),
"CCC": ("Clean cycle complete", "Cleaning cycle completed", "info"),
"CD": ("Cat detected", "Cat is using the litter box", "info"),
"CSI": ("Cat sensor interrupted", "Cat sensor was interrupted during cycle", "info"),
"DFS": ("Drawer full", "Waste drawer is full", "warning"),
"DF1": ("Drawer almost full", "Waste drawer has 2 cycles left", "warning"),
"DF2": ("Drawer almost full", "Waste drawer has 1 cycle left", "warning"),
"BR": ("Bonnet removed", "Bonnet has been removed", "warning"),
"PD": ("Pinch detected", "Pinch was detected", "warning"),
"CSF": ("Cat sensor fault", "Cat sensor has a fault", "error"),
"OTF": ("Over torque fault", "Motor over torque detected", "error"),
"DPF": ("Dump position fault", "Dump position fault detected", "error"),
"HPF": ("Home position fault", "Home position fault detected", "error"),
"DHF": ("Dump + home position fault", "Combined position fault", "error"),
}
class Collector:
"""Collects metrics from Whisker Litter Robot devices."""
@@ -31,6 +64,7 @@ class Collector:
self.config = config
self._account: Account | None = None
self._previous_states: dict[str, dict[str, Any]] = {}
self._last_activity_timestamps: dict[str, datetime] = {}
async def connect(self) -> None:
"""Connect to the Whisker API."""
@@ -73,7 +107,7 @@ class Collector:
# Collect robot metrics
for robot in self._account.robots:
try:
robot_metrics = self._collect_robot_metrics(robot, timestamp)
robot_metrics = await self._collect_robot_metrics(robot, timestamp)
result.robots.append(robot_metrics)
except Exception as e:
logger.error("Failed to collect metrics for robot %s: %s", robot.name, e)
@@ -83,7 +117,7 @@ class Collector:
if self.config.collector.include_pets:
for pet in self._account.pets:
try:
pet_metrics = self._collect_pet_metrics(pet, timestamp)
pet_metrics = await self._collect_pet_metrics(pet, timestamp)
result.pets.append(pet_metrics)
except Exception as e:
logger.error("Failed to collect metrics for pet %s: %s", pet.name, e)
@@ -100,7 +134,7 @@ class Collector:
f"robot_model:{robot.model}",
]
def _collect_robot_metrics(self, robot: Robot, timestamp: datetime) -> RobotMetrics:
async def _collect_robot_metrics(self, robot: Robot, timestamp: datetime) -> RobotMetrics:
"""Collect metrics from a single robot."""
prefix = self.config.datadog.metric_prefix
base_tags = self._get_base_tags(robot)
@@ -149,6 +183,28 @@ class Collector:
metrics.extend(self._collect_litter_robot_metrics(robot, prefix, base_tags, timestamp))
events.extend(self._check_litter_robot_events(robot, base_tags, timestamp))
# Collect activity history (with delay)
if self.config.collector.collect_activity_history:
await asyncio.sleep(self.config.collector.api_call_delay_seconds)
try:
activity_events = await self._collect_activity_history(
robot, base_tags, timestamp
)
events.extend(activity_events)
except Exception as e:
logger.warning("Failed to collect activity history for %s: %s", robot.name, e)
# Collect insight metrics (with delay)
if self.config.collector.collect_insights:
await asyncio.sleep(self.config.collector.api_call_delay_seconds)
try:
insight_metrics = await self._collect_insight_metrics(
robot, prefix, base_tags, timestamp
)
metrics.extend(insight_metrics)
except Exception as e:
logger.warning("Failed to collect insights for %s: %s", robot.name, e)
# Litter Robot 4 specific metrics
if isinstance(robot, LitterRobot4):
metrics.extend(self._collect_lr4_metrics(robot, prefix, base_tags, timestamp))
@@ -174,6 +230,160 @@ class Collector:
events=events,
)
async def _collect_activity_history(
self, robot: LitterRobot, base_tags: list[str], timestamp: datetime
) -> list[Event]:
"""Fetch activity history and emit events for new activities."""
if not self.config.collector.emit_events:
return []
events: list[Event] = []
# Calculate the lookback cutoff time
lookback_minutes = self.config.collector.activity_lookback_minutes
cutoff_time = timestamp - timedelta(minutes=lookback_minutes)
# Get last seen timestamp for this robot, default to cutoff
last_seen = self._last_activity_timestamps.get(robot.id, cutoff_time)
# Use the more recent of last_seen or cutoff_time
effective_cutoff = max(last_seen, cutoff_time)
logger.debug(
"Fetching activity history for %s (cutoff: %s)", robot.name, effective_cutoff
)
try:
activities = await robot.get_activity_history(limit=50)
except Exception as e:
logger.warning("Failed to get activity history for %s: %s", robot.name, e)
return []
# Filter to activities since effective cutoff (only datetime, not date)
new_activities = [
a for a in activities
if isinstance(a.timestamp, datetime) and a.timestamp > effective_cutoff
]
logger.debug("Found %d new activities for %s", len(new_activities), robot.name)
for activity in new_activities:
event = self._activity_to_event(robot, activity, base_tags)
if event:
events.append(event)
# Update last seen timestamp
if activities:
# Filter to only datetime timestamps (not date)
datetime_timestamps = [
a.timestamp for a in activities if isinstance(a.timestamp, datetime)
]
if datetime_timestamps:
newest_timestamp = max(datetime_timestamps)
self._last_activity_timestamps[robot.id] = newest_timestamp
return events
def _activity_to_event(
self, robot: LitterRobot, activity: Any, base_tags: list[str]
) -> Event | None:
"""Convert an activity record to a Datadog event."""
action = activity.action
activity_timestamp = activity.timestamp
# Handle LitterBoxStatus enum
if isinstance(action, LitterBoxStatus):
status_code = action.value
if status_code in LR3_STATUS_EVENTS:
title_suffix, description, alert_type = LR3_STATUS_EVENTS[status_code]
return Event(
title=f"{robot.name}: {title_suffix}",
text=f"{description} on {robot.name} ({robot.serial})",
tags=base_tags + [f"activity:{status_code}", f"status:{status_code}"],
alert_type=alert_type,
timestamp=activity_timestamp,
)
return None
# Handle string actions (LR4 activity format)
if isinstance(action, str):
# Check for known activity types
for key, (title_suffix, description, alert_type) in LR4_ACTIVITY_MAP.items():
if key in action:
text = f"{description} on {robot.name} ({robot.serial})"
# Extract weight if this is a cat weight event
if key == "catWeight" and ":" in action:
# Format is typically "Pet Weight Recorded: X.X lbs"
try:
weight_str = action.split(":")[-1].strip().replace(" lbs", "")
weight = float(weight_str)
text = f"Cat used litter box - weight: {weight} lbs"
except (ValueError, IndexError):
pass
return Event(
title=f"{robot.name}: {title_suffix}",
text=text,
tags=base_tags + [f"activity:{key}"],
alert_type=alert_type,
timestamp=activity_timestamp,
)
# Log unknown activity types for debugging
logger.debug("Unknown activity type for %s: %s", robot.name, action)
return None
async def _collect_insight_metrics(
self, robot: LitterRobot, prefix: str, base_tags: list[str], timestamp: datetime
) -> list[Metric]:
"""Fetch insight data and return metrics."""
metrics: list[Metric] = []
try:
insight = await robot.get_insight(days=30)
except Exception as e:
logger.warning("Failed to get insight data for %s: %s", robot.name, e)
return []
metrics.append(
Metric(
name=f"{prefix}.insight.total_cycles",
value=float(insight.total_cycles),
tags=base_tags,
timestamp=timestamp,
)
)
metrics.append(
Metric(
name=f"{prefix}.insight.average_cycles",
value=float(insight.average_cycles),
tags=base_tags,
timestamp=timestamp,
)
)
# Get today's cycle count from history
today = timestamp.date()
cycles_today = 0
for cycle_date, count in insight.cycle_history:
if cycle_date == today:
cycles_today = count
break
metrics.append(
Metric(
name=f"{prefix}.insight.cycles_today",
value=float(cycles_today),
tags=base_tags,
timestamp=timestamp,
)
)
return metrics
def _collect_litter_robot_metrics(
self, robot: LitterRobot, prefix: str, base_tags: list[str], timestamp: datetime
) -> list[Metric]:
@@ -326,8 +536,10 @@ class Collector:
)
)
# Raw data metrics if available
# Raw data metrics
raw_data = robot.to_dict()
# WiFi signal strength
if "wifiRssi" in raw_data:
metrics.append(
Metric(
@@ -337,6 +549,19 @@ class Collector:
timestamp=timestamp,
)
)
# Lifetime odometer metrics
if "odometerCleanCycles" in raw_data:
metrics.append(
Metric(
name=f"{prefix}.lifetime_cycles",
value=float(raw_data.get("odometerCleanCycles", 0)),
tags=base_tags,
timestamp=timestamp,
metric_type=MetricType.COUNT,
)
)
if "odometerPowerCycles" in raw_data:
metrics.append(
Metric(
@@ -347,6 +572,7 @@ class Collector:
metric_type=MetricType.COUNT,
)
)
if "odometerEmptyCycles" in raw_data:
metrics.append(
Metric(
@@ -358,6 +584,29 @@ class Collector:
)
)
# DFI trigger count
if "DFITriggerCount" in raw_data:
metrics.append(
Metric(
name=f"{prefix}.dfi_trigger_count",
value=float(raw_data.get("DFITriggerCount", 0)),
tags=base_tags,
timestamp=timestamp,
metric_type=MetricType.COUNT,
)
)
# Cat detection state
if "catDetect" in raw_data:
metrics.append(
Metric(
name=f"{prefix}.cat_detected",
value=1.0 if raw_data.get("catDetect") else 0.0,
tags=base_tags,
timestamp=timestamp,
)
)
return metrics
def _collect_feeder_metrics(
@@ -407,7 +656,7 @@ class Collector:
if robot.is_online:
events.append(
Event(
title=f"{robot.name} is back online",
title=f"{robot.name}: Back online",
text=f"Robot {robot.name} ({robot.serial}) has come back online.",
tags=base_tags,
alert_type="success",
@@ -417,7 +666,7 @@ class Collector:
else:
events.append(
Event(
title=f"{robot.name} went offline",
title=f"{robot.name}: Went offline",
text=f"Robot {robot.name} ({robot.serial}) has gone offline.",
tags=base_tags,
alert_type="warning",
@@ -442,7 +691,7 @@ class Collector:
if prev_drawer_full is not None and not prev_drawer_full and robot.is_waste_drawer_full:
events.append(
Event(
title=f"{robot.name} waste drawer is full",
title=f"{robot.name}: Drawer full",
text=f"The waste drawer on {robot.name} ({robot.serial}) needs to be emptied. "
f"Cycle count: {robot.cycle_count}",
tags=base_tags + [f"status:{robot.status.value}"],
@@ -451,33 +700,6 @@ class Collector:
)
)
# Status change events for certain error states
prev_status = prev_state.get("status")
current_status = robot.status.value if hasattr(robot.status, "value") else str(robot.status)
error_statuses = {
"CSF": "Cat Sensor Fault",
"DHF": "Dump + Home Position Fault",
"DPF": "Dump Position Fault",
"HPF": "Home Position Fault",
"OTF": "Over Torque Fault",
"PD": "Pinch Detect",
"SCF": "Cat Sensor Fault At Startup",
"SPF": "Pinch Detect At Startup",
}
if prev_status != current_status and current_status in error_statuses:
events.append(
Event(
title=f"{robot.name} error: {error_statuses[current_status]}",
text=f"Robot {robot.name} ({robot.serial}) has encountered an error: "
f"{error_statuses[current_status]} (status code: {current_status})",
tags=base_tags + [f"status:{current_status}"],
alert_type="error",
timestamp=timestamp,
)
)
return events
def _check_feeder_events(
@@ -497,7 +719,7 @@ class Collector:
if prev_food_level >= 20 and robot.food_level < 20:
events.append(
Event(
title=f"{robot.name} food level low",
title=f"{robot.name}: Food level low",
text=f"Food level on {robot.name} ({robot.serial}) is "
f"low: {robot.food_level}%",
tags=base_tags,
@@ -524,7 +746,7 @@ class Collector:
self._previous_states[robot.id] = state
def _collect_pet_metrics(self, pet, timestamp: datetime) -> PetMetrics:
async def _collect_pet_metrics(self, pet: Any, timestamp: datetime) -> PetMetrics:
"""Collect metrics from a pet profile."""
prefix = self.config.datadog.metric_prefix
pet_type_str = str(pet.pet_type) if pet.pet_type else "unknown"
@@ -596,6 +818,20 @@ class Collector:
)
)
# Visit count in last 24 hours
try:
visits_24h = pet.get_visits_since(timestamp - timedelta(hours=24))
metrics.append(
Metric(
name=f"{prefix}.pet.visits_24h",
value=float(visits_24h),
tags=base_tags,
timestamp=timestamp,
)
)
except Exception as e:
logger.debug("Could not get visit count for pet %s: %s", pet.name, e)
return PetMetrics(
pet_id=pet.id,
pet_name=pet.name,

35
src/datacat/config.py
View File

@@ -33,9 +33,13 @@ class DatadogConfig:
class CollectorConfig:
"""Collector behavior configuration."""
poll_interval_seconds: int = 120
poll_interval_seconds: int = 300 # 5 minutes default
include_pets: bool = True
emit_events: bool = True
collect_activity_history: bool = True
collect_insights: bool = True
api_call_delay_seconds: float = 1.0 # Delay between API calls to avoid rate limits
activity_lookback_minutes: int = 10 # Only emit events for activities within this window
@dataclass
@@ -58,9 +62,13 @@ def load_config(config_path: str | Path | None = None) -> Config:
DATADOG_API_KEY: Datadog API key
DATADOG_APP_KEY: Datadog application key (optional)
DATADOG_SITE: Datadog site (default: datadoghq.com)
DATACAT_POLL_INTERVAL: Polling interval in seconds (default: 120)
DATACAT_POLL_INTERVAL: Polling interval in seconds (default: 300)
DATACAT_INCLUDE_PETS: Include pet metrics (default: true)
DATACAT_EMIT_EVENTS: Emit Datadog events (default: true)
DATACAT_COLLECT_ACTIVITY: Collect activity history (default: true)
DATACAT_COLLECT_INSIGHTS: Collect insight metrics (default: true)
DATACAT_API_DELAY: Delay between API calls in seconds (default: 1.0)
DATACAT_ACTIVITY_LOOKBACK: Only emit events for activities within N minutes (default: 10)
DATACAT_CONFIG_FILE: Path to config file (optional)
"""
# Load .env file if present
@@ -117,6 +125,15 @@ def load_config(config_path: str | Path | None = None) -> Config:
return int(value)
return default
def get_float(env_var: str, file_path: list[str], default: float) -> float:
"""Get float config value."""
value = get_value(env_var, file_path, default)
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
return float(value)
return default
# Build configuration
whisker_username = get_value("WHISKER_USERNAME", ["whisker", "username"])
whisker_password = get_value("WHISKER_PASSWORD", ["whisker", "password"])
@@ -149,9 +166,21 @@ def load_config(config_path: str | Path | None = None) -> Config:
),
collector=CollectorConfig(
poll_interval_seconds=get_int(
"DATACAT_POLL_INTERVAL", ["collector", "poll_interval_seconds"], 120
"DATACAT_POLL_INTERVAL", ["collector", "poll_interval_seconds"], 300
),
include_pets=get_bool("DATACAT_INCLUDE_PETS", ["collector", "include_pets"], True),
emit_events=get_bool("DATACAT_EMIT_EVENTS", ["collector", "emit_events"], True),
collect_activity_history=get_bool(
"DATACAT_COLLECT_ACTIVITY", ["collector", "collect_activity_history"], True
),
collect_insights=get_bool(
"DATACAT_COLLECT_INSIGHTS", ["collector", "collect_insights"], True
),
api_call_delay_seconds=get_float(
"DATACAT_API_DELAY", ["collector", "api_call_delay_seconds"], 1.0
),
activity_lookback_minutes=get_int(
"DATACAT_ACTIVITY_LOOKBACK", ["collector", "activity_lookback_minutes"], 10
),
),
)

22
src/datacat/models.py
View File

@@ -1,7 +1,7 @@
"""Data models for metrics collection."""
from dataclasses import dataclass, field
from datetime import datetime
from datetime import date, datetime
from enum import Enum
@@ -13,6 +13,26 @@ class MetricType(Enum):
RATE = "rate"
@dataclass
class ActivityRecord:
"""A single activity from the robot's history."""
timestamp: datetime
action: str
robot_id: str
robot_name: str
weight: float | None = None # For cat weight events (LR4)
@dataclass
class InsightData:
"""Daily insight data for a robot."""
total_cycles: int
average_cycles: float
cycle_history: list[tuple[date, int]] = field(default_factory=list)
@dataclass
class Metric:
"""A single metric to be submitted to Datadog."""