Implementation Guide: Analyze service bay utilization, technician efficiency, and parts margin by job type

speedtest-cli --simple
nmap -sn 192.168.1.0/24
# Document results in site survey template

# Test each cable run before mounting cameras:
# On laptop connected to switch port:
ping -c 4 <gateway-ip>
# Verify PoE delivery on each port via UniFi Controller:
# Network > Devices > USW-Pro-24-PoE > Ports tab
# Each camera port should show 'PoE: Delivering' after camera connection

# Configure firewall rules (on UDM-Pro or existing firewall):
# ALLOW: VLAN 20 -> Internet (for camera firmware updates, PitCrew cloud)
# ALLOW: VLAN 20 -> VLAN 10 (specific IP of edge server analytics port only)
# DENY: VLAN 20 -> VLAN 10 (all other)
# DENY: VLAN 20 -> VLAN 30 (complete isolation from POS)
# DENY: VLAN 30 -> VLAN 20 (complete isolation)
# ALLOW: MSP VPN -> VLAN 20 (remote management)

# Verify RTSP feed externally:
ffplay rtsp://<UNVR-IP>:7447/<camera-stream-id>
# Or use VLC:
vlc rtsp://<UNVR-IP>:7447/<camera-stream-id>

# Install Ubuntu Server 22.04 LTS on Dell PowerEdge T160
# After OS installation and initial setup:
sudo apt update && sudo apt upgrade -y
sudo apt install -y curl git htop net-tools ufw

# Install Docker and Docker Compose:
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
sudo usermod -aG docker $USER
sudo apt install -y docker-compose-plugin

# Create project directory:
sudo mkdir -p /opt/shop-analytics
cd /opt/shop-analytics

# Create docker-compose.yml:
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  postgres:
    image: postgres:16
    container_name: shop-analytics-db
    environment:
      POSTGRES_DB: shop_analytics
      POSTGRES_USER: analytics_admin
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init-scripts:/docker-entrypoint-initdb.d
    ports:
      - '5432:5432'
    restart: unless-stopped

  n8n:
    image: n8nio/n8n:latest
    container_name: shop-analytics-n8n
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=analytics_admin
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
    volumes:
      - n8ndata:/home/node/.n8n
    ports:
      - '5678:5678'
    depends_on:
      - postgres
    restart: unless-stopped

volumes:
  pgdata:
  n8ndata:
EOF

# Create environment file:
cat > .env << 'EOF'
DB_PASSWORD=<GENERATE_STRONG_PASSWORD_HERE>
N8N_USER=mspadmin
N8N_PASSWORD=<GENERATE_STRONG_PASSWORD_HERE>
EOF
chmod 600 .env

# Start services:
docker compose up -d

# Verify services running:
docker compose ps
docker compose logs --tail=20 postgres
docker compose logs --tail=20 n8n

# Create the initialization SQL script:
cat > /opt/shop-analytics/init-scripts/001-create-schema.sql << 'EOSQL'
-- Automotive Service Analytics Schema
-- Version 1.0

CREATE SCHEMA IF NOT EXISTS analytics;

-- Dimension: Service Bays
CREATE TABLE analytics.dim_bays (
    bay_id SERIAL PRIMARY KEY,
    bay_number VARCHAR(10) NOT NULL,
    bay_type VARCHAR(50), -- 'general', 'alignment', 'quick_lube', 'heavy_repair', 'diagnostic'
    camera_id VARCHAR(100), -- RTSP stream identifier or PitCrew bay ID
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Dimension: Technicians
CREATE TABLE analytics.dim_technicians (
    tech_id SERIAL PRIMARY KEY,
    sms_tech_id VARCHAR(50) NOT NULL, -- ID from shop management system
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    skill_level VARCHAR(20), -- 'A', 'B', 'C', 'apprentice'
    hourly_cost_rate DECIMAL(8,2), -- what shop pays the tech
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Dimension: Job Types
CREATE TABLE analytics.dim_job_types (
    job_type_id SERIAL PRIMARY KEY,
    job_type_code VARCHAR(20) NOT NULL,
    job_type_name VARCHAR(200) NOT NULL,
    category VARCHAR(100), -- 'maintenance', 'repair', 'diagnostic', 'tire', 'body'
    avg_book_hours DECIMAL(6,2),
    target_parts_margin_pct DECIMAL(5,2),
    created_at TIMESTAMP DEFAULT NOW()
);

-- Dimension: Date (for time-series analysis)
CREATE TABLE analytics.dim_date (
    date_key DATE PRIMARY KEY,
    day_of_week INTEGER,
    day_name VARCHAR(10),
    week_number INTEGER,
    month_number INTEGER,
    month_name VARCHAR(10),
    quarter INTEGER,
    year INTEGER,
    is_weekend BOOLEAN,
    is_holiday BOOLEAN DEFAULT false
);

-- Fact: Repair Orders
CREATE TABLE analytics.fact_repair_orders (
    ro_id SERIAL PRIMARY KEY,
    sms_ro_number VARCHAR(50) NOT NULL,
    date_key DATE REFERENCES analytics.dim_date(date_key),
    customer_id VARCHAR(50),
    vehicle_vin VARCHAR(17),
    vehicle_year INTEGER,
    vehicle_make VARCHAR(50),
    vehicle_model VARCHAR(50),
    total_labor_sale DECIMAL(10,2),
    total_parts_sale DECIMAL(10,2),
    total_parts_cost DECIMAL(10,2),
    total_ro_sale DECIMAL(10,2),
    ro_status VARCHAR(20), -- 'open', 'in_progress', 'completed', 'invoiced'
    created_at_sms TIMESTAMP,
    completed_at_sms TIMESTAMP,
    loaded_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(sms_ro_number)
);

-- Fact: Labor Lines (one per job per tech per RO)
CREATE TABLE analytics.fact_labor_lines (
    labor_line_id SERIAL PRIMARY KEY,
    ro_id INTEGER REFERENCES analytics.fact_repair_orders(ro_id),
    tech_id INTEGER REFERENCES analytics.dim_technicians(tech_id),
    job_type_id INTEGER REFERENCES analytics.dim_job_types(job_type_id),
    date_key DATE REFERENCES analytics.dim_date(date_key),
    description TEXT,
    book_hours DECIMAL(6,2), -- standard/book time for the job
    actual_hours DECIMAL(6,2), -- clock time the tech spent
    billed_hours DECIMAL(6,2), -- hours billed to customer
    labor_rate DECIMAL(8,2), -- $/hour charged
    labor_sale DECIMAL(10,2), -- billed_hours * labor_rate
    clock_in TIMESTAMP,
    clock_out TIMESTAMP,
    loaded_at TIMESTAMP DEFAULT NOW()
);

-- Fact: Parts Lines (one per part per RO)
CREATE TABLE analytics.fact_parts_lines (
    parts_line_id SERIAL PRIMARY KEY,
    ro_id INTEGER REFERENCES analytics.fact_repair_orders(ro_id),
    job_type_id INTEGER REFERENCES analytics.dim_job_types(job_type_id),
    date_key DATE REFERENCES analytics.dim_date(date_key),
    part_number VARCHAR(50),
    part_description TEXT,
    vendor VARCHAR(100),
    quantity DECIMAL(8,2),
    cost_per_unit DECIMAL(10,2),
    sale_per_unit DECIMAL(10,2),
    total_cost DECIMAL(10,2),
    total_sale DECIMAL(10,2),
    margin_amount DECIMAL(10,2) GENERATED ALWAYS AS (total_sale - total_cost) STORED,
    margin_pct DECIMAL(5,2) GENERATED ALWAYS AS (
        CASE WHEN total_sale > 0 THEN ((total_sale - total_cost) / total_sale * 100) ELSE 0 END
    ) STORED,
    loaded_at TIMESTAMP DEFAULT NOW()
);

-- Fact: Bay Events (from PitCrew or camera AI)
CREATE TABLE analytics.fact_bay_events (
    event_id SERIAL PRIMARY KEY,
    bay_id INTEGER REFERENCES analytics.dim_bays(bay_id),
    date_key DATE REFERENCES analytics.dim_date(date_key),
    event_type VARCHAR(20) NOT NULL, -- 'vehicle_in', 'vehicle_out', 'idle_start', 'idle_end'
    event_timestamp TIMESTAMP NOT NULL,
    vehicle_detected BOOLEAN,
    tech_detected BOOLEAN,
    ro_id INTEGER REFERENCES analytics.fact_repair_orders(ro_id),
    source VARCHAR(50), -- 'pitcrew', 'custom_ai', 'manual'
    loaded_at TIMESTAMP DEFAULT NOW()
);

-- Aggregation: Daily Bay Utilization Summary
CREATE TABLE analytics.agg_daily_bay_utilization (
    id SERIAL PRIMARY KEY,
    bay_id INTEGER REFERENCES analytics.dim_bays(bay_id),
    date_key DATE REFERENCES analytics.dim_date(date_key),
    total_occupied_minutes INTEGER,
    total_idle_minutes INTEGER,
    total_available_minutes INTEGER, -- shop hours that day
    utilization_pct DECIMAL(5,2),
    vehicle_count INTEGER,
    avg_dwell_minutes INTEGER,
    loaded_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(bay_id, date_key)
);

-- Populate date dimension for 3 years
INSERT INTO analytics.dim_date
SELECT
    d::date AS date_key,
    EXTRACT(DOW FROM d) AS day_of_week,
    TO_CHAR(d, 'Day') AS day_name,
    EXTRACT(WEEK FROM d) AS week_number,
    EXTRACT(MONTH FROM d) AS month_number,
    TO_CHAR(d, 'Month') AS month_name,
    EXTRACT(QUARTER FROM d) AS quarter,
    EXTRACT(YEAR FROM d) AS year,
    EXTRACT(DOW FROM d) IN (0,6) AS is_weekend,
    false AS is_holiday
FROM generate_series('2024-01-01'::date, '2026-12-31'::date, '1 day'::interval) d;

-- Create indexes for query performance
CREATE INDEX idx_fact_ro_date ON analytics.fact_repair_orders(date_key);
CREATE INDEX idx_fact_labor_date ON analytics.fact_labor_lines(date_key);
CREATE INDEX idx_fact_labor_tech ON analytics.fact_labor_lines(tech_id);
CREATE INDEX idx_fact_labor_job ON analytics.fact_labor_lines(job_type_id);
CREATE INDEX idx_fact_parts_date ON analytics.fact_parts_lines(date_key);
CREATE INDEX idx_fact_parts_job ON analytics.fact_parts_lines(job_type_id);
CREATE INDEX idx_fact_bay_events_bay ON analytics.fact_bay_events(bay_id, date_key);
CREATE INDEX idx_agg_bay_util ON analytics.agg_daily_bay_utilization(bay_id, date_key);
EOSQL

# If database is already running, apply manually:
docker exec -i shop-analytics-db psql -U analytics_admin -d shop_analytics < /opt/shop-analytics/init-scripts/001-create-schema.sql

# Test Tekmetric API access:
curl -H 'Authorization: Bearer <TEKMETRIC_API_KEY>' \
  'https://sandbox.tekmetric.com/api/v1/shops/<SHOP_ID>/repair-orders?page=0&size=10'

# Expected response: JSON with repair order data including
# labor lines, parts lines, technician assignments, and status

# For Shopmonkey:
curl -H 'Authorization: Bearer <SHOPMONKEY_API_KEY>' \
  'https://api.shopmonkey.cloud/v3/work_orders?limit=10'

curl -H 'Authorization: Bearer <QB_ACCESS_TOKEN>' \
  -H 'Accept: application/json' \
  'https://quickbooks.api.intuit.com/v3/company/<COMPANY_ID>/query?query=SELECT * FROM Purchase WHERE TxnDate >= 2024-01-01 MAXRESULTS 5'

# expected response: JSON with current status of each bay

curl -H 'Authorization: Bearer <PITCREW_API_KEY>' \
  'https://api.leverege.com/pitcrew/v1/locations/<LOCATION_ID>/bays/status'

{ "bays": [{ "id": "bay-1", "status": "occupied", "since": "2024-...", "vehicle_detected": true }] }

docker exec -i shop-analytics-db psql -U analytics_admin -d shop_analytics << 'EOF'
CREATE USER analytics_readonly WITH PASSWORD '<READONLY_PASSWORD>';
GRANT USAGE ON SCHEMA analytics TO analytics_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics TO analytics_readonly;
ALTER DEFAULT PRIVILEGES IN SCHEMA analytics GRANT SELECT ON TABLES TO analytics_readonly;
EOF

docker exec -i shop-analytics-db psql -U analytics_admin -d shop_analytics << 'EOF'
SELECT COUNT(*) as ro_count FROM analytics.fact_repair_orders;
SELECT COUNT(*) as labor_count FROM analytics.fact_labor_lines;
SELECT COUNT(*) as parts_count FROM analytics.fact_parts_lines;
SELECT COUNT(*) as event_count FROM analytics.fact_bay_events;
EOF

# Enable UFW firewall on edge server:
sudo ufw default deny incoming
sudo ufw default allow outgoing
sudo ufw allow from <MSP_VPN_SUBNET> to any port 22 proto tcp comment 'SSH via VPN'
sudo ufw allow from <MSP_VPN_SUBNET> to any port 5678 proto tcp comment 'n8n via VPN'
sudo ufw allow from <POWER_BI_GATEWAY_IP> to any port 5432 proto tcp comment 'Power BI DB'
sudo ufw allow from <PITCREW_WEBHOOK_IPS> to any port 5678 proto tcp comment 'PitCrew webhooks'
sudo ufw enable

# Enable PostgreSQL SSL:
# Edit postgresql.conf in Docker volume:
docker exec -i shop-analytics-db bash -c "cat >> /var/lib/postgresql/data/postgresql.conf << 'EOF'
ssl = on
ssl_cert_file = '/var/lib/postgresql/data/server.crt'
ssl_key_file = '/var/lib/postgresql/data/server.key'
log_connections = on
log_disconnections = on
EOF"

# Generate self-signed cert (or use Let's Encrypt):
docker exec -i shop-analytics-db bash -c \
  'openssl req -new -x509 -days 365 -nodes -text \
   -out /var/lib/postgresql/data/server.crt \
   -keyout /var/lib/postgresql/data/server.key \
   -subj "/CN=shop-analytics-db" && \
   chmod 600 /var/lib/postgresql/data/server.key && \
   chown postgres:postgres /var/lib/postgresql/data/server.*'

docker compose restart postgres

# Configure automated security updates:
sudo apt install -y unattended-upgrades
sudo dpkg-reconfigure -plow unattended-upgrades

# Set up log forwarding (if MSP uses SIEM):
# Install rsyslog or configure Docker logging driver to forward to SIEM

# Run historical backfill (adjust date range as needed):
# In n8n, open 'SMS Historical Backfill' workflow
# Set parameters: start_date=2024-01-01, end_date=<today>
# Execute workflow and monitor progress

# Validate data counts against SMS:
docker exec -i shop-analytics-db psql -U analytics_admin -d shop_analytics << 'EOF'
-- Compare these numbers against SMS reports:
SELECT 
  COUNT(*) as total_ros,
  SUM(total_labor_sale) as total_labor_revenue,
  SUM(total_parts_sale) as total_parts_revenue,
  SUM(total_ro_sale) as total_revenue
FROM analytics.fact_repair_orders
WHERE date_key >= '2024-01-01' AND date_key < '2025-01-01';

-- Check technician efficiency data:
SELECT 
  t.first_name || ' ' || t.last_name as technician,
  COUNT(*) as job_count,
  ROUND(AVG(l.billed_hours / NULLIF(l.actual_hours, 0) * 100), 1) as avg_efficiency_pct,
  ROUND(SUM(l.billed_hours), 1) as total_billed,
  ROUND(SUM(l.actual_hours), 1) as total_actual
FROM analytics.fact_labor_lines l
JOIN analytics.dim_technicians t ON l.tech_id = t.tech_id
WHERE l.date_key >= '2024-01-01'
GROUP BY t.first_name, t.last_name
ORDER BY avg_efficiency_pct DESC;

-- Check parts margin by job type:
SELECT 
  jt.job_type_name,
  jt.category,
  COUNT(*) as parts_line_count,
  ROUND(SUM(p.total_sale), 2) as total_parts_sale,
  ROUND(SUM(p.total_cost), 2) as total_parts_cost,
  ROUND(AVG(p.margin_pct), 1) as avg_margin_pct
FROM analytics.fact_parts_lines p
JOIN analytics.dim_job_types jt ON p.job_type_id = jt.job_type_id
WHERE p.date_key >= '2024-01-01'
GROUP BY jt.job_type_name, jt.category
ORDER BY avg_margin_pct ASC;
EOF

# Compare Power BI dashboard totals against these SQL results
# Differences should be < 1% (rounding)

# Bay Utilization, Technician Efficiency, Parts Margin, and Composite KPIs

-- POWER BI DAX MEASURES --
-- Add these in Power BI Desktop > Modeling > New Measure

-- ===== BAY UTILIZATION MEASURES =====

Bay Utilization % = 
VAR _OccupiedMinutes = SUM(agg_daily_bay_utilization[total_occupied_minutes])
VAR _AvailableMinutes = SUM(agg_daily_bay_utilization[total_available_minutes])
RETURN
    IF(_AvailableMinutes > 0, DIVIDE(_OccupiedMinutes, _AvailableMinutes, 0), BLANK())

Avg Vehicles Per Bay Per Day = 
AVERAGE(agg_daily_bay_utilization[vehicle_count])

Avg Dwell Time (Minutes) = 
AVERAGE(agg_daily_bay_utilization[avg_dwell_minutes])

Idle Time % = 
VAR _IdleMinutes = SUM(agg_daily_bay_utilization[total_idle_minutes])
VAR _AvailableMinutes = SUM(agg_daily_bay_utilization[total_available_minutes])
RETURN
    IF(_AvailableMinutes > 0, DIVIDE(_IdleMinutes, _AvailableMinutes, 0), BLANK())

Bay Utilization Target = 0.85 -- 85% industry benchmark

Bay Utilization vs Target = [Bay Utilization %] - [Bay Utilization Target]

Total Shop Available Hours = 
VAR _TotalMinutes = SUM(agg_daily_bay_utilization[total_available_minutes])
RETURN DIVIDE(_TotalMinutes, 60, 0)

Lost Revenue from Idle Bays = 
VAR _IdleHours = DIVIDE(SUM(agg_daily_bay_utilization[total_idle_minutes]), 60, 0)
VAR _AvgLaborRate = AVERAGE(fact_labor_lines[labor_rate])
VAR _AvgEfficiency = [Overall Tech Efficiency %]
RETURN _IdleHours * _AvgLaborRate * _AvgEfficiency

-- ===== TECHNICIAN EFFICIENCY MEASURES =====

Total Billed Hours = SUM(fact_labor_lines[billed_hours])

Total Actual Hours = SUM(fact_labor_lines[actual_hours])

Total Book Hours = SUM(fact_labor_lines[book_hours])

Overall Tech Efficiency % = 
DIVIDE([Total Billed Hours], [Total Actual Hours], 0)

Tech Productivity % = 
DIVIDE([Total Billed Hours], [Total Shop Available Hours], 0)

Tech Proficiency % = 
DIVIDE([Total Billed Hours], [Total Book Hours], 0)

Effective Labor Rate = 
DIVIDE(SUM(fact_labor_lines[labor_sale]), [Total Actual Hours], 0)

Posted Labor Rate = 
AVERAGE(fact_labor_lines[labor_rate])

Labor Rate Recovery % = 
DIVIDE([Effective Labor Rate], [Posted Labor Rate], 0)

Avg RO Labor Revenue = 
DIVIDE(SUM(fact_repair_orders[total_labor_sale]), DISTINCTCOUNT(fact_repair_orders[ro_id]), 0)

Tech Efficiency Target = 1.0 -- 100% = billed equals actual (break-even for flat rate)

Top Performer Threshold = 1.25 -- 125% efficiency = A-tech benchmark

Jobs Per Tech Per Day = 
VAR _JobCount = COUNTROWS(fact_labor_lines)
VAR _WorkDays = DISTINCTCOUNT(fact_labor_lines[date_key])
VAR _TechCount = DISTINCTCOUNT(fact_labor_lines[tech_id])
RETURN DIVIDE(_JobCount, _WorkDays * _TechCount, 0)

Comebacks % = 
VAR _TotalROs = DISTINCTCOUNT(fact_repair_orders[ro_id])
VAR _Comebacks = CALCULATE(
    DISTINCTCOUNT(fact_repair_orders[ro_id]),
    FILTER(fact_repair_orders, fact_repair_orders[ro_status] = "comeback")
)
RETURN DIVIDE(_Comebacks, _TotalROs, 0)

-- ===== PARTS MARGIN MEASURES =====

Total Parts Revenue = SUM(fact_parts_lines[total_sale])

Total Parts Cost = SUM(fact_parts_lines[total_cost])

Total Parts Gross Profit = [Total Parts Revenue] - [Total Parts Cost]

Parts Margin % = 
DIVIDE([Total Parts Gross Profit], [Total Parts Revenue], 0)

Parts Margin Target = 0.50 -- 50% GPM industry target for independents

Parts Margin vs Target = [Parts Margin %] - [Parts Margin Target]

Parts-to-Labor Ratio = 
DIVIDE(SUM(fact_repair_orders[total_parts_sale]), SUM(fact_repair_orders[total_labor_sale]), 0)

Avg RO Parts Revenue = 
DIVIDE(SUM(fact_repair_orders[total_parts_sale]), DISTINCTCOUNT(fact_repair_orders[ro_id]), 0)

Avg RO Value (ARO) = 
DIVIDE(SUM(fact_repair_orders[total_ro_sale]), DISTINCTCOUNT(fact_repair_orders[ro_id]), 0)

Low Margin Parts Lines Count = 
CALCULATE(
    COUNTROWS(fact_parts_lines),
    fact_parts_lines[margin_pct] < 35
)

Low Margin Parts Revenue = 
CALCULATE(
    SUM(fact_parts_lines[total_sale]),
    fact_parts_lines[margin_pct] < 35
)

Margin Leakage Amount = 
VAR _TargetMarginRate = 0.50
RETURN
    SUMX(
        fact_parts_lines,
        VAR _ActualMargin = fact_parts_lines[margin_amount]
        VAR _TargetMargin = fact_parts_lines[total_sale] * _TargetMarginRate
        RETURN IF(_TargetMargin > _ActualMargin, _TargetMargin - _ActualMargin, 0)
    )

-- ===== COMPOSITE / SUMMARY MEASURES =====

Total Revenue = SUM(fact_repair_orders[total_ro_sale])

Total Gross Profit = 
[Total Parts Gross Profit] + SUM(fact_labor_lines[labor_sale]) - 
    SUMX(fact_labor_lines, fact_labor_lines[actual_hours] * RELATED(dim_technicians[hourly_cost_rate]))

Gross Profit % = DIVIDE([Total Gross Profit], [Total Revenue], 0)

RO Count = DISTINCTCOUNT(fact_repair_orders[ro_id])

Car Count = DISTINCTCOUNT(fact_repair_orders[vehicle_vin])

{
  "name": "SMS Data Sync - Tekmetric",
  "nodes": [
    {
      "parameters": {
        "rule": {
          "interval": [{ "field": "hours", "hoursInterval": 2 }]
        }
      },
      "name": "Schedule Trigger",
      "type": "n8n-nodes-base.scheduleTrigger",
      "position": [250, 300]
    },
    {
      "parameters": {
        "url": "=https://sandbox.tekmetric.com/api/v1/shops/{{$env.TEKMETRIC_SHOP_ID}}/repair-orders",
        "authentication": "genericCredentialType",
        "genericAuthType": "httpHeaderAuth",
        "sendQuery": true,
        "queryParameters": {
          "parameters": [
            { "name": "updatedDateStart", "value": "={{$now.minus(3, 'hours').toISO()}}" },
            { "name": "size", "value": "100" },
            { "name": "page", "value": "0" }
          ]
        },
        "options": { "response": { "response": { "responseFormat": "json" } } }
      },
      "name": "Fetch Repair Orders",
      "type": "n8n-nodes-base.httpRequest",
      "position": [450, 300],
      "credentials": { "httpHeaderAuth": { "name": "Tekmetric API" } }
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.fact_repair_orders (sms_ro_number, date_key, customer_id, vehicle_vin, vehicle_year, vehicle_make, vehicle_model, total_labor_sale, total_parts_sale, total_parts_cost, total_ro_sale, ro_status, created_at_sms, completed_at_sms) VALUES ($1, $2::date, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13::timestamp, $14::timestamp) ON CONFLICT (sms_ro_number) DO UPDATE SET total_labor_sale = EXCLUDED.total_labor_sale, total_parts_sale = EXCLUDED.total_parts_sale, total_parts_cost = EXCLUDED.total_parts_cost, total_ro_sale = EXCLUDED.total_ro_sale, ro_status = EXCLUDED.ro_status, completed_at_sms = EXCLUDED.completed_at_sms RETURNING ro_id",
        "options": {}
      },
      "name": "Upsert Repair Orders",
      "type": "n8n-nodes-base.postgres",
      "position": [650, 300],
      "credentials": { "postgres": { "name": "Shop Analytics DB" } }
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.fact_labor_lines (ro_id, tech_id, job_type_id, date_key, description, book_hours, actual_hours, billed_hours, labor_rate, labor_sale, clock_in, clock_out) SELECT ro.ro_id, t.tech_id, COALESCE(jt.job_type_id, 1), ro.date_key, $1, $2, $3, $4, $5, $6, $7::timestamp, $8::timestamp FROM analytics.fact_repair_orders ro LEFT JOIN analytics.dim_technicians t ON t.sms_tech_id = $9 LEFT JOIN analytics.dim_job_types jt ON jt.job_type_code = $10 WHERE ro.sms_ro_number = $11 ON CONFLICT DO NOTHING",
        "options": {}
      },
      "name": "Insert Labor Lines",
      "type": "n8n-nodes-base.postgres",
      "position": [850, 200],
      "credentials": { "postgres": { "name": "Shop Analytics DB" } }
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.fact_parts_lines (ro_id, job_type_id, date_key, part_number, part_description, vendor, quantity, cost_per_unit, sale_per_unit, total_cost, total_sale) SELECT ro.ro_id, COALESCE(jt.job_type_id, 1), ro.date_key, $1, $2, $3, $4, $5, $6, $7, $8 FROM analytics.fact_repair_orders ro LEFT JOIN analytics.dim_job_types jt ON jt.job_type_code = $9 WHERE ro.sms_ro_number = $10 ON CONFLICT DO NOTHING",
        "options": {}
      },
      "name": "Insert Parts Lines",
      "type": "n8n-nodes-base.postgres",
      "position": [850, 400],
      "credentials": { "postgres": { "name": "Shop Analytics DB" } }
    },
    {
      "parameters": {
        "conditions": { "boolean": [{ "value1": "={{$json.error}}", "value2": true }] }
      },
      "name": "Error Check",
      "type": "n8n-nodes-base.if",
      "position": [1050, 300]
    },
    {
      "parameters": {
        "resource": "message",
        "channel": "#msp-alerts",
        "text": "=⚠️ SMS Data Sync Failed for {{$env.CLIENT_NAME}}\nError: {{$json.error}}\nTimestamp: {{$now.toISO()}}"
      },
      "name": "Alert MSP (Slack)",
      "type": "n8n-nodes-base.slack",
      "position": [1250, 200]
    }
  ],
  "connections": {
    "Schedule Trigger": { "main": [[{ "node": "Fetch Repair Orders", "type": "main", "index": 0 }]] },
    "Fetch Repair Orders": { "main": [[{ "node": "Upsert Repair Orders", "type": "main", "index": 0 }]] },
    "Upsert Repair Orders": { "main": [[{ "node": "Insert Labor Lines", "type": "main", "index": 0 }, { "node": "Insert Parts Lines", "type": "main", "index": 0 }]] },
    "Insert Labor Lines": { "main": [[{ "node": "Error Check", "type": "main", "index": 0 }]] },
    "Error Check": { "main": [[{ "node": "Alert MSP (Slack)", "type": "main", "index": 0 }], []] }
  },
  "settings": { "executionOrder": "v1", "saveManualExecutions": true, "saveDataSuccessExecution": "all" }
}

-- Daily Aggregation SQL (run via n8n Postgres node on nightly schedule)
-- This processes the previous day's bay events into the aggregation table

WITH shop_hours AS (
    -- Define shop operating hours (adjust per client)
    SELECT 
        (CURRENT_DATE - INTERVAL '1 day')::date AS work_date,
        ((CURRENT_DATE - INTERVAL '1 day') + TIME '07:30:00')::timestamp AS open_time,
        ((CURRENT_DATE - INTERVAL '1 day') + TIME '18:00:00')::timestamp AS close_time,
        630 AS available_minutes -- 10.5 hours * 60 minutes
),
bay_transitions AS (
    SELECT 
        be.bay_id,
        be.event_type,
        be.event_timestamp,
        LEAD(be.event_timestamp) OVER (
            PARTITION BY be.bay_id ORDER BY be.event_timestamp
        ) AS next_event_timestamp,
        LEAD(be.event_type) OVER (
            PARTITION BY be.bay_id ORDER BY be.event_timestamp
        ) AS next_event_type
    FROM analytics.fact_bay_events be
    CROSS JOIN shop_hours sh
    WHERE be.date_key = sh.work_date
    AND be.event_timestamp BETWEEN sh.open_time AND sh.close_time
),
bay_durations AS (
    SELECT
        bay_id,
        event_type,
        EXTRACT(EPOCH FROM (COALESCE(next_event_timestamp, (SELECT close_time FROM shop_hours)) - event_timestamp)) / 60.0 AS duration_minutes
    FROM bay_transitions
),
bay_summary AS (
    SELECT
        bd.bay_id,
        (SELECT work_date FROM shop_hours) AS date_key,
        COALESCE(SUM(CASE WHEN bd.event_type = 'vehicle_in' THEN bd.duration_minutes END), 0)::integer AS total_occupied_minutes,
        COALESCE(SUM(CASE WHEN bd.event_type IN ('vehicle_out', 'idle_start') THEN bd.duration_minutes END), 0)::integer AS total_idle_minutes,
        (SELECT available_minutes FROM shop_hours) AS total_available_minutes,
        COUNT(CASE WHEN bd.event_type = 'vehicle_in' THEN 1 END) AS vehicle_count
    FROM bay_durations bd
    GROUP BY bd.bay_id
)
INSERT INTO analytics.agg_daily_bay_utilization 
    (bay_id, date_key, total_occupied_minutes, total_idle_minutes, total_available_minutes, utilization_pct, vehicle_count, avg_dwell_minutes)
SELECT
    bs.bay_id,
    bs.date_key,
    bs.total_occupied_minutes,
    bs.total_idle_minutes,
    bs.total_available_minutes,
    ROUND((bs.total_occupied_minutes::decimal / NULLIF(bs.total_available_minutes, 0)) * 100, 1) AS utilization_pct,
    bs.vehicle_count,
    CASE WHEN bs.vehicle_count > 0 
        THEN (bs.total_occupied_minutes / bs.vehicle_count)::integer 
        ELSE 0 
    END AS avg_dwell_minutes
FROM bay_summary bs
ON CONFLICT (bay_id, date_key) DO UPDATE SET
    total_occupied_minutes = EXCLUDED.total_occupied_minutes,
    total_idle_minutes = EXCLUDED.total_idle_minutes,
    utilization_pct = EXCLUDED.utilization_pct,
    vehicle_count = EXCLUDED.vehicle_count,
    avg_dwell_minutes = EXCLUDED.avg_dwell_minutes;

-- After aggregation, clean up raw events older than 90 days to manage storage:
DELETE FROM analytics.fact_bay_events 
WHERE date_key < CURRENT_DATE - INTERVAL '90 days';

{
  "name": "PitCrew Webhook Ingest",
  "nodes": [
    {
      "parameters": {
        "httpMethod": "POST",
        "path": "pitcrew-events",
        "authentication": "headerAuth",
        "options": { "responseCode": 200, "responseData": "{ \"status\": \"received\" }" }
      },
      "name": "PitCrew Webhook",
      "type": "n8n-nodes-base.webhook",
      "position": [250, 300]
    },
    {
      "parameters": {
        "jsCode": "// Map PitCrew event to our schema\nconst event = $input.first().json;\nconst bayMapping = {\n  'pitcrew-bay-1': 1,\n  'pitcrew-bay-2': 2,\n  'pitcrew-bay-3': 3,\n  'pitcrew-bay-4': 4,\n  'pitcrew-bay-5': 5,\n  'pitcrew-bay-6': 6\n};\n\nconst eventTypeMapping = {\n  'bay_occupied': 'vehicle_in',\n  'bay_vacated': 'vehicle_out',\n  'idle_detected': 'idle_start',\n  'activity_resumed': 'idle_end'\n};\n\nconst mapped = {\n  bay_id: bayMapping[event.bay_id] || null,\n  date_key: event.timestamp.substring(0, 10),\n  event_type: eventTypeMapping[event.event_type] || event.event_type,\n  event_timestamp: event.timestamp,\n  vehicle_detected: event.vehicle_detected || false,\n  tech_detected: event.technician_detected || false,\n  source: 'pitcrew'\n};\n\nif (!mapped.bay_id) {\n  throw new Error(`Unknown bay_id: ${event.bay_id}`);\n}\n\nreturn [{ json: mapped }];"
      },
      "name": "Map Event",
      "type": "n8n-nodes-base.code",
      "position": [450, 300]
    },
    {
      "parameters": {
        "operation": "executeQuery",
        "query": "INSERT INTO analytics.fact_bay_events (bay_id, date_key, event_type, event_timestamp, vehicle_detected, tech_detected, source) VALUES ($1, $2::date, $3, $4::timestamp, $5, $6, $7)",
        "options": {}
      },
      "name": "Insert Bay Event",
      "type": "n8n-nodes-base.postgres",
      "position": [650, 300],
      "credentials": { "postgres": { "name": "Shop Analytics DB" } }
    }
  ],
  "connections": {
    "PitCrew Webhook": { "main": [[{ "node": "Map Event", "type": "main", "index": 0 }]] },
    "Map Event": { "main": [[{ "node": "Insert Bay Event", "type": "main", "index": 0 }]] }
  }
}

# weekly scheduled script

#!/usr/bin/env python3
"""
Parts Margin Anomaly Detection Agent
Runs weekly via cron or n8n schedule.
Detects margin anomalies and generates alerts.

Dependencies: pip install psycopg2-binary pandas numpy
"""

import os
import json
import smtplib
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
from decimal import Decimal

import psycopg2
import psycopg2.extras
import pandas as pd
import numpy as np

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Configuration (set via environment variables)
DB_HOST = os.environ.get('DB_HOST', 'localhost')
DB_PORT = os.environ.get('DB_PORT', '5432')
DB_NAME = os.environ.get('DB_NAME', 'shop_analytics')
DB_USER = os.environ.get('DB_USER', 'analytics_admin')
DB_PASS = os.environ.get('DB_PASSWORD')
SMTP_HOST = os.environ.get('SMTP_HOST', 'smtp.office365.com')
SMTP_PORT = int(os.environ.get('SMTP_PORT', '587'))
SMTP_USER = os.environ.get('SMTP_USER')
SMTP_PASS = os.environ.get('SMTP_PASSWORD')
ALERT_RECIPIENTS = os.environ.get('ALERT_RECIPIENTS', '').split(',')
CLIENT_NAME = os.environ.get('CLIENT_NAME', 'Auto Shop')

TARGET_MARGIN_PCT = 50.0  # Industry target: 50% GPM
CRITICAL_MARGIN_PCT = 25.0  # Below this is critical
BELOW_COST_THRESHOLD = 0.0  # Sold below cost
TREND_LOOKBACK_WEEKS = 12  # Compare current week to trailing average
MIN_PARTS_LINES = 5  # Minimum lines to trigger job-type alert


def get_connection():
    return psycopg2.connect(
        host=DB_HOST, port=DB_PORT, dbname=DB_NAME,
        user=DB_USER, password=DB_PASS
    )


def detect_below_cost_parts(conn, start_date, end_date):
    """Find parts sold at or below cost in the period."""
    query = """
    SELECT p.part_number, p.part_description, p.vendor,
           p.cost_per_unit, p.sale_per_unit, p.quantity,
           p.total_cost, p.total_sale, p.margin_pct,
           jt.job_type_name, ro.sms_ro_number
    FROM analytics.fact_parts_lines p
    JOIN analytics.fact_repair_orders ro ON p.ro_id = ro.ro_id
    LEFT JOIN analytics.dim_job_types jt ON p.job_type_id = jt.job_type_id
    WHERE p.date_key BETWEEN %s AND %s
      AND p.margin_pct <= %s
    ORDER BY p.margin_amount ASC
    LIMIT 50
    """
    df = pd.read_sql(query, conn, params=[start_date, end_date, BELOW_COST_THRESHOLD])
    return df


def detect_low_margin_job_types(conn, start_date, end_date):
    """Find job types with margin significantly below target."""
    query = """
    SELECT jt.job_type_name, jt.category, jt.target_parts_margin_pct,
           COUNT(*) as parts_line_count,
           ROUND(SUM(p.total_sale)::numeric, 2) as total_parts_sale,
           ROUND(SUM(p.total_cost)::numeric, 2) as total_parts_cost,
           ROUND(AVG(p.margin_pct)::numeric, 1) as avg_margin_pct,
           ROUND((SUM(p.total_sale) - SUM(p.total_cost))::numeric, 2) as total_margin
    FROM analytics.fact_parts_lines p
    JOIN analytics.dim_job_types jt ON p.job_type_id = jt.job_type_id
    WHERE p.date_key BETWEEN %s AND %s
    GROUP BY jt.job_type_name, jt.category, jt.target_parts_margin_pct
    HAVING COUNT(*) >= %s
    ORDER BY AVG(p.margin_pct) ASC
    """
    df = pd.read_sql(query, conn, params=[start_date, end_date, MIN_PARTS_LINES])
    df['below_target'] = df['avg_margin_pct'] < TARGET_MARGIN_PCT
    df['margin_gap'] = TARGET_MARGIN_PCT - df['avg_margin_pct']
    df['potential_recovery'] = (df['margin_gap'] / 100) * df['total_parts_sale']
    return df[df['below_target']]


def detect_margin_trend_decline(conn, current_week_start):
    """Detect if this week's margin is declining vs trailing average."""
    lookback_start = current_week_start - timedelta(weeks=TREND_LOOKBACK_WEEKS)
    query = """
    SELECT
        DATE_TRUNC('week', p.date_key)::date as week_start,
        ROUND(AVG(p.margin_pct)::numeric, 1) as avg_margin_pct,
        ROUND(SUM(p.total_sale)::numeric, 2) as total_parts_sale,
        COUNT(*) as parts_lines
    FROM analytics.fact_parts_lines p
    WHERE p.date_key >= %s
    GROUP BY DATE_TRUNC('week', p.date_key)
    ORDER BY week_start
    """
    df = pd.read_sql(query, conn, params=[lookback_start])
    if len(df) < 4:
        return None, None  # Not enough data
    trailing_avg = df['avg_margin_pct'].iloc[:-1].mean()
    current_margin = df['avg_margin_pct'].iloc[-1]
    decline = trailing_avg - current_margin
    return current_margin, decline if decline > 3.0 else None  # Alert if >3% decline


def generate_alert_email(below_cost_df, low_margin_df, current_margin, margin_decline, period_str):
    """Generate HTML email with anomaly findings."""
    html = f"""
    <html><body style='font-family: Arial, sans-serif;'>
    <h2>📊 {CLIENT_NAME} — Parts Margin Alert Report</h2>
    <p>Period: {period_str}</p>
    """

    if margin_decline:
        html += f"""
        <div style='background: #fff3cd; border-left: 4px solid #ffc107; padding: 12px; margin: 12px 0;'>
            <h3>⚠️ Margin Trend Declining</h3>
            <p>Current week average margin: <strong>{current_margin:.1f}%</strong></p>
            <p>Decline from trailing 12-week average: <strong>{margin_decline:.1f} percentage points</strong></p>
            <p><em>Action: Review parts pricing matrix and vendor costs for recent changes.</em></p>
        </div>
        """

    if not below_cost_df.empty:
        html += f"""
        <div style='background: #f8d7da; border-left: 4px solid #dc3545; padding: 12px; margin: 12px 0;'>
            <h3>🚨 Parts Sold Below Cost ({len(below_cost_df)} items)</h3>
            <table border='1' cellpadding='6' cellspacing='0' style='border-collapse: collapse; font-size: 13px;'>
            <tr style='background: #dc3545; color: white;'>
                <th>Part #</th><th>Description</th><th>Cost</th><th>Sale</th><th>Margin %</th><th>Job Type</th><th>RO #</th>
            </tr>
        """
        for _, row in below_cost_df.head(20).iterrows():
            html += f"<tr><td>{row['part_number']}</td><td>{row['part_description'][:40]}</td><td>${row['cost_per_unit']:.2f}</td><td>${row['sale_per_unit']:.2f}</td><td>{row['margin_pct']:.1f}%</td><td>{row['job_type_name']}</td><td>{row['sms_ro_number']}</td></tr>"
        html += "</table></div>"

    if not low_margin_df.empty:
        total_recovery = low_margin_df['potential_recovery'].sum()
        html += f"""
        <div style='background: #fff3cd; border-left: 4px solid #ffc107; padding: 12px; margin: 12px 0;'>
            <h3>💰 Job Types Below {TARGET_MARGIN_PCT:.0f}% Target Margin</h3>
            <p>Estimated recoverable margin: <strong>${total_recovery:,.2f}</strong></p>
            <table border='1' cellpadding='6' cellspacing='0' style='border-collapse: collapse; font-size: 13px;'>
            <tr style='background: #ffc107;'>
                <th>Job Type</th><th>Category</th><th>Avg Margin</th><th>Parts Revenue</th><th>Gap to Target</th><th>Recovery Potential</th>
            </tr>
        """
        for _, row in low_margin_df.iterrows():
            html += f"<tr><td>{row['job_type_name']}</td><td>{row['category']}</td><td>{row['avg_margin_pct']:.1f}%</td><td>${row['total_parts_sale']:,.2f}</td><td>{row['margin_gap']:.1f}%</td><td>${row['potential_recovery']:,.2f}</td></tr>"
        html += "</table></div>"

    if below_cost_df.empty and low_margin_df.empty and not margin_decline:
        html += "<p style='color: green;'>✅ No margin anomalies detected this period. All job types within target range.</p>"

    html += """
    <hr><p style='font-size: 11px; color: #666;'>Generated by Shop Intelligence Analytics Platform | Managed by your MSP partner</p>
    </body></html>
    """
    return html


def send_email(subject, html_body):
    msg = MIMEMultipart('alternative')
    msg['Subject'] = subject
    msg['From'] = SMTP_USER
    msg['To'] = ', '.join(ALERT_RECIPIENTS)
    msg.attach(MIMEText(html_body, 'html'))
    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASS)
        server.sendmail(SMTP_USER, ALERT_RECIPIENTS, msg.as_string())
    logger.info(f'Alert email sent to {len(ALERT_RECIPIENTS)} recipients')


def main():
    today = datetime.now().date()
    week_start = today - timedelta(days=today.weekday())
    week_end = today
    period_str = f"{week_start.strftime('%b %d')} - {week_end.strftime('%b %d, %Y')}"

    conn = get_connection()
    try:
        below_cost = detect_below_cost_parts(conn, week_start, week_end)
        low_margin = detect_low_margin_job_types(conn, week_start, week_end)
        current_margin, decline = detect_margin_trend_decline(conn, week_start)

        has_anomalies = not below_cost.empty or not low_margin.empty or decline
        if has_anomalies:
            subject = f'⚠️ {CLIENT_NAME} Parts Margin Alert — {period_str}'
        else:
            subject = f'✅ {CLIENT_NAME} Parts Margin OK — {period_str}'

        html = generate_alert_email(below_cost, low_margin, current_margin, decline, period_str)
        send_email(subject, html)

        logger.info(f'Analysis complete: {len(below_cost)} below-cost items, '
                    f'{len(low_margin)} low-margin job types, '
                    f'margin decline: {decline}')
    finally:
        conn.close()


if __name__ == '__main__':
    main()

# Generate Technician Efficiency Report via OpenAI

import openai

def generate_tech_efficiency_report(tech_data, shop_context):
    prompt = USER_PROMPT_TEMPLATE.format(
        CLIENT_NAME=shop_context['name'],
        PERIOD_START=shop_context['period_start'],
        PERIOD_END=shop_context['period_end'],
        BAY_COUNT=shop_context['bay_count'],
        SHOP_HOURS=shop_context['shop_hours'],
        POSTED_LABOR_RATE=shop_context['labor_rate'],
        TECH_COUNT=len(tech_data),
        TECH_DATA_TABLE=format_tech_table(tech_data),
        TOTAL_BILLED=shop_context['total_billed'],
        TOTAL_AVAILABLE=shop_context['total_available'],
        BAY_UTIL_PCT=shop_context['bay_util_pct'],
        AVG_RO_VALUE=shop_context['avg_ro_value'],
        RO_COUNT=shop_context['ro_count']
    )
    
    response = openai.chat.completions.create(
        model='gpt-5.4',
        messages=[
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': prompt}
        ],
        temperature=0.3,
        max_tokens=3000
    )
    return response.choices[0].message.content

# Cost: ~$0.02-0.05 per report generation (GPT-5.4 pricing)
# Schedule: Monthly, timed before QBR meeting

# fallback alternative to Leverege PitCrew. Deploy as a systemd service on
# the edge server.

#!/usr/bin/env python3
"""
Custom Bay Occupancy Detector
Fallback alternative to Leverege PitCrew.
Uses YOLOv8 nano model for vehicle detection on RTSP camera feeds.

Requirements:
  pip install ultralytics opencv-python-headless psycopg2-binary numpy
  Hardware: CPU-only is sufficient for 30-second frame intervals
            GPU recommended if processing more frequently

Deploy: Run as a systemd service on the edge server.
"""

import os
import sys
import time
import logging
from datetime import datetime
from typing import Dict, Optional, Tuple

import cv2
import numpy as np
import psycopg2
from ultralytics import YOLO

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(), logging.FileHandler('/var/log/bay-detector.log')]
)
logger = logging.getLogger('bay_detector')

# Configuration
CAMERAS = {
    1: {'name': 'Bay 1', 'rtsp_url': os.environ.get('CAM_BAY1_RTSP', 'rtsp://192.168.20.10:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},  # Polygon ROI points
    2: {'name': 'Bay 2', 'rtsp_url': os.environ.get('CAM_BAY2_RTSP', 'rtsp://192.168.20.11:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},
    3: {'name': 'Bay 3', 'rtsp_url': os.environ.get('CAM_BAY3_RTSP', 'rtsp://192.168.20.12:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},
    4: {'name': 'Bay 4', 'rtsp_url': os.environ.get('CAM_BAY4_RTSP', 'rtsp://192.168.20.13:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},
    5: {'name': 'Bay 5', 'rtsp_url': os.environ.get('CAM_BAY5_RTSP', 'rtsp://192.168.20.14:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},
    6: {'name': 'Bay 6', 'rtsp_url': os.environ.get('CAM_BAY6_RTSP', 'rtsp://192.168.20.15:7447/stream1'),
        'roi': [(100, 200), (900, 200), (900, 700), (100, 700)]},
}

DB_CONFIG = {
    'host': os.environ.get('DB_HOST', 'localhost'),
    'port': os.environ.get('DB_PORT', '5432'),
    'dbname': os.environ.get('DB_NAME', 'shop_analytics'),
    'user': os.environ.get('DB_USER', 'analytics_admin'),
    'password': os.environ.get('DB_PASSWORD'),
}

CHECK_INTERVAL = 30  # seconds between frame captures
VEHICLE_CLASSES = [2, 5, 7]  # COCO: car=2, bus=5, truck=7
CONFIDENCE_THRESHOLD = 0.45
STATE_CHANGE_CONFIRMATIONS = 3  # Require N consecutive consistent readings


class BayState:
    def __init__(self, bay_id: int):
        self.bay_id = bay_id
        self.occupied = False
        self.last_change = None
        self.confirmation_count = 0
        self.pending_state = None

    def update(self, vehicle_detected: bool) -> Optional[str]:
        """Update bay state. Returns event_type if state changed, None otherwise."""
        if vehicle_detected != self.occupied:
            if self.pending_state == vehicle_detected:
                self.confirmation_count += 1
            else:
                self.pending_state = vehicle_detected
                self.confirmation_count = 1

            if self.confirmation_count >= STATE_CHANGE_CONFIRMATIONS:
                self.occupied = vehicle_detected
                self.last_change = datetime.now()
                self.pending_state = None
                self.confirmation_count = 0
                return 'vehicle_in' if vehicle_detected else 'vehicle_out'
        else:
            self.pending_state = None
            self.confirmation_count = 0
        return None


class BayDetector:
    def __init__(self):
        logger.info('Loading YOLOv8n model...')
        self.model = YOLO('yolov8n.pt')  # Downloads automatically on first run (~6MB)
        self.bay_states: Dict[int, BayState] = {
            bay_id: BayState(bay_id) for bay_id in CAMERAS
        }
        self.db_conn = None
        self._connect_db()

    def _connect_db(self):
        try:
            self.db_conn = psycopg2.connect(**DB_CONFIG)
            self.db_conn.autocommit = True
            logger.info('Database connected')
        except Exception as e:
            logger.error(f'Database connection failed: {e}')
            self.db_conn = None

    def _ensure_db(self):
        if self.db_conn is None or self.db_conn.closed:
            self._connect_db()

    def capture_frame(self, rtsp_url: str) -> Optional[np.ndarray]:
        """Capture a single frame from RTSP stream."""
        try:
            cap = cv2.VideoCapture(rtsp_url)
            cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
            ret, frame = cap.read()
            cap.release()
            if ret:
                return frame
            else:
                logger.warning(f'Failed to capture frame from {rtsp_url}')
                return None
        except Exception as e:
            logger.error(f'Error capturing from {rtsp_url}: {e}')
            return None

    def detect_vehicle_in_roi(self, frame: np.ndarray, roi_points: list) -> bool:
        """Detect if a vehicle is present within the ROI polygon."""
        results = self.model(frame, verbose=False, conf=CONFIDENCE_THRESHOLD)
        roi_polygon = np.array(roi_points, dtype=np.int32)

        for result in results:
            for box in result.boxes:
                cls = int(box.cls[0])
                if cls in VEHICLE_CLASSES:
                    x1, y1, x2, y2 = map(int, box.xyxy[0])
                    center_x = (x1 + x2) // 2
                    center_y = (y1 + y2) // 2
                    if cv2.pointPolygonTest(roi_polygon, (center_x, center_y), False) >= 0:
                        return True
        return False

    def record_event(self, bay_id: int, event_type: str):
        """Write bay event to database."""
        self._ensure_db()
        if not self.db_conn:
            logger.error('Cannot record event: no database connection')
            return
        try:
            now = datetime.now()
            with self.db_conn.cursor() as cur:
                cur.execute(
                    """INSERT INTO analytics.fact_bay_events 
                       (bay_id, date_key, event_type, event_timestamp, vehicle_detected, source)
                       VALUES (%s, %s, %s, %s, %s, 'custom_ai')""",
                    (bay_id, now.date(), event_type, now, event_type == 'vehicle_in')
                )
            logger.info(f'Bay {bay_id}: {event_type} at {now}')
        except Exception as e:
            logger.error(f'Database insert failed: {e}')

    def run(self):
        """Main detection loop."""
        logger.info(f'Starting bay detection for {len(CAMERAS)} bays, interval={CHECK_INTERVAL}s')
        while True:
            for bay_id, cam_config in CAMERAS.items():
                frame = self.capture_frame(cam_config['rtsp_url'])
                if frame is None:
                    continue

                vehicle_present = self.detect_vehicle_in_roi(frame, cam_config['roi'])
                event = self.bay_states[bay_id].update(vehicle_present)

                if event:
                    self.record_event(bay_id, event)

            time.sleep(CHECK_INTERVAL)


if __name__ == '__main__':
    detector = BayDetector()
    detector.run()

# --- SYSTEMD SERVICE FILE ---
# Save as /etc/systemd/system/bay-detector.service
# [Unit]
# Description=Bay Occupancy Detector
# After=network.target docker.service
# Requires=docker.service
#
# [Service]
# Type=simple
# User=analytics
# WorkingDirectory=/opt/shop-analytics/scripts
# EnvironmentFile=/opt/shop-analytics/.env
# ExecStart=/usr/bin/python3 /opt/shop-analytics/scripts/bay_detector.py
# Restart=always
# RestartSec=10
#
# [Install]
# WantedBy=multi-user.target
#
# Enable: sudo systemctl enable bay-detector && sudo systemctl start bay-detector

# --- ROI CALIBRATION TOOL ---
# Run once per camera to define the ROI polygon interactively:
# python3 -c "
# import cv2
# cap = cv2.VideoCapture('rtsp://...')
# ret, frame = cap.read()
# cap.release()
# points = []
# def click(event, x, y, flags, param):
#     if event == cv2.EVENT_LBUTTONDOWN:
#         points.append((x, y))
#         print(f'Point: ({x}, {y}), Total: {len(points)}')
# cv2.imshow('ROI', frame)
# cv2.setMouseCallback('ROI', click)
# cv2.waitKey(0)
# print(f'ROI: {points}')
# "
# Copy the printed ROI points into the CAMERAS config above.

# --- PERFORMANCE NOTES ---
# YOLOv8n on CPU (Intel Xeon E-2314): ~50-80ms per frame inference
# 6 cameras * 1 frame / 30 sec = 12 inferences/minute = very low CPU load
# Memory: ~500MB for model + frame buffers
# Disk: Negligible (no video storage, only event records)

docker compose pull && docker compose up -d