Skip to content

Instantly share code, notes, and snippets.

@khskekec

khskekec/libre-link-up-http-dump.md

Last active April 16, 2026 03:35
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save khskekec/6c13ba01b10d3018d816706a32ae8ab2 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save khskekec/6c13ba01b10d3018d816706a32ae8ab2 to your computer and use it in GitHub Desktop.
Download ZIP
HTTP dump of Libre Link Up used in combination with FreeStyle Libre 3
@Mynuggets-dev
Copy link
Copy Markdown

Mynuggets-dev commented Apr 16, 2026 •
edited
Loading

im not a coder and can be cleaner but if this helps others than thats my intent, here is some of the code ive been using, hasnt broken in months of pulling readings per min

"""
LibreLinkUp API client.

  • Authenticates against LibreView/LibreLinkUp endpoints.
  • Fetches connections and glucose readings.
  • Uses Android-style headers and Account-Id hashing for compatibility.
  • Implements exponential backoff for transient network/rate-limit errors.
  • Normalizes timestamps to naive UTC for storage consistency.

Created by Mynuggets.dev
Developer: Luke Fordham
"""
import hashlib
import logging
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from functools import wraps
from typing import Any, Callable, Generic, List, Mapping, Optional, ParamSpec, TypeVar, cast

import requests

from backend.core.models import GlucoseReading

logger = logging.getLogger(name)

def _mask_email(email: str) -> str:
s = (email or "").strip()
if "@" not in s:
return (s[:3] + "") if s else ""
local, domain = s.split("@", 1)
local_masked = (local[:3] + "") if local else ""
return f"{local_masked}@{domain}"

P = ParamSpec("P")
R = TypeVar("R")
T = TypeVar("T")

@DataClass(frozen=True)
class LluResult(Generic[T]):
ok: bool
data: Optional[T] = None
error: Optional[str] = None
status: Optional[int] = None
retry_after: Optional[int] = None
lockout_seconds: Optional[int] = None

def _is_non_retryable_client_error(status: Optional[int]) -> bool:
return status is not None and 400 <= status < 500 and status not in (429, 401, 430)

def _retry_after_seconds(resp: Any) -> Optional[int]:
if resp is None:
return None
headers = getattr(resp, "headers", None)
if not isinstance(headers, Mapping):
return None
headers_typed = cast(Mapping[str, Any], headers)
ra = headers_typed.get("Retry-After")
if not ra:
return None
try:
return int(float(ra))
except (TypeError, ValueError):
return None

def _compute_retry_delay(
*,
attempt: int,
base_delay: float,
max_delay: float,
status: Optional[int],
retry_after: Optional[int],
) -> float:
delay = min(base_delay * (2 ** attempt), max_delay)
if status in (429, 430) and retry_after is not None:
delay = min(max(delay, retry_after), max_delay)
return delay

def exponential_backoff_retry(
max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""Decorator for exponential backoff retry on transient errors"""
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
last_exception: Optional[BaseException] = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.exceptions.RequestException as e:
last_exception = e
resp = getattr(e, "response", None)
status = getattr(resp, "status_code", None)

                if _is_non_retryable_client_error(status):
                    logger.debug(f"Client error {status}, not retrying")
                    raise

                if attempt == max_retries - 1:
                    break

                retry_after = _retry_after_seconds(resp)
                delay = _compute_retry_delay(
                    attempt=attempt,
                    base_delay=base_delay,
                    max_delay=max_delay,
                    status=status,
                    retry_after=retry_after,
                )
                logger.warning(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}. Retrying in {delay:.1f}s...")
                time.sleep(delay)
        

        logger.error(f"All retries exhausted for {func.__name__}")
        if last_exception is not None:
            raise last_exception
        raise RuntimeError(f"All retries exhausted for {func.__name__}, but no exception was captured")
    
    return wrapper
return decorator

class LibreLinkUpClient:
"""Client for interacting with LibreLinkUp API (spoofed Android client)."""

def __init__(
    self,
    email: str,
    password: str,
    region: str = "au",
    client_version: str = "4.17.0",
    base_url: str = "https://api.libreview.io",
):
    """
    Args:
        email: LibreView / LibreLinkUp account email
        password: account password
        region: region code (e.g. 'au', 'eu', 'us')
        client_version: LibreLinkUp client version to spoof
        base_url: API base URL
    """
    self.email = email.strip().lower()
    self.password = password
    self.region = region.lower()
    self.client_version = client_version
    self.base_url = base_url.rstrip("/")
    
    self.session = requests.Session()

    self.session.cookies.clear()
    self.session.headers.clear()
    self.token: Optional[str] = None
    self.user_id: Optional[str] = None
    self.account_id_hash: Optional[str] = None
    # Last error info (used by polling + UI to show better messages)
    self.last_error: Optional[str] = None
    self.lockout_seconds: Optional[int] = None

def _base_headers(self) -> dict[str, str]:
    """Base headers mimicking the official Android client.
    
    Product/version are required after recent API changes.
    """
    return {
        "accept-encoding": "gzip",
        "cache-control": "no-cache, no-store, must-revalidate",
        "pragma": "no-cache",
        "expires": "0",
        "connection": "Keep-Alive",
        "content-type": "application/json",
        "accept": "application/json",
        "product": "llu.android",
        "version": self.client_version,
        "user-agent": f"LibreLinkUp/{self.client_version}",
    }

def _timeout(self) -> tuple[int, int]:
    """Timeout split (connect, read) to avoid long stalls."""
    return (5, 15)

@staticmethod
def _redact(value: Any) -> Any:
    """Best-effort redaction for debug logging."""
    if isinstance(value, dict):
        out: dict[str, Any] = {}
        for k, v in value.items():
            kl = str(k).lower()
            if kl in {"token", "authticket", "authorization", "password", "email", "auth_token", "refresh_token"}:
                out[str(k)] = "<redacted>"
            else:
                out[str(k)] = LibreLinkUpClient._redact(v)
        return out
    if isinstance(value, list):
        return [LibreLinkUpClient._redact(v) for v in value]
    return value

@staticmethod
def _select_connection(connections: list[dict[str, Any]], patient_id: Optional[str]) -> Optional[dict[str, Any]]:
    """Select the matching connection by patient id (string-compared)."""
    if not connections:
        return None
    if not patient_id:
        return connections[0]

    target = str(patient_id)
    for conn in connections:
        pid = conn.get("patientId") or conn.get("patientID") or conn.get("id")
        if pid is not None and str(pid) == target:
            return conn
    return None

def _extract_account_id(self, data: Mapping[str, Any]) -> Optional[str]:
    """Try to locate a stable account identifier from a login response.

    LibreLinkUp response shapes vary; prefer a server-provided account id when present.
    """
    keys = ("accountId", "accountID", "account_id", "account")

    def search(obj: Any, depth: int) -> Optional[str]:
        if depth <= 0:
            return None
        if isinstance(obj, dict):
            for k in keys:
                if k in obj and obj[k] is not None:
                    return str(obj[k])
            for v in obj.values():
                found = search(v, depth - 1)
                if found:
                    return found
        if isinstance(obj, list):
            for v in obj:
                found = search(v, depth - 1)
                if found:
                    return found
        return None

    return search(data, 3)

@staticmethod
def _json_object(value: Any) -> Optional[dict[str, Any]]:
    if isinstance(value, dict):
        return cast(dict[str, Any], value)
    return None

@staticmethod
def _json_list(value: Any) -> Optional[list[Any]]:
    if isinstance(value, list):
        return cast(list[Any], value)
    return None

@staticmethod
def _coerce_bool(value: Any, *, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return bool(value)
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in ("true", "1", "yes", "y", "on"):
            return True
        if lowered in ("false", "0", "no", "n", "off"):
            return False
    return default

@staticmethod
def _coerce_int(value: Any, *, default: int = 0) -> int:
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        try:
            return int(float(value))
        except ValueError:
            return default
    return default

@staticmethod
def _epoch_to_utc_naive_seconds_or_ms(value: float) -> datetime:
    seconds = value / 1000 if value > 1e12 else value
    return datetime.fromtimestamp(seconds, tz=timezone.utc).replace(tzinfo=None)

def _parse_timestamp_str(self, ts: str) -> Optional[datetime]:
    formats = (
        "%m/%d/%Y %I:%M:%S %p",  # "11/24/2025 8:57:12 AM"
        "%m/%d/%Y %H:%M:%S",     # "11/24/2025 08:57:12"
        "%Y-%m-%d %H:%M:%S",     # "2025-11-24 08:57:12"
        "%Y-%m-%dT%H:%M:%S",     # ISO without timezone
        "%Y-%m-%dT%H:%M:%S.%f",  # ISO with microseconds
        "%Y-%m-%dT%H:%M:%SZ",    # ISO with Z
        "%Y-%m-%dT%H:%M:%S%z",   # ISO with timezone
    )

    for fmt in formats:
        try:
            parsed = datetime.strptime(ts, fmt)
            return self._to_utc_naive(parsed)
        except ValueError:
            continue

    try:
        ts_iso = ts.replace("Z", "+00:00")
        parsed = datetime.fromisoformat(ts_iso)
        return self._to_utc_naive(parsed)
    except ValueError:
        pass

    try:
        numeric = float(ts)
        return self._epoch_to_utc_naive_seconds_or_ms(numeric)
    except (ValueError, TypeError):
        return None

def _dedupe_connections(self, conns_list: list[Any]) -> tuple[List[dict[str, Any]], int]:
    seen: set[str] = set()
    unique: List[dict[str, Any]] = []
    dupes = 0

    for item in conns_list:
        obj = self._json_object(item)
        if obj is None:
            continue
        pid = obj.get("patientId") or obj.get("patientID") or obj.get("id")
        key = str(pid) if pid is not None else None
        if not key:
            unique.append(obj)
            continue
        if key in seen:
            dupes += 1
            continue
        seen.add(key)
        unique.append(obj)

    return unique, dupes

def _connections_from_response(self, resp: requests.Response) -> List[dict[str, Any]]:
    if resp.status_code == 204 or not resp.text:
        logger.debug("Connections endpoint returned 204 or empty")
        return []

    data = self._json_object(resp.json())
    if data is None:
        logger.warning("Connections response is not a JSON object")
        return []

    logger.debug(f"Connections response keys: {list(data.keys())}")

    conns_any = data.get("data")
    conns_list = self._json_list(conns_any)
    if conns_list is None:
        logger.warning(f"Connections data is not a list: {type(conns_any)}")
        if conns_any:
            logger.warning(f"Connections data type: {type(conns_any)}, value: {str(conns_any)[:200]}")
        return []

    unique, dupes = self._dedupe_connections(conns_list)
    if dupes:
        logger.warning("De-duped %d duplicate connection(s) from LibreLinkUp response", dupes)

    logger.info(f"Found {len(unique)} connection(s)")
    if unique:
        logger.debug(f"First connection keys: {list(unique[0].keys())}")
    return unique

def _extract_graph_entries(self, data: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Extract graph entries as a list of JSON objects, filtering non-dicts."""
    candidates = (
        self._json_object(data.get("data", {})) or {}
    )

    graph_data_any = (
        candidates.get("graphData")
        or candidates.get("GraphData")
        or data.get("graphData")
        or data.get("GraphData")
        or data.get("data")
    )

    graph_list = self._json_list(graph_data_any)
    if not graph_list:
        return []

    entries: list[dict[str, Any]] = []
    for item in graph_list:
        obj = self._json_object(item)
        if obj is not None:
            entries.append(obj)
    return entries

def _get_headers(self) -> dict[str, str]:
    """Headers for authenticated calls (connections, graph)."""
    headers = self._base_headers()
    
    if self.token:
        headers["authorization"] = f"Bearer {self.token}"
    

    if self.account_id_hash:
        headers["Account-Id"] = self.account_id_hash
        headers["account-id"] = self.account_id_hash  # both casings, to be safe
    

    headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    headers["Pragma"] = "no-cache"
    
    return headers

def _to_utc_naive(self, dt: datetime) -> datetime:
    """Normalize a datetime to naive UTC.

    We store timestamps as naive datetimes throughout the app. To avoid server-
    local timezone shifts, we normalize any timezone-aware timestamp to UTC and
    drop tzinfo.
    """
    if dt.tzinfo is None:
        return dt
    return dt.astimezone(timezone.utc).replace(tzinfo=None)

@exponential_backoff_retry(max_retries=3, base_delay=1.0)
def login(self) -> bool:
    """Authenticate with LibreLinkUp API and initialise token + account-id hash."""

    existing_token = self.token
    existing_user_id = self.user_id
    existing_account_id_hash = self.account_id_hash
    

    self.token = None
    self.user_id = None
    self.account_id_hash = None
    self.last_error = None
    self.lockout_seconds = None

    self.session.cookies.clear()
    
    login_url = f"{self.base_url}/llu/auth/login"
    payload = {
        "email": self.email,
        "password": self.password,
    }
    
    headers = self._base_headers()
    
    logger.info("Authenticating with LibreLinkUp API for %s", _mask_email(self.email))
    
    try:
        resp = self.session.post(login_url, json=payload, headers=headers, timeout=self._timeout())
        

        if resp.status_code == 429:
            retry_after = resp.headers.get('Retry-After', 'unknown')
            logger.error(f"RATE LIMITED (429): Too many requests. Retry-After: {retry_after}s")
            logger.debug("Rate limit response body suppressed")

            self.token = existing_token
            self.user_id = existing_user_id
            self.account_id_hash = existing_account_id_hash
            return False
        
        if resp.status_code == 430:
            logger.error("RATE LIMITED (430): Too many requests to LibreLinkUp API")
            logger.debug("Rate limit response body suppressed")

            self.token = existing_token
            self.user_id = existing_user_id
            self.account_id_hash = existing_account_id_hash
            return False
        
        resp.raise_for_status()
        
        data_any = resp.json()
        if not isinstance(data_any, dict):
            self.last_error = "unexpected_login_response"
            logger.error("Login failed: unexpected response type")

            self.token = existing_token
            self.user_id = existing_user_id
            self.account_id_hash = existing_account_id_hash
            return False

        data = cast(dict[str, Any], data_any)

        # Handle LibreLinkUp explicit error shapes first (prevents confusing "missing token" logs)
        # Example: {'status': 2, 'error': {'message': 'incorrect username/password'}}
        err_msg = None
        error_obj = data.get("error")
        if isinstance(error_obj, dict):
            error_obj_typed = cast(dict[str, Any], error_obj)
            err_msg = error_obj_typed.get("message")
        if err_msg:
            self.last_error = str(err_msg)
            return False

        # Example lockout: {'status': 429, 'data': {'code': 60, 'data': {'lockout': 300}, 'message': 'locked'}}
        if data.get("status") == 429:
            data_obj = data.get("data")
            if isinstance(data_obj, dict):
                data_obj_typed = cast(dict[str, Any], data_obj)
                msg = data_obj_typed.get("message") or data_obj_typed.get("Message")
                if str(msg).lower() == "locked":
                    self.last_error = "locked"
                    details_any = data_obj_typed.get("data")
                    details = cast(dict[str, Any], details_any) if isinstance(details_any, dict) else {}
                    try:
                        self.lockout_seconds = int(details.get("lockout") or 300)
                    except Exception:
                        self.lockout_seconds = 300
                    return False
        

        logger.debug("Login response keys: %s", list(data.keys()))
        if "data" in data and isinstance(data.get("data"), dict):
            logger.debug("Login data keys: %s", list(cast(dict[str, Any], data.get("data", {})).keys()))
        

        token = None
        user_id = None
        

        if "data" in data:
            data_obj = data.get("data", {})

            auth_ticket = data_obj.get("authTicket") or data_obj.get("auth_ticket") or data_obj.get("ticket")
            if auth_ticket:
                token = auth_ticket.get("token") or auth_ticket.get("Token")
            

            user = data_obj.get("user") or data_obj.get("User")
            if user:
                user_id = user.get("id") or user.get("Id") or user.get("userId") or user.get("user_id")
        

        if not token:
            token = data.get("token") or data.get("Token") or data.get("authToken") or data.get("auth_token")
        if not user_id:
            user_obj = data.get("user") or data.get("User")
            if user_obj:
                user_id = user_obj.get("id") or user_obj.get("Id") or user_obj.get("userId") or user_obj.get("user_id")
        
        if not token or not user_id:
            # Some failures come back without "error" but still indicate locked/invalid credentials in "status".
            if data.get("status") and not self.last_error:
                self.last_error = f"login_failed_status_{data.get('status')}"
            logger.error("Login failed: missing token or user.id in response")

            self.token = existing_token
            self.user_id = existing_user_id
            self.account_id_hash = existing_account_id_hash
            return False
        
        self.token = token
        self.user_id = str(user_id)

        # Prefer server-provided account id when present; fall back to user_id.
        account_id = self._extract_account_id(data) or str(user_id)
        self.account_id_hash = hashlib.sha256(str(account_id).encode("utf-8")).hexdigest()

        logger.info("Authentication successful")
        return True
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            self.last_error = "incorrect username/password"
            logger.error("Login failed: Invalid credentials")
        elif e.response.status_code == 429:
            retry_after = e.response.headers.get('Retry-After', 'unknown')
            self.last_error = "rate_limited"
            logger.error(f"RATE LIMITED (429): Too many requests. Retry-After: {retry_after}s")
        elif e.response.status_code == 430:
            self.last_error = "rate_limited"
            logger.error("RATE LIMITED (430): Too many requests to LibreLinkUp API")
        else:
            self.last_error = f"http_{e.response.status_code}"
            logger.error(f"Login failed: HTTP {e.response.status_code}")

        self.token = existing_token
        self.user_id = existing_user_id
        self.account_id_hash = existing_account_id_hash
        return False
    except Exception as e:
        self.last_error = "exception"
        logger.exception("Login failed: unexpected exception")

        self.token = existing_token
        self.user_id = existing_user_id
        self.account_id_hash = existing_account_id_hash
        return False

def _request_with_refresh(
    self,
    method: str,
    url: str,
    *,
    headers: dict[str, str],
    json_body: Any | None = None,
    allow_refresh: bool = True,
) -> requests.Response:
    """Perform one request, optionally retrying once after re-auth on 401.

    This is intentionally non-recursive to avoid rate-limit amplification.
    """
    resp = self.session.request(method, url, headers=headers, json=json_body, timeout=self._timeout())
    if resp.status_code == 401 and allow_refresh:
        # Attempt to refresh auth once.
        self.token = None
        self.account_id_hash = None
        if self.login():
            refreshed_headers = self._get_headers()
            return self.session.request(method, url, headers=refreshed_headers, json=json_body, timeout=self._timeout())
    return resp

def _result_from_response(self, *, resp: requests.Response, data: Optional[T] = None, error: Optional[str] = None) -> LluResult[T]:
    status = getattr(resp, "status_code", None)
    retry_after = _retry_after_seconds(resp)
    lockout_seconds = self.lockout_seconds
    return LluResult(ok=bool(status and 200 <= status < 300), data=data, error=error, status=status, retry_after=retry_after, lockout_seconds=lockout_seconds)

@exponential_backoff_retry(max_retries=3, base_delay=1.0)
def get_connections(self) -> List[dict[str, Any]]:
    return self.get_connections_result().data or []

@exponential_backoff_retry(max_retries=3, base_delay=1.0)
def get_connections_result(self) -> LluResult[List[dict[str, Any]]]:
    """Get connected patients / followers (Result form)."""
    if not self.token or not self.account_id_hash:
        self.last_error = self.last_error or "unauthorized"
        return LluResult(ok=False, data=None, error="unauthorized", status=401)

    cache_buster = int(time.time() * 1000)
    url = f"{self.base_url}/llu/connections?_t={cache_buster}"

    try:
        resp = self._request_with_refresh("GET", url, headers=self._get_headers(), allow_refresh=True)

        if resp.status_code in (429, 430):
            self.last_error = "rate_limited"
            return self._result_from_response(resp=resp, data=None, error="rate_limited")

        if resp.status_code == 401:
            self.last_error = "unauthorized"
            return self._result_from_response(resp=resp, data=None, error="unauthorized")

        resp.raise_for_status()
        conns = self._connections_from_response(resp)
        return self._result_from_response(resp=resp, data=conns, error=None)
    except requests.exceptions.HTTPError as e:
        r = getattr(e, "response", None)
        status = getattr(r, "status_code", None)
        if status == 401:
            self.last_error = "unauthorized"
            return LluResult(ok=False, data=None, error="unauthorized", status=401)
        if status in (429, 430):
            self.last_error = "rate_limited"
            return LluResult(ok=False, data=None, error="rate_limited", status=status, retry_after=_retry_after_seconds(r))
        self.last_error = f"http_{status}" if status else "http_error"
        return LluResult(ok=False, data=None, error=self.last_error, status=status)
    except requests.exceptions.RequestException as e:
        self.last_error = "network_error"
        logger.debug("Connections request error: %s", e)
        raise

def get_latest_reading(self, patient_id: Optional[str] = None) -> Optional[GlucoseReading]:
    return self.get_latest_reading_result(patient_id=patient_id).data

def get_latest_reading_result(self, patient_id: Optional[str] = None) -> LluResult[GlucoseReading]:
    """Fetch ONLY the most recent glucose reading (Result form).

    Minimize calls:
    - Prefer connections.glucoseMeasurement
    - Fall back to graph only when needed
    """

    if not self.token or not self.account_id_hash:
        logger.debug("No token, attempting login...")
        if not self.login():
            self.last_error = self.last_error or "unauthorized"
            return LluResult(ok=False, data=None, error=self.last_error, status=401, retry_after=None, lockout_seconds=self.lockout_seconds)

    # Connections call (at most once per latest-reading request)
    connections_result = self.get_connections_result()
    if not connections_result.ok:
        return LluResult(
            ok=False,
            data=None,
            error=connections_result.error,
            status=connections_result.status,
            retry_after=connections_result.retry_after,
            lockout_seconds=connections_result.lockout_seconds,
        )

    connections = connections_result.data or []
    conn = self._select_connection(connections, patient_id)

    target_id: Optional[str] = None
    if conn:
        target_id = conn.get("patientId") or conn.get("id")
    if not target_id and patient_id:
        target_id = str(patient_id)
    if not target_id and self.user_id:
        target_id = str(self.user_id)

    if not target_id:
        self.last_error = "no_patient"
        return LluResult(ok=False, data=None, error="no_patient", status=404)

    # Prefer glucoseMeasurement for the selected patient.
    if conn and isinstance(conn, dict):
        glucose_measurement = conn.get("glucoseMeasurement") or conn.get("glucoseItem")
        
        if glucose_measurement and isinstance(glucose_measurement, dict):
            gm = cast(dict[str, Any], glucose_measurement)

            ts = (
                gm.get("Timestamp") or
                gm.get("FactoryTimestamp") or
                gm.get("timestamp") or
                gm.get("factoryTimestamp")
            )
            timestamp = self._parse_timestamp(ts)
            value = self._parse_glucose_value(gm)
            
            if timestamp and value is not None:

                timestamp = self._to_utc_naive(timestamp)
                

                is_high = self._coerce_bool(gm.get("isHigh", gm.get("IsHigh", False)))
                is_low = self._coerce_bool(gm.get("isLow", gm.get("IsLow", False)))
                

                if not is_high and not is_low:
                    is_high = value > 180
                    is_low = value < 70
                
                reading = GlucoseReading(
                    timestamp=timestamp,
                    value=value,
                    is_high=is_high,
                    is_low=is_low,
                    trend=self._coerce_int(gm.get("Trend", gm.get("trend", 0))) or 0,
                    notes="",
                    source="librelinkup",
                    device_id=str(target_id),
                )
                
                logger.debug("Latest reading from glucoseMeasurement for %s: %s mg/dL", str(target_id), value)
                return LluResult(ok=True, data=reading, error=None, status=200)
    


    cache_buster = int(time.time() * 1000)
    url = f"{self.base_url}/llu/connections/{target_id}/graph?minutes=60&_t={cache_buster}"
    headers = self._get_headers()
    
    logger.debug("Fetching latest reading from graph for %s", str(target_id))
    
    try:
        resp = self._request_with_refresh("GET", url, headers=headers, allow_refresh=True)
        
        if resp.status_code in (429, 430):
            self.last_error = "rate_limited"
            return LluResult(
                ok=False,
                data=None,
                error="rate_limited",
                status=resp.status_code,
                retry_after=_retry_after_seconds(resp),
                lockout_seconds=self.lockout_seconds,
            )

        if resp.status_code == 401:
            self.last_error = "unauthorized"
            return LluResult(
                ok=False,
                data=None,
                error="unauthorized",
                status=401,
                retry_after=_retry_after_seconds(resp),
                lockout_seconds=self.lockout_seconds,
            )
        
        resp.raise_for_status()
        
        data_any = resp.json()
        data = self._json_object(data_any)
        if data is None:
            self.last_error = "parse_error"
            logger.debug("Graph response is not a JSON object")
            return LluResult(ok=False, data=None, error="parse_error", status=502)

        graph_entries = self._extract_graph_entries(data)
        if not graph_entries:
            logger.debug("No graph data in response for latest reading")
            self.last_error = "no_data"
            return LluResult(ok=False, data=None, error="no_data", status=204)

        latest_entry: Optional[dict[str, Any]] = None
        latest_timestamp = None
        
        for entry in graph_entries:
            ts = (
                entry.get("Timestamp") or 
                entry.get("FactoryTimestamp") or 
                entry.get("timestamp") or 
                entry.get("factoryTimestamp")
            )
            timestamp = self._parse_timestamp(ts)
            value = self._parse_glucose_value(entry)
            
            if timestamp is None or value is None:
                continue
            

            timestamp = self._to_utc_naive(timestamp)
            
            if latest_timestamp is None or timestamp > latest_timestamp:
                latest_timestamp = timestamp
                latest_entry = entry
        
        if not latest_entry or latest_timestamp is None:
            logger.debug("No valid reading found in graph data")
            self.last_error = "no_data"
            return LluResult(ok=False, data=None, error="no_data", status=204)
        
        value = self._parse_glucose_value(latest_entry)
        if value is None:
            self.last_error = "parse_error"
            return LluResult(ok=False, data=None, error="parse_error", status=502)
        

        is_high = self._coerce_bool(latest_entry.get("isHigh", latest_entry.get("IsHigh", False)))
        is_low = self._coerce_bool(latest_entry.get("isLow", latest_entry.get("IsLow", False)))
        

        if not is_high and not is_low:
            measurement_color = self._coerce_int(
                latest_entry.get("MeasurementColor", latest_entry.get("measurementColor")),
                default=-1,
            )
            if measurement_color in (2, 3):
                is_high = True
            elif measurement_color == 1:
                is_high = False
                is_low = False
        

        if not is_high and not is_low:
            is_high = value > 180
            is_low = value < 70
        
        reading = GlucoseReading(
            timestamp=self._to_utc_naive(latest_timestamp),
            value=value,
            is_high=is_high,
            is_low=is_low,
            trend=self._coerce_int(latest_entry.get("Trend", latest_entry.get("trend", 0))) or 0,
            notes="",
            source="librelinkup",
            device_id=str(target_id),
        )
        
        logger.debug("Latest reading from graph for %s: %s mg/dL", str(target_id), value)
        return LluResult(ok=True, data=reading, error=None, status=200)
        
    except requests.exceptions.HTTPError as e:
        status = getattr(getattr(e, "response", None), "status_code", None)
        if status in (429, 430):
            self.last_error = "rate_limited"
            return LluResult(ok=False, data=None, error="rate_limited", status=status, retry_after=_retry_after_seconds(getattr(e, "response", None)))
        if status == 401:
            self.last_error = "unauthorized"
            return LluResult(ok=False, data=None, error="unauthorized", status=401)
        self.last_error = f"http_{status}" if status else "http_error"
        return LluResult(ok=False, data=None, error=self.last_error, status=status)
    except Exception as e:
        self.last_error = "exception"
        logger.debug("Failed to fetch latest reading: %s", e)
        return LluResult(ok=False, data=None, error="exception", status=502)

@exponential_backoff_retry(max_retries=3, base_delay=1.0)
def get_glucose_readings(self, patient_id: Optional[str] = None, days: int = 90) -> Optional[List[GlucoseReading]]:
    return self.get_glucose_readings_result(patient_id=patient_id, days=days).data

@exponential_backoff_retry(max_retries=3, base_delay=1.0)
def get_glucose_readings_result(self, patient_id: Optional[str] = None, days: int = 90) -> LluResult[List[GlucoseReading]]:
    """Fetch glucose readings for a patient."""

    if not self.token or not self.account_id_hash:
        logger.debug("No token, attempting login...")
        if not self.login():
            self.last_error = self.last_error or "unauthorized"
            return LluResult(ok=False, data=None, error=self.last_error, status=401, lockout_seconds=self.lockout_seconds)
    
    target_id = patient_id
    if not target_id:
        connections_result = self.get_connections_result()
        if not connections_result.ok:
            return LluResult(ok=False, data=None, error=connections_result.error, status=connections_result.status, retry_after=connections_result.retry_after, lockout_seconds=connections_result.lockout_seconds)
        connections = connections_result.data or []
        conn = self._select_connection(connections, None)
        if conn:
            target_id = conn.get("patientId") or conn.get("id")
        if not target_id and self.user_id:
            target_id = self.user_id
    
    if not target_id:
        logger.error("No patient ID available for fetching readings")
        self.last_error = "no_patient"
        return LluResult(ok=False, data=None, error="no_patient", status=404)
    
    minutes = days * 1440
    cache_buster = int(time.time() * 1000)
    url = f"{self.base_url}/llu/connections/{target_id}/graph?minutes={minutes}&_t={cache_buster}"
    headers = self._get_headers()
    
    logger.debug("Fetching glucose readings (days=%s) for %s", str(days), str(target_id))
    
    try:
        resp = self._request_with_refresh("GET", url, headers=headers, allow_refresh=True)

        if resp.status_code in (429, 430):
            self.last_error = "rate_limited"
            return LluResult(ok=False, data=None, error="rate_limited", status=resp.status_code, retry_after=_retry_after_seconds(resp))

        if resp.status_code == 401:
            self.last_error = "unauthorized"
            return LluResult(ok=False, data=None, error="unauthorized", status=401)

        resp.raise_for_status()
        
        data_any = resp.json()
        data = self._json_object(data_any)
        if data is None:
            logger.warning("Graph API response is not a JSON object")
            self.last_error = "parse_error"
            return LluResult(ok=False, data=None, error="parse_error", status=502)

        logger.debug("Graph API response keys: %s", list(data.keys()))

        graph_entries = self._extract_graph_entries(data)
        

        if not graph_entries and "data" in data:
            data_data = data.get("data")
            data_obj = self._json_object(data_data)
            logger.info(
                "Response data structure: %s, keys: %s",
                type(data_data).__name__,
                list(data_obj.keys()) if data_obj is not None else "not a dict",
            )
        
        if not graph_entries:
            logger.warning(f"No graph data in response. Response keys: {list(data.keys())}")
            self.last_error = "no_data"
            return LluResult(ok=False, data=None, error="no_data", status=204)
        
        logger.debug("Found graph data entries: %d", len(graph_entries))
        

        if graph_entries:
            logger.debug("First graph entry keys: %s", list(graph_entries[0].keys()))
        
        readings: List[GlucoseReading] = []
        skipped_count = 0
        for idx, entry in enumerate(graph_entries):
            

            ts = (
                entry.get("Timestamp") or 
                entry.get("FactoryTimestamp") or 
                entry.get("timestamp") or 
                entry.get("factoryTimestamp") or
                entry.get("Date") or
                entry.get("date") or
                entry.get("Time") or
                entry.get("time")
            )
            timestamp = self._parse_timestamp(ts)
            value = self._parse_glucose_value(entry)
            
            if timestamp is None or value is None:
                skipped_count += 1
                if idx < 5:
                    logger.warning(f"Skipping entry {idx}: timestamp={ts} (type: {type(ts)}), value={entry.get('ValueInMgPerDl') or entry.get('Value') or entry.get('value')}, entry keys: {list(entry.keys())[:10]}")
                continue
            

            timestamp = self._to_utc_naive(timestamp)
            

            is_high = self._coerce_bool(entry.get("isHigh", entry.get("IsHigh", False)))
            is_low = self._coerce_bool(entry.get("isLow", entry.get("IsLow", False)))
            

            if not is_high and not is_low:
                measurement_color = self._coerce_int(
                    entry.get("MeasurementColor", entry.get("measurementColor")),
                    default=-1,
                )
                if measurement_color in (2, 3):
                    is_high = True
                elif measurement_color == 1:
                    is_high = False
                    is_low = False
            

            if not is_high and not is_low:
                is_high = value > 180
                is_low = value < 70
            
            reading = GlucoseReading(
                timestamp=timestamp,
                value=value,
                is_high=is_high,
                is_low=is_low,
                trend=self._coerce_int(entry.get("Trend", entry.get("trend", 0))) or 0,
                notes="",
                source="librelinkup",
                device_id=str(target_id),
            )
            readings.append(reading)
        

        readings.sort(key=lambda r: r.timestamp)
        logger.debug("Fetched %d readings (skipped %d)", len(readings), skipped_count)
        return LluResult(ok=True, data=readings, error=None, status=200)
    except requests.exceptions.HTTPError as e:
        status = getattr(getattr(e, "response", None), "status_code", None)
        if status in (429, 430):
            self.last_error = "rate_limited"
            return LluResult(ok=False, data=None, error="rate_limited", status=status, retry_after=_retry_after_seconds(getattr(e, "response", None)))
        if status == 401:
            self.last_error = "unauthorized"
            return LluResult(ok=False, data=None, error="unauthorized", status=401)
        self.last_error = f"http_{status}" if status else "http_error"
        return LluResult(ok=False, data=None, error=self.last_error, status=status)
    except Exception as e:
        self.last_error = "exception"
        logger.debug("Failed to fetch glucose readings: %s", e)
        return LluResult(ok=False, data=None, error="exception", status=502)

def _parse_timestamp(self, ts: Any) -> Optional[datetime]:
    """Parse timestamp from various formats."""
    if not ts:
        return None
    if isinstance(ts, (int, float)):
        return self._epoch_to_utc_naive_seconds_or_ms(float(ts))
    if isinstance(ts, str):
        parsed = self._parse_timestamp_str(ts)
        if parsed is None:
            logger.warning(f"Could not parse timestamp: {ts}")
        return parsed
    return None

def _parse_glucose_value(self, entry: Mapping[str, Any]) -> Optional[float]:
    """Parse glucose value from entry.
    
    Prefers ValueInMgPerDl (from graphData) as it's always in mg/dL.
    Falls back to Value (which may be in mmol/L) if needed.
    """
    if "ValueInMgPerDl" in entry and entry["ValueInMgPerDl"] is not None:
        try:
            return float(entry["ValueInMgPerDl"])
        except (TypeError, ValueError):
            pass
    

    value_keys = [
        "valueInMgPerDl", "valueInMgPerDL", "ValueInMgPerDL",
        "Value", "value", "glucoseValue", "GlucoseValue",
        "glucose", "Glucose", "BGValue", "bgValue"
    ]
    
    for key in value_keys:
        if key in entry and entry[key] is not None:
            try:
                value = float(entry[key])


                if value < 20 and key != "ValueInMgPerDl":
                    value = value * 18.0182
                return value
            except (TypeError, ValueError):
                continue
    
    return None`

@Mynuggets-dev
Copy link
Copy Markdown

Mynuggets-dev commented Apr 16, 2026

image i use for my app now, using libre2

@Mynuggets-dev
Copy link
Copy Markdown

Mynuggets-dev commented Apr 16, 2026

Has anyone actually managed to log insulin doses into Libre 3 via API?

I have kind of a niche goal, and I am hoping someone here has already gone down this rabbit hole.

What I want is pretty simple in theory:

  • tap a Siri Shortcut on my iPhone
  • log something like Rapid-Acting Insulin, 4 units, optional note
  • have that show up in the Abbott / Libre 3 ecosystem the same way a manually entered dose does
  • ideally in a place my doctor can also see

I am not trying to replace the sensor data side. I only want a faster way to log insulin doses than opening the app and tapping through the note flow every time.

What I found so far

I spent quite a bit of time inspecting the iPhone app traffic and digging through community projects.

Here is what I was able to confirm:

  • The Libre 3 iOS app definitely supports manual note/logbook entries.
  • The web side seems much more limited, and I cannot access my own logbook there.
  • The iPhone app appears to send note / insulin data to:

POST https://api-c-de.libreview.io/lsl/api/measurements

  • The payload includes things like:

    • UserToken
    • device metadata
    • insulinEntries
    • genericEntries with com.abbottdiabetescare.informatics.customnote
    • links to recent glucose record numbers / timestamps

I also found another real endpoint that appears to bind the account/device/sensor state:

PUT /lsl/api/nisperson

That call accepts things like active sensor, name, DOB, token, etc.

What I tried

I tried to recreate the app behavior as closely as I could:

  • used a real current UserToken
  • used the real device metadata
  • used the real active sensor serial
  • called the nisperson binding endpoint successfully
  • matched the observed recordNumber patterns from real uploads
  • linked the synthetic note to recent glucose timestamps / glucose record numbers
  • sent both insulin and custom note entries together

The frustrating part is:

  • the server returns success
  • the API calls look valid
  • but the synthetic note still does not show up in the Libre 3 app logbook

So I am stuck in this weird state where it looks like I found the right API, but I am still missing some hidden piece.

Where I am stuck

At this point I do not know whether the missing piece is:

  • some local app database/state that also has to be updated
  • another hidden endpoint before/after the measurements upload
  • some signature/integrity field
  • another rule around record numbers or sequencing
  • or whether Abbott silently accepts third-party writes but does not surface them the same way first-party app writes are surfaced

I also checked the older sharing / LibreLinkUp-style APIs, but they did not give me a useful way to validate whether the synthetic note was really stored in a doctor-visible way.

What I am hoping someone here knows

If anyone has actually done this, I would love to hear about it.

Specifically:

  1. Has anyone successfully created insulin / note entries in Libre 3 using their own code?

  2. If yes, did those entries actually show up in:

    • the Libre 3 app?
    • LibreView?
    • your doctor / clinic view?

  3. Did you use:

    • direct API calls?
    • Juggluco?
    • another third-party uploader?
    • UI automation?

  4. If you got it working, what was the missing piece?

  5. Has anyone confirmed that third-party uploaded insulin amounts really become visible inside Abbott's ecosystem for Libre 3?

I am mainly looking for real-world experience from someone who has actually tried to get this working, not just guesses.

If you have done anything similar, even partially, I would really appreciate hearing what worked and what did not.

^ and https://diabetes.mynuggets.dev/
i have global fallback cause AU based, few fixes and that will work other regions
( could work global )

@Mynuggets-dev
Copy link
Copy Markdown

Mynuggets-dev commented Apr 16, 2026

@Mynuggets-dev Great that you exposed your scripting with others with your website! It looks good. Nice example for other developers. I hope one day I will provide machine-learned alerting service to others. At the moment I am too busy with other things. Keep going!

I’ve shared a simplified client API example I’ve been using.
On my side I run a more complete setup with additional layers for reliability, including an admin panel and data storage for analytics (I log readings roughly every minute for visualisation purposes).
I also have the ability to adjust ranges/alerts within my own tooling for testing and experimentation.
This is purely a personal project and not intended for medical use or as a clinical decision tool. Anyone building on this should implement their own safeguards and validation.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment