Utilities & Enums Reference

info

Reference for the shared utilities, constants, enums, data classes, and request marshalling patterns used across the AquaGen API codebase. These are not feature-specific — they are building blocks used throughout routes, services, formatters, and batch processors.

DateTimeUtil (`app/util/date_time_util.py`)

warning

Never use datetime.now() directly or hardcode timezone offsets — always use DateTimeUtil for all timezone-aware date/time operations.

Timezone resolution — get_timezone(t_zone):
If t_zone is not passed, reads g.timezone set by the JWT user loader on every authenticated request. Falls back to 'Asia/Kolkata' for public routes where no user context exists.

Method	Signature	Purpose
`convert_date_from_local_to_utc`	`(value, input_format, t_zone)`	Converts a local datetime string to UTC. Default format: `%Y-%m-%dT%H:%M:%SZ`. Used by routes to convert `DD/MM/YYYY` request dates before Cosmos DB queries.
`convert_date_from_utc_to_string`	`(value, output_format, input_format, t_zone)`	Converts a UTC ISO string to a formatted local date string. Default output: `%d/%m/%Y`.
`convert_date_time_from_utc_to_string`	`(value, output_format, t_zone)`	Same as above but includes time. Default output: `%d/%m/%Y %I:%M %p`.
`convert_date_from_utc_to_string_with_round_off`	`(value, output_format, input_format, nearest_minute, t_zone)`	Converts UTC to local, then rounds to nearest `nearest_minute` seconds (default `300` = 5 min).
`convert_date_from_utc_to_string_with_round_down`	`(value, ..., nearest_minute, t_zone)`	Same but always rounds down (floor).
`convert_date_from_utc_to_string_with_round_up`	`(value, ..., nearest_minute, t_zone)`	Same but always rounds up (ceiling).
`strptime`	`(value, time_format, t_zone)`	`datetime.strptime` with automatic timezone attachment. Handles both aware and naive datetime strings via a try/catch fallback.
`get_today_date`	`(server, t_zone)`	Returns `datetime.now()` when `server=True`. When `False`, converts now to UTC in the industry timezone.
`get_current_time`	`(output_format, t_zone)`	Returns current local time as a formatted string.
`get_n_dates`	`(date, count)`	Returns a list of `count` dates counting back from `date - 2 days`. Used for quality week graphs.
`get_dates_between_dates`	`(start_date, end_date)`	Returns every calendar day between two `datetime` objects as `DD/MM/YYYY` strings (inclusive).
`convert_seconds`	`(seconds)`	Converts a seconds integer to a human-readable string: `"X Days Y Hours Z Minutes"`.

Usage example:

from app.util.date_time_util import DateTimeUtil

date1_utc = DateTimeUtil().convert_date_from_local_to_utc(
    f'{date1}T{shift_start}', '%d/%m/%YT%H:%M:%S'
)

Constants (`app/constants/constants.py`)

note

Constants is a class with only class-level attributes — no instantiation needed. Import and reference directly.

Constant	Value	Where used
`max_daily_device_data_per_day`	`288`	Maximum expected 5-minute readings per day (24h × 12)
`daily_min_data_threshold`	`284`	Minimum readings for a day to be considered "complete"
`OFFLINE_THRESHOLD_HOUR`	`15` min	Device considered offline in hour view
`OFFLINE_THRESHOLD_DAY`	`60` min	Device considered offline in day view
`OFFLINE_THRESHOLD_MONTH`	`1800` min	Device considered offline in month view
`DEFAULT_USER_LIMIT`	`5`	Default max users per industry
`DEFAULT_SSO_USER_LIMIT`	`10`	Default max SSO users per industry
`DEFAULT_ALERTS_LIMIT`	`10`	Default max alerts per query
`RO_PLANT_EFFICIENCY_THRESHOLD`	`50`	Below this % → RO plant is inefficient
`RO_PLANT_SUDDEN_DROP_THRESHOLD`	`15`	Drop greater than this % in one period → sudden drop alert
`RO_PLANT_HEALTH_SCORE_POOR`	`50`	Health score below this → "Poor"
`RO_PLANT_HEALTH_SCORE_FAIR`	`70`	Health score below this → "Fair"
`MONTHLY_BATCH_PROCESS_EXCLUDED_INDUSTRY_IDS`	List	Industries skipped by the automated monthly batch shell script
`TEST_INDUSTRY_IDS`	List	Demo/test industries — excluded from production metrics
`ADMIN_ALERT_TYPES`	List	Alert types that are admin-only: `device_status`, `no_flow_n_days`, `no_level_variation_n_days`, `water_balance_leakage`
`PUBLIC_ROUTES`	List	Exact paths that skip JWT auth entirely
`PUBLIC_PREFIXES`	List	Path prefixes that skip JWT auth (Swagger UI, virtual routes, batch trigger)
`quality_si_units`	Dict	Display units for quality parameters: `{"pH": "", "TDS": "ppm", "COD": "mg/L", ...}`

Usage example:

from app.constants.constants import Constants

if elapsed_minutes > Constants.OFFLINE_THRESHOLD_DAY:
    mark_offline()

Enums (`app/enum/`)

info

All string literals used for category types, report types, permissions, and statuses are defined as Python Enum classes. Always reference enum values rather than hardcoding strings.

StandardCategoryType — `standard_category_type_enum.py`

Maps sensor categories to their Cosmos DB string identifiers. Used in batch processing dispatch, formatter selection, and device data routing.

Member	Value
`SOURCE_CATEGORY`	`'SOURCE_CATEGORY'`
`ENERGY_CATEGORY`	`'ENERGY_CATEGORY'`
`STOCK_CATEGORY`	`'STOCK_CATEGORY'`
`GROUND_WATER_LEVEL`	`'GROUND_WATER_LEVEL'`
`QUALITY_CATEGORY`	`'QUALITY_CATEGORY'`
`VIRTUAL_CATEGORY`	`'VIRTUAL_CATEGORY'`

StandardCategoryType.all_categories() returns all values as a flat list.

CategoryType — `category_type.py`

Same categories but with camelCase member names — used when keying into current_user['industryData']['categories'].

Member	Value
`sourceCategory`	`'SOURCE_CATEGORY'`
`energyCategory`	`'ENERGY_CATEGORY'`
`stockCategory`	`'STOCK_CATEGORY'`
`groundWaterLevel`	`'GROUND_WATER_LEVEL'`
`qualityCategory`	`'QUALITY_CATEGORY'`
`rainWaterCategory`	`'RAIN_WATER_CATEGORY'`
`virtualCategory`	`'VIRTUAL_CATEGORY'`

Why two category enums?

StandardCategoryType uses UPPER_CASE members matching Cosmos DB document values. CategoryType uses camelCase members matching the current_user dict keys. Use StandardCategoryType in batch processing and formatters; use CategoryType when accessing industryData.categories.

PatternType / PatternTypeV2 — `pattern_type.py`

Controls aggregation granularity in device data endpoints.

PatternType (v1 — GET /api/user/deviceData):

Member	Value
`hour`	`'HOUR'`
`date`	`'DATE'`
`month`	`'MONTH'`
`year`	`'YEAR'`

PatternTypeV2 (v2 — GET /api/user/deviceDataV2):

Member	Value	`graph_key()`
`day`	`'DAY'`	`'hourly'`
`month`	`'MONTH'`	`'daily'`
`year`	`'YEAR'`	`'monthly'`
`custom`	`'CUSTOM'`	`'daily'`

graph_key() returns the array field name used in the formatted response for that pattern — e.g. DAY view returns data in hourly[].

ReportType — `report_type.py`

Member	Value
`DAILY`	`'daily'`
`MONTHLY`	`'monthly'`
`HOURLY`	`'hourly'`
`GRANULAR`	`'granular'`
`WEEKLY`	`'weekly'`
`CUSTOM`	`'custom'`
`SUMMARY`	`'summary'`
`DAILY_RANGE`	`'daily_range'`
`MONTHLY_RANGE`	`'monthly_range'`
`AQUA_GPT`	`'aqua_gpt'`

ServiceType — `service_type.py`

Maps the service request parameter to a report class. Also provides standard_category_id() — returns the CategoryType member that corresponds to this service (e.g. ServiceType.LEVEL.standard_category_id() → CategoryType.stockCategory).

Value	Maps to
`water`	`FlowReport`
`rain_water`	`RainWaterReport`
`custom_water`	`CustomFlowReport`
`custom_withdrawal`	`CustomWithdrawalReport`
`level`	`LevelReport`
`energy`	`EnergyReport`
`borewell`	`BorewellReport`
`quality`	`QualityReport`
`summary`	`SummaryReport`
`consolidated`	`ConsolidatedReport`
`ground_water_withdrawal`	`GroundWaterWithdrawalReport`
`alert`	`AlertReport`
`custom_level`	`CustomLevelReport`
`custom_quality`	`CustomQualityReport`
`custom_energy`	`CustomEnergyReport`
`smart_city`	`SmartCityReport`
`water_balance`	`WaterBalanceReport`
`executive`	`ExecutiveReport`

ReportFormat — `report_format.py`

Member	Value	Helper methods
`PDF`	`'pdf'`	`is_pdf()` → `True`, `is_html()` → `True`
`XLSX`	`'xlsx'`	`download_images()` → `True`
`CSV`	`'csv'`	—
`HTML`	`'html'`	`is_html()` → `True`

note

is_html() returns True for both PDF and HTML because PDF generation first renders the Jinja2 HTML template before passing it to xhtml2pdf.

Permission — `permission.py`

Controls feature access per user. Checked via @permission_required(Permission.X) decorator or permission_util.is_authorized(Permission.X).

Value	Feature
`ACCOUNT_SETTINGS`	Manage account settings
`SUPER_USER`	Super-user access
`ALERTS`	View and manage alerts
`LANDING_PAGE`	Access dashboard landing page
`MANUAL_NODE_ENTRY`	Enter data for virtual nodes manually
`WATER_BALANCE`	Water balance graph and analysis
`AQUAGPT`	AquaGPT AI assistant
`EFFICIENCY`	RO efficiency tracking
`UWI`	Urban Water Intelligence
`RWI`	Recycled Water Intelligence
`DAILY_SUMMARY`	Daily summary reports
`ENERGY_ALERT`	Energy-specific alerts
`WATER_NEUTRALITY`	Water Neutrality Index
`WATER_RATIO`	Water ratio standards
`USER`	User management
`ADMIN`	Admin panel access
`DISABLE_ENERGY_TOGGLE_BUTTON`	Hide energy toggle in UI
`WRI`	Water Reuse Intelligence

NodeTypes — `node_types.py`

Used in water balance graph construction to distinguish physical meters from computed aggregate nodes.

Member	Value
`physicalNode`	`'PHYSICAL_NODE'`
`virtualNode`	`'VIRTUAL_NODE'`

ResponseStatus — `response_status.py`

Member	Value
`success`	`'success'`
`microsoft`	`'failed'`

Utility Functions (`app/util/`)

`number_util.py`

Function	Purpose
`trunc(value, ndigits=2)`	Truncates (not rounds) a float to `ndigits` decimal places. Returns `value` unchanged if falsy. Used everywhere values are displayed to avoid floating-point rounding surprises.
`is_number(data)`	Returns `True` if `data` is `int`, `float`, or `None`.

from app.util.number_util import trunc

trunc(12.3456, 2)  # → 12.34  (truncated, not 12.35)
trunc(0, 2)        # → 0
trunc(None, 2)     # → None

Use trunc, not round

trunc truncates toward zero — it does not round. trunc(12.999, 2) → 12.99, not 13.00. Use this wherever sensor values are displayed to users.

`si_unit_util.py`

Function	Purpose
`get_si_unit(industry, category_id)`	Returns the display SI unit string for a category. Checks `industry['meta']['siUnit'][category_id]` first; falls back to `CachedData.standardCategoriesMap[category_id]['siUnit']`. `STOCK_CATEGORY` is remapped to `SOURCE_CATEGORY` before lookup.

`permission_util.py`

Function	Purpose
`is_authorized(permission)`	Checks `current_user['permissions']` for the given `Permission` enum value or list of values. If a list is passed, all permissions must be present. Returns `True`/`False`.

from flask_restx import marshal
from app.util.permission_util import is_authorized
from app.enum.permission import Permission
from app.models.error_models import GenericErrorModel

if not is_authorized(Permission.WATER_BALANCE):
    return marshal({'status': 'failed', 'message': 'Unauthorized'}, GenericErrorModel), 200

# Check multiple permissions — all must be present
if not is_authorized([Permission.ALERTS, Permission.SUPER_USER]):
    return marshal({'status': 'failed', 'message': 'Unauthorized'}, GenericErrorModel), 200

`filter_units_util.py`

Function	Purpose
`get_units_for_category(current_user, category_type, exclude_virtual=True)`	Returns a deduplicated flat list of unit IDs for a given `category_type` string from `current_user['industryData']['categories']`. Excludes virtual units by default.
`get_selected_units(data, unit_ids=None)`	Filters `data['subCategories']` to only include units present in `unit_ids`. Returns `data` unchanged if `unit_ids` is `None`.

from app.util.filter_units_util import get_units_for_category
from app.enum.category_type import CategoryType

unit_ids = get_units_for_category(
    current_user,
    CategoryType.sourceCategory.value,
    exclude_virtual=True
)

`shift_and_unit_count_util.py`

Function	Purpose
`get_shift_and_total_units(data, unit_ids=None)`	Returns a human-readable shift + unit count string: `"06:00 AM to 06:00 AM \| Total units : 12"`. Used in report headers and daily summary pages.

`calculation_util.py`

Function	Purpose
`level_calculation_w_percent(max_capacity, percent_remaining)`	Converts a percentage reading to an absolute level: `round(maxCapacity × (percent / 100), 3)`. Used in stock and water balance level computations.

Data Classes (`app/data_class/`)

info

Data classes are typed Python @dataclass objects used to pass structured, validated arguments between the route layer and services. They replace raw dict passing and make service constructors self-documenting. Use them when a service takes many arguments, the same bundle crosses multiple layers, or optional fields need meaningful defaults.

`ReportRequestData` — `data_class/report.py`

The primary DTO passed from the report route into ReportService and all report type classes.

@dataclass
class ReportRequestData:
    industry_id: str
    start_date: str
    end_date: str
    end_date_actual: str       # may differ from end_date for monthly/yearly reports
    report_type: ReportType
    serviceType: ServiceType
    report_format: ReportFormat
    timezone: str
    use_shift: bool
    division_factor: int = 1000
    unit_id: str = ''
    unit_ids: List[str] = None
    additional_data: object = None

`DeviceDataV2Args` — `data_class/device_data_v2_args.py`

Passed from the v2 device data route into DeviceDataV2FormatterBuilder and all v2 formatter classes. Wraps both UTC and local date representations so formatters never recompute timezone conversions.

@dataclass
class Date:
    utc: datetime
    local: datetime

@dataclass
class DeviceDataV2Args:
    date1: Date
    categoryID: str
    category: dict
    divisionFactor: int
    industryId: str
    type: PatternTypeV2        # DAY / MONTH / YEAR / CUSTOM
    dates: List[Date]
    sdId: StandardCategoryType
    date2: Date = None         # comparison date (optional)
    v2: bool = False

Other data classes follow the same pattern — see data_class/alert/, data_class/rainfall.py, and data_class/landingPage/ for further examples.

Request Marshalling & Parsing (`app/models/`)

info

Flask-RESTX uses two complementary mechanisms:

Request parsers (reqparse) — parse and validate incoming request arguments (headers, query params, POST form/JSON body)
Response models (Model + fields) — document and serialize outgoing responses in Swagger UI

Request Parsers (`reqparse`)

Request parsers extract and validate arguments from an incoming request. They are the Flask-RESTX equivalent of form validation.

from flask_restx import reqparse

parser = reqparse.RequestParser()
parser.add_argument('start_date', type=str, required=True,  location='headers', help='Start date DD/MM/YYYY')
parser.add_argument('end_date',   type=str, required=True,  location='headers')
parser.add_argument('unit_id',    type=str, required=False, location='headers', default='')
parser.add_argument('page',       type=int, required=False, location='args',    default=1)

`location` value	Where Flask-RESTX looks
`'headers'`	HTTP request headers
`'args'`	URL query string (`?key=value`)
`'json'`	JSON request body (`Content-Type: application/json`)
`'form'`	Form-encoded body (`Content-Type: application/x-www-form-urlencoded`)
`'files'`	Multipart file uploads

Parse inside a route handler:

def get(self):
    args = parser.parse_args()
    start_date = args['start_date']
    unit_id    = args.get('unit_id', '')

note

In AquaGen API, most parameters arrive as request headers (not query string or body). Always check an existing route to confirm the expected location before adding a new parser.

If a required=True argument is missing, Flask-RESTX automatically returns a 400 with a message from the help field — no manual checking needed.

POST Request Body (JSON)

For POST endpoints that receive a JSON body, use location='json':

post_parser = reqparse.RequestParser()
post_parser.add_argument('industry_id', type=str, required=True, location='json')
post_parser.add_argument('alert_type',  type=str, required=True, location='json')
post_parser.add_argument('threshold',   type=float, required=False, location='json', default=0.0)

Alternatively, document the expected body as a Model and use @namespace.expect():

AlertPayload = namespace.model('AlertPayload', {
    'industry_id': fields.String(required=True),
    'alert_type':  fields.String(required=True),
    'threshold':   fields.Float(default=0.0),
})

@namespace.route('/alert')
class AlertRoute(Resource):
    @namespace.expect(AlertPayload, validate=True)  # validates body against model
    def post(self):
        data = request.json
        industry_id = data['industry_id']
        ...

validate=True makes Flask-RESTX reject the request (400) if required fields are missing — and the model is displayed as the request body schema in Swagger.

Response Models (`Model` + `fields`)

Models document and serialize outgoing responses. Define them in app/models/, register them on the namespace, then attach via @namespace.response().

error_models.py

GenericErrorModel = Model('GenericErrorModel', {
    'message': fields.String,
    'status': fields.String,
})

success_models.py

GenericSuccessModel = Model('GenericSuccessModel', {
    'message': fields.String,
    'status': fields.String,
})

GenericSuccessDataModel = Model('GenericSuccessDataModel', {
    'status': fields.String(),
    'data': fields.Raw(),     # untyped — use for complex/variable response shapes
})

Registering and using a model in a route

from flask_restx import Namespace, Resource, marshal
from app.models.my_feature_models import MyFeatureResponseModel

myFeatureNamespace = Namespace('My Feature', path='/myFeature')

# Register the model on the namespace for Swagger
myFeatureNamespace.add_model(MyFeatureResponseModel.name, MyFeatureResponseModel)

@myFeatureNamespace.route('')
class MyFeatureRoute(Resource):
    @jwt_required()
    @myFeatureNamespace.response(200, 'Success', model=MyFeatureResponseModel)
    @myFeatureNamespace.response(1403, 'Missing Keys')
    def get(self):
        result = ...
        return marshal({'status': 'success', 'data': result}, MyFeatureResponseModel), 200

Flask-RESTX field types

Field	Use case
`fields.String`	Plain string
`fields.Integer`	Integer number
`fields.Float`	Float number
`fields.Boolean`	Boolean
`fields.Raw`	Untyped — any JSON value. Use for variable-shape data (nested dicts, arrays)
`fields.List(fields.Nested(Model))`	Typed array of nested model objects
`fields.Nested(Model)`	Inline nested model
`CustomDateFormat`	`dd-mm-yyyy` — defined in `app/util/custom_flask_restx_fields.py`
`CustomTimeFormat`	`HH:MM AM/PM` — same file
`CustomDateTimeFormat`	`dd-mm-yyyy HH:MM AM/PM` — same file

Custom field types (`app/util/custom_flask_restx_fields.py`)

The three custom fields extend fields.Raw and override format() to control serialization and the Swagger schema type hint:

class CustomDateFormat(fields.Raw):
    __schema_type__ = "dd-mm-yyyy"
    def format(self, value):
        return value.strftime('%d-%m-%Y')

class CustomTimeFormat(fields.Raw):
    __schema_type__ = "HH:MM AM/PM"
    def format(self, value):
        return value.strftime('%I:%M %p')

tip

Use these in models wherever a datetime object needs to be serialized to a display string rather than an ISO timestamp.

Marshalling with `marshal()`

marshal() is used directly in every route's return statement to serialize the response through the model. This is the standard pattern across AquaGen API:

from flask_restx import marshal
from app.models.error_models import GenericErrorModel
from app.models.success_models import GenericSuccessDataModel

# Success response
return marshal({'status': 'success', 'data': result}, GenericSuccessDataModel), 200

# Error response
return marshal({'message': 'Unauthorized', 'status': 'failed'}, GenericErrorModel), 401

Model nesting pattern

For complex responses, nest models rather than using fields.Raw — this gives Swagger accurate schema documentation:

AlertModel = Model('Alert', {
    'id': fields.String,
    'description': fields.Nested(AlertDescriptionModel)
})

AlertCategoryModel = Model('AlertCategory', {
    'meta': fields.Nested(AlertMetaModel),
    'alerts': fields.List(fields.Nested(AlertModel))
})

note

fields.Raw is appropriate only when the shape genuinely varies at runtime (e.g. data fields that differ per report type).

DateTimeUtil (app/util/date_time_util.py)​

Constants (app/constants/constants.py)​

Enums (app/enum/)​

StandardCategoryType — standard_category_type_enum.py​

CategoryType — category_type.py​

PatternType / PatternTypeV2 — pattern_type.py​

ReportType — report_type.py​

ServiceType — service_type.py​

ReportFormat — report_format.py​

Permission — permission.py​

NodeTypes — node_types.py​

ResponseStatus — response_status.py​

Utility Functions (app/util/)​

number_util.py​

si_unit_util.py​

permission_util.py​

filter_units_util.py​

shift_and_unit_count_util.py​

calculation_util.py​

Data Classes (app/data_class/)​

ReportRequestData — data_class/report.py​

DeviceDataV2Args — data_class/device_data_v2_args.py​

Request Marshalling & Parsing (app/models/)​

Request Parsers (reqparse)​

POST Request Body (JSON)​

Response Models (Model + fields)​

Registering and using a model in a route​

Flask-RESTX field types​

Custom field types (app/util/custom_flask_restx_fields.py)​

Marshalling with marshal()​

Model nesting pattern​

Further Reading​

DateTimeUtil (`app/util/date_time_util.py`)

Constants (`app/constants/constants.py`)

Enums (`app/enum/`)

StandardCategoryType — `standard_category_type_enum.py`

CategoryType — `category_type.py`

PatternType / PatternTypeV2 — `pattern_type.py`

ReportType — `report_type.py`

ServiceType — `service_type.py`

ReportFormat — `report_format.py`

Permission — `permission.py`

NodeTypes — `node_types.py`

ResponseStatus — `response_status.py`

Utility Functions (`app/util/`)

`number_util.py`

`si_unit_util.py`

`permission_util.py`

`filter_units_util.py`

`shift_and_unit_count_util.py`

`calculation_util.py`

Data Classes (`app/data_class/`)

`ReportRequestData` — `data_class/report.py`

`DeviceDataV2Args` — `data_class/device_data_v2_args.py`

Request Marshalling & Parsing (`app/models/`)

Request Parsers (`reqparse`)

POST Request Body (JSON)

Response Models (`Model` + `fields`)

Registering and using a model in a route

Flask-RESTX field types

Custom field types (`app/util/custom_flask_restx_fields.py`)

Marshalling with `marshal()`

Model nesting pattern

Further Reading