Real-Time Location System (RTLS)

Overview

Haltian IoT includes a Real-Time Location System (RTLS) based on the Haltian Inventory Tracking Solution (HITS). This guide covers how to work with position data in your integrations — querying device locations, streaming real-time position updates, and analyzing movement patterns.

The system consists of:

• Haltian NANO Tags and compatible Wirepas-enabled tags attached to tracked assets
• Haltian EH Locators providing the positioning infrastructure (energy-harvesting, no cabling)
• Haltian Positioning Engine calculating positions server-side from collected RSSI data
• Haltian IoT APIs delivering position data to your applications

Position data is accessed through the same Service API, Stream API, and Data API used for all other Haltian IoT sensor data.

Location Data Types

Haltian IoT provides multiple location data formats:

System Overview

For a complete explanation of how the positioning system works — architecture, components, deployment, and security — see the Indoor Positioning System concept page.

Key points for integration:

• Position is calculated server-side by the Haltian Positioning Engine — tags do not know their own coordinates
• Tags wake on a configurable interval (typically minutes) or on motion, collect RSSI from nearby locators, and transmit a single DA message
• The positioning engine computes coordinates from the RSSI values using patented algorithms
• Calculated positions are published as standard measurements through the Service API, Stream API, and Data API

Position Measurement Types

position (Precise Coordinates)

Detailed position with local and/or global coordinates.

Topic:

haltian-iot/events/{integration}/{api-key}/measurements/position/{device-id}

Payload:

{
  "measured_at": "2026-02-05T10:30:00.000Z",
  "value": {
    "position_global": {
      "type": "Point",
      "coordinates": [25.4538, 65.0121]
    },
    "accuracy": 2.5,
    "floor": 1,
    "building_id": "550e8400-e29b-41d4-a716-446655440000",
    "space_id": "space-123"
  }
}

position_zone (Named Area)

Zone-based position indicating which area/room the device occupies.

Topic:

haltian-iot/events/{integration}/{api-key}/measurements/position_zone/{device-id}

Payload:

{
  "measured_at": "2026-02-05T10:30:00.000Z",
  "value": {
    "zone_id": "zone-uuid-123",
    "zone_name": "Conference Room A",
    "zone_type": "MEETING_ROOM",
    "entered_at": "2026-02-05T10:25:00.000Z",
    "confidence": 0.95
  }
}

Coordinate Systems

Global Coordinates

Global coordinates use WGS84 (standard GPS) and are the primary coordinate system for all positioning in Haltian IoT:

       North (+Y)
          ↑
          |
West ←----+----→ East (+X)
          |
          ↓
       South (-Y)
       
       ↑ Up (+Z)
       |
       + (Floor level)
       |
       ↓ Down (-Z)

{
  "type": "Point",
  "coordinates": [24.9384, 60.1699]
}

Format: GeoJSON Point [longitude, latitude]
Longitude: -180° to +180° (East-West)
Latitude: -90° to +90° (North-South)
Use Cases: All positioning — indoor and outdoor

Local Coordinates (Legacy — Removed in R3)

Local coordinates use a Cartesian system relative to a building origin:

Units: Meters
Origin: Defined per building (typically SW corner of ground floor)
Use Cases: Legacy indoor positioning (being replaced by geo-referenced global coordinates)

Using the APIs for Position Data

Location tracking uses the same Haltian IoT APIs as all other measurement types. The position-specific details are covered below — for full API documentation, see the linked references.

Service API (GraphQL) — Query Positions

Use the Service API to query device positions on demand.

Key fields for position queries:

Complete Position Query Example

This query retrieves all position-related data for a device in a single call:

query GetDevicePosition($deviceId: ID!) {
  device(id: $deviceId) {
    id
    tuid
    name
    deviceModel { name }
    
    fixedPositionGlobal { type coordinates }
    space { id name path }
    
    lastPosition: measurements(
      filter: { types: ["position"] }
      last: 1
      orderBy: { measuredAt: DESC }
    ) {
      edges {
        node { measuredAt value }
      }
    }
    
    lastZone: measurements(
      filter: { types: ["position_zone"] }
      last: 1
      orderBy: { measuredAt: DESC }
    ) {
      edges {
        node { measuredAt value }
      }
    }
  }
}

Response:

{
  "data": {
    "device": {
      "id": "550e8400-e29b-41d4-a716-446655440000",
      "tuid": "TSTG01ABC12345678",
      "name": "Asset Tag #1234",
      "deviceModel": { "name": "Haltian Asset Tag" },
      "fixedPositionGlobal": null,
      "space": {
        "id": "space-123",
        "name": "Warehouse Floor 1",
        "path": "Building A / Warehouse Floor 1"
      },
      "lastPosition": {
        "edges": [{
          "node": {
            "measuredAt": "2026-02-05T10:30:00Z",
            "value": {
              "position_global": {
                "type": "Point",
                "coordinates": [25.4538, 65.0121]
              },
              "accuracy": 2.5,
              "floor": 1,
              "space_id": "space-123"
            }
          }
        }]
      },
      "lastZone": {
        "edges": [{
          "node": {
            "measuredAt": "2026-02-05T10:25:00Z",
            "value": {
              "zone_id": "space-123",
              "zone_name": "Warehouse Floor 1",
              "zone_type": "WAREHOUSE"
            }
          }
        }]
      }
    }
  }
}

For query syntax, authentication, and pagination, see:

• Service API Queries — Device and measurement queries
• Service API Authentication — API key setup

Stream API (MQTT) — Real-Time Position Updates

Use the Stream API to receive position updates in real time via MQTT.

Position topics to subscribe:

# Precise position updates (all devices)
haltian-iot/events/{integration}/{api-key}/measurements/position/#

# Zone entry/exit events (all devices)
haltian-iot/events/{integration}/{api-key}/measurements/position_zone/#

# Single device position
haltian-iot/events/{integration}/{api-key}/measurements/position/{device-id}

For MQTT connection setup, authentication, and code examples, see:

• Stream API Topics — Full topic reference
• Stream API Examples — Connection code in Python, JavaScript, etc.
• Real-Time Data Streaming Guide — Complete MQTT tutorial

Data API (Parquet) — Historical Position Analysis

Use the Data API to analyze position history from exported Parquet files. Position measurements are stored with the same schema as other measurement types.

For Parquet schema, download setup, and analysis examples, see:

• Data API Schema Reference — Column definitions
• Integration Examples — Pandas, DuckDB, Power BI

Distance Calculation with Global Coordinates

When analyzing position history, use the haversine formula to calculate distances between WGS84 points:

import math
from typing import List, Dict

def haversine_distance(lon1: float, lat1: float, lon2: float, lat2: float) -> float:
    """Calculate distance between two WGS84 points in meters."""
    R = 6_371_000  # Earth radius in meters
    phi1, phi2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlambda = math.radians(lon2 - lon1)
    
    a = (math.sin(dphi / 2) ** 2 +
         math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2)
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))


def calculate_total_distance(positions: List[Dict]) -> float:
    """Calculate total distance traveled in meters using global coordinates."""
    distance = 0.0
    
    for i in range(1, len(positions)):
        prev = positions[i - 1]
        curr = positions[i]
        
        if all(v is not None for v in [prev["longitude"], prev["latitude"],
                                        curr["longitude"], curr["latitude"]]):
            distance += haversine_distance(
                prev["longitude"], prev["latitude"],
                curr["longitude"], curr["latitude"]
            )
    
    return distance

Best Practices

1. Choose Appropriate Position Type

def select_position_type(use_case: str) -> str:
    """Select position measurement type based on use case."""
    
    position_types = {
        "asset_tracking": "position",        # Precise coordinates
        "room_occupancy": "position_zone",   # Zone-based
        "desk_booking": "position",          # Precise for desk-level
        "wayfinding": "position",            # Precise for navigation
        "heat_map": "position_zone",         # Zones for aggregation
    }
    
    return position_types.get(use_case, "position")

2. Handle Missing Position Data

def get_best_available_position(device_data: Dict) -> Dict:
    """
    Get position from best available source.
    Priority: Measurement > Fixed > Space assignment
    """
    # 1. Try measurement position
    measurements = device_data.get("lastPosition", {}).get("edges", [])
    if measurements:
        return {
            "type": "measured",
            "data": measurements[0]["node"]["value"]
        }
    
    # 2. Try fixed installation position
    fixed = device_data.get("fixedPositionGlobal")
    if fixed:
        return {
            "type": "fixed",
            "data": {"position_global": fixed}
        }
    
    # 3. Fall back to space assignment
    space = device_data.get("space")
    if space:
        return {
            "type": "space",
            "data": {"zone_name": space["name"]}
        }
    
    return {"type": "unknown", "data": None}

3. Filter by Position Accuracy

def filter_accurate_positions(
    positions: List[Dict],
    max_accuracy: float = 5.0
) -> List[Dict]:
    """
    Filter positions by accuracy threshold.
    
    Args:
        positions: List of position measurements
        max_accuracy: Maximum acceptable accuracy in meters
        
    Returns:
        Filtered list of positions
    """
    return [
        pos for pos in positions
        if pos.get("accuracy") is not None 
        and pos["accuracy"] <= max_accuracy
    ]

4. Detect Stale Position Data

from datetime import datetime, timedelta

def is_position_current(
    timestamp: str,
    max_age: timedelta = timedelta(minutes=5)
) -> bool:
    """
    Check if position data is recent enough.
    
    Args:
        timestamp: ISO 8601 timestamp string
        max_age: Maximum acceptable age
        
    Returns:
        True if position is current
    """
    measured_at = datetime.fromisoformat(
        timestamp.replace("Z", "+00:00")
    )
    age = datetime.now(measured_at.tzinfo) - measured_at
    return age < max_age

5. Generate Heat Map Data

from collections import defaultdict
from typing import Dict, List

class HeatMapGenerator:
    """Generate zone occupancy heat maps."""
    
    def __init__(self):
        self.zone_visits = defaultdict(int)
        self.zone_dwell_time = defaultdict(float)
    
    def record_zone_entry(self, zone_name: str, dwell_seconds: float = 60):
        """Record zone visit."""
        self.zone_visits[zone_name] += 1
        self.zone_dwell_time[zone_name] += dwell_seconds
    
    def get_visit_heat_map(self) -> Dict[str, float]:
        """Get normalized visit frequency heat map."""
        total_visits = sum(self.zone_visits.values())
        
        if total_visits == 0:
            return {}
        
        return {
            zone: count / total_visits
            for zone, count in self.zone_visits.items()
        }
    
    def get_dwell_heat_map(self) -> Dict[str, float]:
        """Get normalized dwell time heat map."""
        total_time = sum(self.zone_dwell_time.values())
        
        if total_time == 0:
            return {}
        
        return {
            zone: time / total_time
            for zone, time in self.zone_dwell_time.items()
        }

AI Agent Instructions

This guide is designed for AI agent implementation of location tracking. Position data uses the same Haltian IoT APIs documented in the Service API, Stream API, and Data API sections.

Architecture Summary

Position is calculated server-side — tags do not know their own coordinates. The flow is:

1. Tag collects RSSI values from nearby locators (beacon + router mode)
2. Tag sends RSSI data as a Wirepas DA message via router → gateway → cloud
3. Positioning engine computes position using RSSI values and patented algorithms
4. Calculated position is published as a position measurement

Required Inputs

1. Device ID: UUID of tracked device
2. API Credentials: API key ID and token
3. Time Range: For historical queries (optional)
4. Accuracy Threshold: Maximum acceptable position error (optional)

Expected Outputs

Position data containing:

• Global coordinates (longitude/latitude in WGS84 as GeoJSON [lon, lat])
• Position accuracy in meters
• Zone/space assignment (zone name, zone type)
• Timestamp of measurement
• Movement history (for time-series queries)

Position-Specific Measurement Types

Implementation Checklist

• Determine position type needed (position for coordinates, position_zone for room-level)
• Parse position_global.coordinates as [longitude, latitude] (GeoJSON order)
• Filter by accuracy field — discard positions with accuracy > threshold
• Detect stale positions using measuredAt timestamp (typical update interval: minutes)
• Use haversine formula for distance calculations between WGS84 points
• Handle missing position data — fall back to fixedPositionGlobal or space assignment
• For real-time: subscribe to MQTT topics measurements/position/# and measurements/position_zone/#
• For historical: query via GraphQL with from/to date filters

Query Patterns

# Get latest position (Service API)
GET_LATEST_POSITION = """
query GetLatestPosition($deviceId: ID!) {
  device(id: $deviceId) {
    fixedPositionGlobal { type coordinates }
    space { id name path }
    measurements(filter: { types: ["position"] }, last: 1) {
      edges {
        node { measuredAt value }
      }
    }
  }
}
"""

# Get position history (Service API)
GET_POSITION_HISTORY = """
query GetPositionHistory($deviceId: ID!, $from: DateTime!, $to: DateTime!) {
  device(id: $deviceId) {
    measurements(
      filter: { types: ["position"], from: $from, to: $to }
      orderBy: { measuredAt: ASC }
      first: 1000
    ) {
      edges {
        node { measuredAt value }
      }
      pageInfo { hasNextPage endCursor }
    }
  }
}
"""

# Get devices in a zone (Service API)
GET_DEVICES_IN_ZONE = """
query GetDevicesInZone($zoneId: ID!) {
  space(id: $zoneId) {
    id name
    devices(filter: { state: ACTIVE }) {
      edges {
        node {
          id tuid name
          measurements(filter: { types: ["position"] }, last: 1) {
            edges { node { measuredAt value } }
          }
        }
      }
    }
  }
}
"""

MQTT Topic Patterns (Stream API)

# All position updates for your integration
haltian-iot/events/{integration}/{api-key}/measurements/position/#

# Zone events only
haltian-iot/events/{integration}/{api-key}/measurements/position_zone/#

# Single device
haltian-iot/events/{integration}/{api-key}/measurements/position/{device-id}

Key API References

• Service API Queries — GraphQL query syntax and examples
• Stream API Topics — MQTT topic structure
• Stream API Examples — MQTT connection code
• Measurement Types — All measurement type definitions

Accuracy Considerations

For accuracy ranges, factors affecting accuracy, and deployment guidelines, see Indoor Positioning — Accuracy.

Key takeaway for integration: Always check the accuracy field in position measurements. Filter or flag positions with accuracy worse than your application’s threshold (e.g., > 5m). Adding beacon-mode EH Locators is the easiest way to improve positioning accuracy — they are battery-powered and require no cabling.

Example Position Data Flow

MQTT Position Event

When subscribed to position topics via the Stream API, you receive events like:

{
  "measured_at": "2026-02-05T10:30:00.000Z",
  "value": {
    "position_global": {
      "type": "Point",
      "coordinates": [25.4538, 65.0121]
    },
    "accuracy": 2.5,
    "floor": 1,
    "space_id": "space-123"
  }
}

Zone Change Event

{
  "measured_at": "2026-02-05T10:25:00.000Z",
  "value": {
    "zone_id": "zone-uuid-123",
    "zone_name": "Warehouse Floor 1",
    "zone_type": "WAREHOUSE",
    "confidence": 0.95
  }
}

Troubleshooting

No Position Updates Received

Causes:

• Device not transmitting
• Wrong MQTT topic subscription
• Device outside locator coverage

Solutions:

1. Verify device is active via GraphQL query
2. Check MQTT topic pattern matches credentials
3. Confirm locators are deployed and operational
4. Test with wildcard subscription for debugging

Low Position Accuracy

Causes:

• Insufficient locator density
• Physical obstructions (walls, metal)
• Low tag battery

Solutions:

1. Deploy additional locators per grid guidelines
2. Position locators with line-of-sight to tracking area
3. Monitor and replace low batteries
4. Use accuracy filtering in application code

Position Data is Stale

Causes:

• Device in sleep mode
• Network connectivity issues
• Low battery reducing update frequency

Solutions:

1. Implement timestamp-based staleness detection
2. Configure device reporting interval
3. Check network infrastructure health
4. Replace batteries if below threshold

Migrating from Local to Global Coordinates (R3)

What Is Changing

Why

• Simplifies the positioning model to a single standard
• Eliminates ambiguity between building-relative and absolute positions
• Enables consistent integration with mapping and GIS tools

Who Is Affected

Take action if your integration:

• Consumes positionLocal (X/Y/Z) from the Stream API
• Reads local coordinate columns from Data API (Parquet exports)
• Queries local coordinate fields via the Service API (GraphQL)
• Uses Cartesian reference points in floor plan configurations
• Has dashboards built on local coordinate data

Affected APIs

Migration Steps

1. Audit — Identify integrations consuming positionLocal
2. Configure — Ensure floor plans have global coordinate reference points set
3. Update — Modify integrations to use global coordinates: position_global in Stream API (MQTT) payloads, positionGlobal in Service API (GraphQL) and Data API (Parquet) fields
4. Convert — Transform existing local coordinate data to global using floor plan geo-reference
5. Test — Verify position data flow with global coordinates only

Recommended Timeline

Position Payload After R3

{
  "measured_at": "2026-08-15T10:30:00.000Z",
  "value": {
    "position_global": {
      "type": "Point",
      "coordinates": [24.9384, 60.1699]
    },
    "accuracy": 2.5,
    "floor": 2
  }
}

Next Steps

• Indoor Positioning System - Architecture, components, deployment, and security
• Building a Complete Application - Combine with search and context
• Occupancy Data Engine - Zone-based occupancy tracking
• Real-Time Data Streaming - MQTT fundamentals
• Service API Queries - Advanced GraphQL patterns

Summary

This guide covered:

✅ Position Types - Global GPS and zone-based tracking
✅ Coordinate Systems - WGS84 global coordinates (local coordinates deprecated, removed in R3)
✅ API Integration - GraphQL queries, MQTT topics, and Parquet access for position data
✅ Best Practices - Accuracy filtering, staleness detection, distance calculation
✅ R3 Migration - Transitioning from local to global coordinates

All code examples are production-ready and verified against Haltian IoT’s positioning infrastructure.