Signals Pipeline Resources

AWS Accounts

Region: us-west-2

S3 Buckets

Raw Telemetry (Input)

Path pattern: raw/json_telemetry/year=YYYY/month=MM/day=DD/hour=HH/*.gz

Verified Parquet (Output)

Path pattern: telemetry_flat/type={lock,bridge,video_doorbell}/day=YYYY-MM-DD/hour=HH/*.parquet

SQS Queues

Telemetry Processing

Redrive Queue

Athena Configuration

Workgroups

Glue Databases

Glue Tables (telemetry-parser-db)

Key Commands

Check SQS Queue Status

# Dev
AWS_PROFILE=platform-dev aws sqs get-queue-attributes \
  --queue-url https://sqs.us-west-2.amazonaws.com/905418337205/telemetry-processing-redrive-queue \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

# Prod
AWS_PROFILE=platform-prod aws sqs get-queue-attributes \
  --queue-url https://sqs.us-west-2.amazonaws.com/891377356712/telemetry-processing-redrive-queue \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible

Redrive Telemetry

cd ~/workplace/platform-tools

# Redrive specific day
AWS_PROFILE=platform-dev go run ./cmd/telemetry-redrive \
  --env dev \
  --s3-prefix "s3://dev-signal-data-lake-raw/raw/json_telemetry/year=2025/month=12/day=01/" \
  --no-assume-role \
  --monitor --max-queue-visible 500

Delete Parquet Files for Redrive

# Delete a day's Parquet files before redrive
AWS_PROFILE=platform-dev aws s3 rm \
  s3://dev-verified-parquet-v2-df1e4600/telemetry_flat/type=lock/day=2025-12-01/ \
  --recursive

Verify Telemetry Data

cd ~/workplace/platform-tools

# Check for at-least-once duplicates
uv run python -m platform_tools.telemetry_verify \
  --env dev --date 2025-12-01 --table bolt_controller_v2 --check-duplicates -v

# Brute-force verification (dev only - small datasets)
uv run python -m platform_tools.telemetry_verify \
  --env dev --date 2025-12-01 --table bolt_controller_v2 --brute-force -v

Query Parquet with DuckDB

uv run python -c "
from platform_tools.telemetry_verify.duckdb_client import DuckDBClient
from platform_tools.telemetry_verify.config import ENV_CONFIG
import os

os.environ['AWS_PROFILE'] = 'platform-dev'
client = DuckDBClient(ENV_CONFIG['dev'])
result = client.query('''
    SELECT envelope__event_kind, count(*) as cnt
    FROM read_parquet('s3://dev-verified-parquet-v2-df1e4600/telemetry_flat/type=lock/day=2025-12-01/hour=*/*.parquet')
    GROUP BY 1
    ORDER BY 2 DESC
''')
for row in result:
    print(row)
"

Parse Raw Telemetry with bixby-rs

cd ~/workplace/bixby-rs
cargo run -p bixby-cli -- parse --hex '<telemetry_raw_hex>'

Lambda Functions

Related Repositories