Database Schema

info

AquaGen API uses Azure Cosmos DB (NoSQL) as its primary database, organized in two databases: standard-categories (production data) and test (legacy data).

Cosmos DB Containers

industries

Database: standard-categories
Partition Key: industryId
Purpose: Stores all client/industry metadata and configuration.

Key fields:

{
  "industryId": "IND001",
  "industryName": "Acme Water Works",
  "units": [
    {
      "unitId": "unit-123",
      "unitName": "Main Water Meter",
      "standardCategoryId": "source",
      "subCategoryId": "municipal",
      "isDeployed": true
    }
  ],
  "categories": {
    "source": {
      "standardCategoryId": "source",
      "enabled": true,
      "subCategories": []
    }
  },
  "standardCategoriesView": {
    "source": {
      "standardCategoryId": "source",
      "subCategories": []
    }
  },
  "categoriesV2": {},
  "meta": {
    "timezone": "Asia/Kolkata",
    "shift": { "startAt": "06:00:00", "endAt": "18:00:00" },
    "siUnit": { "source": "KL", "energy": "kWh" },
    "logo": "https://storage.blob.core.windows.net/...",
    "autoReportConfigs": []
  }
}

users

Database: standard-categories
Partition Key: industryId
Purpose: Stores user accounts linked to industries.

Key fields:

{
  "userId": "IND001_john",
  "industryId": "IND001",
  "email": "john@example.com",
  "username": "john",
  "passwordHash": "...",
  "role": "user",
  "permissions": [],
  "phoneNo": "+919999999999",
  "emailIds": ["john@example.com"],
  "fcmTokens": ["firebase_token_..."]
}

devices_data

Database: standard-categories
Partition Key: date
Purpose: Raw IoT device readings. Each document is one reading for one unit on one date.

Key fields:

{
  "id": "2025-08-07T05:55:32Z_FG23462F",
  "unitId": "FG23462F",
  "industryId": "I2319",
  "summedValue": 9253,
  "value": 0,
  "createdOn": "2025-08-07T05:55:32Z",
  "deviceId": "H_care_4",
  "flowFactor": 1000,
  "meterType": "RS485",
  "date": "07082025"
}

warning

The devices_data container is partitioned by date. Queries that do not include a date filter will trigger expensive cross-partition scans. Always use processed_data for range-based report queries.

processed_data

Database: standard-categories
Purpose: Pre-aggregated data per unit — stored at day, month, and year granularity. Used for report generation instead of scanning raw devices_data.

The document id encodes the unit, period, and granularity — e.g. PI216_07-2026_FG27200F_MONTHLY.

Key fields:

{
  "id": "PI216_07-2026_FG27200F_MONTHLY",
  "unitId": "FG27200F",
  "industryId": "PI216",
  "month": "07/2026",
  "tag": "07/2026_MONTHLY",
  "value": 0,
  "daily": [
    {
      "x": "01/07/2026",
      "y": null,
      "ir": null,
      "fr": null,
      "nonZero": 0,
      "offlineCount": 0,
      "offlineDuration": 0,
      "offlineValue": 0
    }
  ],
  "meta": {
    "ir": null,
    "fr": null,
    "nonZero": 0,
    "offlineCount": 0,
    "offlineDuration": 2880,
    "offlineValue": 576
  }
}

daily array contains one entry per day of the month (x = date, y = aggregated value)
offlineDuration is in minutes, offlineValue is the number of offline readings (288 = full day offline at 5-min intervals)
Same structure is used for daily (DATE) and yearly (YEARLY) granularity with date/year fields instead of month

notification

Database: standard-categories
Purpose: Notification history per industry. Records alert triggers, device status events, and system messages with read/delivered tracking.

Key fields:

{
  "id": "b401c0e6-40b3-11ee-8e89-00155d98dd6c",
  "industryId": "I2276",
  "unitId": "FG22580F",
  "createdOn": "2023-08-22T06:18:25Z",
  "date": "22-08-2023",
  "receivedAt": "2023-08-22T06:18:25Z",
  "value": "online",
  "type": "device_status",
  "isRead": false,
  "isDelivered": false,
  "meta": {
    "offlineThreshold": 3600
  }
}

alerts_config

Database: standard-categories
Purpose: Alert rule configuration per industry. One document per industry containing per-user notification permissions and per-unit alert thresholds.

Key fields:

{
  "id": "PI216",
  "industryId": "PI216",
  "permissions": {
    "mobileEnabled": false,
    "customUsers": {
      "PI216_DEFAULT_PI216_3": {
        "permissions": ["unit_threshold", "monthly_threshold", "device_status", "quality_threshold", "..."],
        "channels": ["email", "sms", "whatsapp", "fcm"]
      }
    },
    "groups": {},
    "roles": {}
  },
  "units": {
    "TFG0035F": {
      "alertEnabled": {
        "daily_threshold": true,
        "monthly_threshold": true,
        "flow_rate": true,
        "stable_flow": false,
        "hourly": false,
        "consumption": false
      },
      "configs": {
        "unitThreshold": 15000,
        "monthlyThreshold": 30000,
        "maxFlowRate": 50,
        "minFlowRate": 10,
        "hourlyThreshold": 10,
        "stableFlowWindowMinutes": 60,
        "stableFlowAlertWindow": 60
      },
      "isDeployed": true,
      "standardCategoryId": "SOURCE_CATEGORY"
    },
    "FG26518Q": {
      "alertEnabled": { "quality_threshold": true },
      "params": {
        "pH": { "alertEnabled": false, "lt": 20, "ht": 80, "min": 0, "max": 14 },
        "TDS": { "alertEnabled": true, "lt": 100, "ht": 1000, "min": 0, "max": 1280 }
      },
      "isDeployed": true,
      "standardCategoryId": "QUALITY_CATEGORY"
    },
    "TFG25245E": {
      "alertEnabled": {
        "energy_threshold": true,
        "energy_daily_threshold": true,
        "energy_monthly_threshold": true
      },
      "configs": { "unitThreshold": 1000, "monthlyThreshold": 10000 },
      "params": {
        "kwh": { "alertEnabled": true, "lt": 10, "ht": 7627, "min": 0, "max": 10000 },
        "v1": { "alertEnabled": true, "lt": 230, "ht": 280, "min": 150, "max": 300 }
      },
      "isDeployed": true,
      "standardCategoryId": "ENERGY_CATEGORY"
    },
    "FG26841L": {
      "alertEnabled": { "level_exceed": true },
      "configs": { "maxCapacity": 1000, "lowThreshold": 10, "highThreshold": 90 },
      "isDeployed": true,
      "standardCategoryId": "STOCK_CATEGORY"
    },
    "TFG0029B": {
      "alertEnabled": { "borewell_threshold": true },
      "configs": { "unitThreshold": 1000 },
      "isDeployed": true,
      "standardCategoryId": "GROUND_WATER_LEVEL"
    }
  }
}

permissions.customUsers maps userId → alert types the user is subscribed to + delivery channels
units maps unitId → alertEnabled flags, numeric configs (thresholds), and for quality/energy units a params map with per-parameter low/high thresholds (lt/ht) and value range (min/max)

standard_categories_master

Database: standard-categories
Purpose: Reference data for device categories. Defines the standardCategoryId, display names, SI units, and icon mappings for all recognized device types.

{
  "id": "SOURCE_CATEGORY",
  "standardCategoryId": "SOURCE_CATEGORY",
  "displayName": "Source Category",
  "alertName": "Flow",
  "siUnit": "kL",
  "order": 1,
  "icon": "https://firebasestorage.googleapis.com/..."
}

tip

standard_categories_master documents are loaded into CachedData at startup and kept in memory for the lifetime of the process. This avoids repeated database reads for every request that needs category metadata.

processed_raw_data

Database: standard-categories
Purpose: Processed satellite/sensor data per unit per day (Lake Pulse). Each document contains aggregated water quality readings (turbidity, chlorophyll, etc.) with per-sample spatial coordinates and raw model outputs.

The document id encodes the industry, date, unit, and tag — e.g. DEMO221249_2025-02-11_DEMO_RK_LAKE.

Key fields:

{
  "id": "DEMO221249_2025-02-11_DEMO_RK_LAKE",
  "unitId": "DEMO_RK_LAKE",
  "industryId": "DEMO221249",
  "tag": "DAILY",
  "date": "11/02/2025",
  "cloudCover": 0.001635,
  "sampleCount": 424,
  "data": [
    {
      "turbidity": 7.506903006466207,
      "chlorophyll": 0.061330703874810995,
      "latitude": 12.825646597331527,
      "longitude": 77.50511403956367,
      "meta": {
        "bga": 0.5526695251464844,
        "chl": 0.34164589643478394,
        "ndwi": -0.16981132328510284,
        "tss": 0.3647541105747223,
        "turbidity": 0.5984455943107605
      }
    }
  ]
}

data array contains one entry per spatial sample point — each with latitude/longitude, derived water quality values (turbidity, chlorophyll), and raw model scores in meta (bga, chl, ndwi, tss, turbidity)
sampleCount is the total number of valid pixel samples processed from the satellite image
cloudCover is the fraction of the scene obscured by cloud (0.0–1.0) — used to determine data reliability
tag is always DAILY for this container

additional_feature

Database: standard-categories
Purpose: Feature-specific configuration documents per industry. Each document has a type field identifying the feature it configures.

Supported type values:

Type	Purpose
`waterBalance`	Water balance graph nodes, edges, and summary configuration
`waterBalanceLeakage`	Leakage detection configuration
`scadaWaterBalance`	SCADA water balance diagram
`adminScadaWaterBalance`	Admin SCADA water balance configuration
`alerts`	Additional alert configuration
`executiveSummary`	Executive summary dashboard configuration
`roEfficiencyLog`	RO efficiency log settings
`GWI`	Groundwater Intelligence configuration
`HMI_CONFIG`	HMI (Human Machine Interface) configuration
`webHMI`	Web HMI configuration
`map_view`	Map view settings
`water_ratio`	Water ratio standards configuration
`BATHY_DATA`	Bathymetry data (Lake Pulse)
`LAKE_BOUNDARY`	Lake boundary polygon (Lake Pulse)
`SATELLITE_MODEL`	Satellite model configuration (Lake Pulse)

industry_metrics

Database: standard-categories
Purpose: Latest data snapshot for all units of an industry. One document per industry, updated on every device reading. Used for online/offline detection and latest value lookups without querying devices_data.

Key fields:

{
  "id": "I2319_Metrics",
  "industryId": "I2319",
  "units": {
    "FG23455F": {
      "unitId": "FG23455F",
      "lastUpdated": "2026-07-17T13:07:15Z",
      "metadata": {
        "summedValue": 379670,
        "value": 0
      }
    },
    "FG23456F": {
      "unitId": "FG23456F",
      "lastUpdated": "2026-07-17T13:07:16Z",
      "metadata": {
        "summedValue": 5039442,
        "value": 46
      }
    }
  }
}

One entry per unit under units, keyed by unitId
lastUpdated is used by formatters to determine online/offline status (unit is online if last update is within its dataUpdateFrequency window, default 1 hour)
metadata.value is the latest instantaneous reading; metadata.summedValue is the cumulative meter total

admin_app_logs

Database: standard-categories
Purpose: Audit trail for admin-facing write operations. One document per admin action — records who made the change, which API was called, and the full before/after state of the affected data.

Key fields:

{
  "id": "2026-07-23T14:56:05Z_87792056-8678-11f1-8819-f20fb0e3c067",
  "createdOn": "2026-07-23T14:56:05Z",
  "updatedOn": "2026-07-23T14:56:05Z",
  "date": "23/07/2026",
  "month": "07/2026",
  "userName": "Varun M Chakravarthy",
  "api": "https://...southindia.azurewebsites.net/api/admin/units_meta/v2",
  "operation": "bulk_update_unit_meta",
  "industryId": "PI216",
  "beforeData": {
    "units": [
      { "unitId": "FG27199B", "unitName": "Borewell 1", "before": { "deviceOwner": "", "..." : "..." } }
    ]
  },
  "afterData": {
    "units": [
      { "unitId": "FG27199B", "unitName": "Borewell 1", "after": { "deviceOwner": "Customer", "..." : "..." } }
    ]
  }
}

id is composed of ISO timestamp + UUID
operation is a human-readable name for the admin action (e.g. bulk_update_unit_meta, update_industry_meta)
beforeData / afterData capture the full state of the affected records before and after the operation — used for change review and rollback

aqua_gpt_logs

Database: standard-categories
Purpose: Audit and analytics log for AquaGPT (AI assistant) sessions. One document per prompt/response exchange.

Key fields:

{
  "id": "2026-07-23T15:32:31Z_I2443_178482075129702",
  "industryId": "I2443",
  "userId": "DEMOAQ_utkarsh@fluxgentech.com",
  "date": "23/07/2026",
  "sessionId": "DEMOAQ_utkarsh@fluxgentech.com_nt1hgi1ay",
  "prompt": "Analyse my water balance and share insights?",
  "response": "Here's your **Water Balance Analysis**...",
  "timeTaken": 49.698,
  "isLiked": null,
  "runId": "2a1be57b-06f6-4d55-bbaa-ac583dea1c2c",
  "sourceHost": "pre-prod-aquagen.web.app",
  "version": "v3"
}

id is composed of ISO timestamp + industryId + epoch milliseconds
sessionId groups multiple exchanges in the same conversation session (format: userId_sessionKey)
timeTaken is the end-to-end response latency in seconds
isLiked is true / false / null depending on whether the user gave a thumbs up, thumbs down, or no feedback
version tracks the AquaGPT prompt/pipeline version used to generate the response

global_logs

Database: standard-categories
Purpose: System-level operation log for cross-cutting events that don't belong to a single industry flow. Currently used for PLC write command audit — records every Modbus command sent to an edge device via the bridge module.

Key fields:

{
  "id": "plc_write_I2425_1769763045_Guwahati_4c7b2d83",
  "type": "PLC_WRITE",
  "userId": "I2425_1769763045_Guwahati",
  "industryId": "I2425",
  "edgeDeviceId": "EDI2425HMI",
  "moduleId": "bridgemodule",
  "frames": [
    { "id": "motor6_on", "data": "01 06 00 95 00 01", "ackData": "01 03 00 95" }
  ],
  "asset": "Filter Feed Pump 2",
  "command": "on",
  "success": true,
  "timestamp": "2026-07-09T08:06:41Z"
}

type identifies the log category — currently PLC_WRITE; may expand to other types
edgeDeviceId and moduleId identify the IoT Edge device and the specific bridge module that executed the command
frames is the raw Modbus frame array — each entry has the command bytes (data) and the expected acknowledgement bytes (ackData)
command is the human-readable action name (e.g. on, off, set_speed)
success records whether the edge device acknowledged the command correctly

industry_checkpoint

Database: standard-categories
Purpose: Incremental patch log for industry configuration changes. Each document records a single field-level mutation applied to an industry document — used to track who changed what and when, without storing the full before/after snapshot.

Key fields:

{
  "id": "395ea752-4b1d-4f3d-9485-cc9339523527",
  "industryId": "I2373",
  "userId": "I2373_1742892424_EmersonProcess",
  "createdOn": "2026-06-30T13:22:49Z",
  "op": "set",
  "path": "/categories/SOURCE_CATEGORY/subCategories/1/alertEnabled",
  "value": false
}

op is the patch operation — set (write a value), unset (remove a field), or append (add to an array)
path is a JSON Pointer-style string identifying the exact field within the industry document that was changed
value is what the field was set to — only present for set and append ops
Documents are append-only — no updates or deletes. The full mutation history can be reconstructed by replaying all checkpoints for an industry in createdOn order

ip_logs

Database: standard-categories
Purpose: Login IP tracking. One document per unique IP address per user. Records geolocation data resolved at login time and classifies the access as internal or external.

Key fields:

{
  "id": "89.187.163.170_DEMO221249_DEFAULT_demo",
  "user_id": "DEMO221249_DEFAULT_demo",
  "industry_id": "DEMO221249",
  "created_on": "2025-04-05T17:55:58.303850",
  "ip": "89.187.163.170",
  "city": "Singapore",
  "region": "Singapore",
  "country": "SG",
  "type": "External"
}

id is composed of ip + userId — one document per unique IP per user; upserted on repeat logins from the same IP
type is "Internal" for Fluxgen IPs, "External" for all other addresses
city, region, country are resolved at login time via an IP geolocation lookup

units_meta

Database: standard-categories
Purpose: Physical installation metadata for individual units — IoT device details, sensor specs, network config, and installation notes. Kept separate from the main industries document to avoid bloating the industry record.

Key fields:

{
  "id": "FG27149L_PI216",
  "unitId": "FG27149L",
  "industryId": "PI216",
  "createdOn": "2026-07-23T13:54:01Z",
  "iot": {
    "deviceOwner": "Customer",
    "deviceMake": "Teltonika",
    "deviceMacId": "AA:BB:CC:DD:EE:FF",
    "deviceSerialNo": "SN001",
    "deviceImei": "123456789012345",
    "slaveId": ""
  },
  "network": {
    "networkOwner": "Customer",
    "networkProvider": "Airtel",
    "networkCondition": "Low",
    "simNumber": "9876543210",
    "ssid": null,
    "wifiPassword": null
  },
  "sensor": {
    "sensorOwner": "Customer",
    "partnerName": "Deewan",
    "sensorType": "Piezo Sensor",
    "sensorModel": "Ultrasonic",
    "sensorMake": "Deewan",
    "supplyDate": "2024-01-15",
    "sensorSlNo": "SEN001",
    "diameter": null,
    "kFactor": null,
    "sensorHeight": null,
    "maxWaterHeight": null,
    "borewellDepth": null,
    "cableLength": null
  },
  "installation": {
    "installationOwner": "Customer",
    "voltageVariation": "Yes",
    "earthing": "Yes",
    "surgeProtectors": "Yes",
    "flowType": "",
    "nrvInstalled": "",
    "upstream5D": "",
    "downstream5D": "",
    "uBendInstalled": "",
    "groundRings": ""
  },
  "general": {
    "remarks": "",
    "problemsFaced": ""
  }
}

id is unitId + "_" + industryId
The iot section covers the gateway/logger device (Teltonika, Peplink, etc.) and its identifiers
The sensor section covers the physical measurement sensor (flow sensor, pressure transducer, etc.)
The installation section captures field installation quality factors (earthing, NRV, pipe diameter alignment)
This data is written by the admin app via the units_meta endpoint and is not used in any real-time data processing path — it is reference/audit data only

token_logs

Database: standard-categories
Purpose: Active JWT session tracking. One document per user session (keyed by industryId + userId + deviceId + baseUrl). Used by TokenService.validate_token() on every request to confirm the token is still active.

Key fields:

{
  "id": "I2439_I2439_1778654435_Aquazig_47d86edadf8a39671cdc912898c4bf5a_ade9aa26",
  "industryId": "I2439",
  "userId": "I2439_1778654435_Aquazig",
  "accessToken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "refreshToken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "loginType": "DEFAULT",
  "ipAddress": "106.222.243.217:58629",
  "deviceId": "47d86edadf8a39671cdc912898c4bf5a",
  "baseUrl": "https://pre-prod-aquagen.web.app",
  "isActive": true,
  "status": "ACTIVE",
  "createdAt": "2026-05-27T09:27:52",
  "lastActivity": "2026-05-27T13:00:59"
}

id is composed of industryId + userId + deviceId + md5(baseUrl)[:8] — one active session per device+app combination
status / isActive are set to INACTIVE / false on logout or token revocation
lastActivity is updated on each successful token validation

DatabaseSupporter Class

app/database/database_supporter.py

info

The DatabaseSupporter class provides 200+ static methods as a unified data access layer. All services interact with Cosmos DB exclusively through this class.

Key Patterns

Single document queries:

industry = DatabaseSupporter.get_industry_details_by_id("IND001")
user = DatabaseSupporter.get_user_details_by_user_id("IND001_john")

Date-range device data:

data = DatabaseSupporter.get_device_data_by_date_range(
    industry_id="IND001",
    start_date="01/11/2024",
    end_date="30/11/2024",
    unit="unit-123"
)

Parallel queries with ThreadPoolExecutor:

For report generation covering multiple dates or units, DatabaseSupporter uses concurrent.futures.ThreadPoolExecutor to run Cosmos DB queries in parallel, reducing latency significantly for large date ranges.

with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(query_func, date) for date in dates]
    results = [f.result() for f in futures]

Upsert pattern:

DatabaseSupporter.create_user(user_data)  # Uses upsert_item internally

pandas DataFrame processing:

Raw Cosmos DB results are converted to pandas DataFrames for aggregation, then back to dicts:

dataframe = pd.DataFrame(raw_results)
dataframe = dataframe.replace({np.nan: None})
return dataframe.to_dict(orient="records")

Cosmos DB Query Patterns

Cross-Partition Queries

Most queries use enable_cross_partition_query=True for flexibility, but partitioned queries (e.g., devices_data by date) are used where performance is critical.

Cost Optimization

tip

The following patterns are applied consistently to reduce Cosmos DB RU consumption:

processed_data is used for monthly/range reports instead of scanning raw devices_data
CachedData holds standard_categories_master in memory to avoid repeated DB reads
Industry data is cached in current_user context per request (loaded once via load_user)

Cosmos DB Containers​

industries​

users​

devices_data​

processed_data​

notification​

alerts_config​

standard_categories_master​

processed_raw_data​

additional_feature​

industry_metrics​

admin_app_logs​

aqua_gpt_logs​

global_logs​

industry_checkpoint​

ip_logs​

units_meta​

token_logs​

DatabaseSupporter Class​

Key Patterns​

Cosmos DB Query Patterns​

Cross-Partition Queries​

Cost Optimization​

Cosmos DB Containers

industries

users

devices_data

processed_data

notification

alerts_config

standard_categories_master

processed_raw_data

additional_feature

industry_metrics

admin_app_logs

aqua_gpt_logs

global_logs

industry_checkpoint

ip_logs

units_meta

token_logs

DatabaseSupporter Class

Key Patterns

Cosmos DB Query Patterns

Cross-Partition Queries

Cost Optimization