Batch Processor — Reference

Complete reference for all processor types, the virtual formula system, unit config keys that control processor behaviour, how offline tracking works, and the BatchProcessRequestData fields.

---

Dispatcher — how BatchProcessing.process() works

BatchProcessing.process() in app/services/batch_processing/batch_processing.py groups all industry units by standardCategoryId and dispatches to the matching processor class:

for key, value in groupby(self.industryData['units'], lambda k: k['standardCategoryId']):
    ...
    if key == StandardCategoryType.SOURCE_CATEGORY.value:
        result_data += ConsumptionProcess(request_data).process()
    elif key == StandardCategoryType.ENERGY_CATEGORY.value:
        normal_units  = [uid for uid in units if 'params' not in units_mapping[uid]]
        complex_units = [uid for uid in units if 'params' in units_mapping[uid]]
        if normal_units:  result_data += EnergyProcess(request_data).process()
        if complex_units: result_data += EnergyProcessV2(request_data).process()
    elif key == StandardCategoryType.QUALITY_CATEGORY.value:
        result_data += QualityProcess(request_data).process()
    elif key == StandardCategoryType.GROUND_WATER_LEVEL.value:
        result_data += GroundWaterLevelProcess(request_data).process()
    elif key == StandardCategoryType.STOCK_CATEGORY.value:
        result_data += StockProcess(request_data).process()
    elif key == StandardCategoryType.VIRTUAL_CATEGORY.value:
        virtual_process_data.append(request_data)   # deferred — runs last

# After all physical processors:
for virtual_data in virtual_process_data:
    daily_data = [d for d in result_data if d['id'].endswith('_DAILY')]
    result_data += VirtualProcess(virtual_data, daily_data).process()

VirtualProcess always runs last because it needs the daily documents produced by all physical processors.

---

BatchProcessRequestData fields

Field	Type	Description
unitsList	list[str]	Unit IDs belonging to this category group
unitsMapping	dict	unitId → full unit document — includes all config keys
industryId	str	Industry being processed
yesterdayDate	datetime	The date being aggregated (shift-adjusted "yesterday")
todayDate	datetime	yesterdayDate + 1 day — used for document IDs
offsetMinutes	int	Industry timezone offset in minutes from UTC
industryData	dict	Full industry document (meta, units, categories)
oldData	bool	Re-process historical data (not just yesterday)
dailyEnabled	bool	Run daily aggregation (default True)
monthlyEnabled	bool	Run monthly aggregation (default True)
yearlyEnabled	bool	Run yearly aggregation (default True)
updateData	bool	Write results to Cosmos DB (dry run when False)
updateOnlyVirtual	bool	Skip physical processor writes, only run virtual
daysOffset	int	1 if shift starts at or after noon, else 0

---

Base class helpers (ProcessTemplate)

Method	Returns
get_daily_common_data(unit_id)	Dict with id, unitId, industryId, date, hourly: [], tag
get_monthly_common_data(unit_id)	Dict with id, unitId, industryId, month, daily: [], tag
get_yearly_common_data(unit_id)	Dict with id, unitId, industryId, year, monthly: [], tag
get_days_list()	All calendar days in the current month as DD/MM/YYYY strings (shift-adjusted)
get_months_list()	All 12 months of the current year as MM/YYYY strings

Document ID format:

daily:   {industryId}_{DD-MM-YYYY}_{unitId}_DAILY
monthly: {industryId}_{MM-YYYY}_{unitId}_MONTHLY
yearly:  {industryId}_{YYYY}_{unitId}_YEARLY

---

Processor types — full reference

ConsumptionProcess — SOURCE_CATEGORY

File: app/services/batch_processing/types/consumption_process.py

Flow meters. Reads a cumulative register (odometer). Per-hour delta:

ir    = firstSample.summedValue × flowFactor
fr    = lastSample.summedValue  × flowFactor
total = fr - ir

If total < 0 (meter rollover), falls back to summing interval value fields:

total = sum(sample.value for sample in hour) × flowFactor
ir    = fr - total

Daily value = sum of all per-hour totals.
Monthly/yearly value = sum of child daily value fields.

meta fields:

Field	Description
flowFactor	Unit's flow conversion multiplier
ir	Initial register — cumulative position at start of day
fr	Final register — cumulative position at end of day
nonZero	Count of samples with value > minNonZeroThreshold
offlineCount	Samples with offlinePeriod > 0
offlineDuration	Total offline minutes
offlineValue	Estimated volume during offline gaps

No-data document: value: null, offlineCount: 0, offlineDuration: 1440, offlineValue: 288.

---

EnergyProcess — ENERGY_CATEGORY (no params)

File: app/services/batch_processing/types/energy_process.py

Energy meters without additional parameter tracking. Same IR/FR delta as ConsumptionProcess. Adds energy savings calculation after daily aggregation.

Energy savings algorithm (calculate_energy_savings):

For each consecutive sample pair (prev, curr):

1. dt_hours = elapsed hours between timestamps
2. energy_delta_kwh = (curr.summedValue - prev.summedValue) × flowFactor
3. p_act = energy_delta_kwh / dt_hours — interval-average power in kW
4. Skip interval if: sample is interpolated/offline gap, p_act < 0, or p_act > 1.5 × pRef (bad meter)
5. If motor is off (curr.value ≤ minNonZeroThreshold) — valid interval, zero saving
6. Otherwise: saving = (pRef × dt_hours) - (p_act × dt_hours)

pRef is read from unit.meta.pRef (kW). If absent, savings are 0.
Coverage = validIntervals / expectedIntervals (capped at 1.0), where expectedIntervals is derived from unit.dataUpdateFrequency.

Extra meta fields:

Field	Description
energySaved	kWh saved vs reference for the day (signed — negative if unit used more than reference)
energySavedCoverage	Fraction of intervals (0–1) with valid data for savings calculation

No-data meta: offlineCount: 0, offlineDuration: 1440, offlineValue: 288, energySaved: null, energySavedCoverage: null.

---

EnergyProcessV2 — ENERGY_CATEGORY (with params)

File: app/services/batch_processing/types/energy_process_v2.py

Energy units that have a params dict in their unit config — for meters that also capture additional readings (voltage, current, power factor, etc.) alongside the main energy counter.

Triggered when: unit document contains a params key (dispatched separately from plain energy units in BatchProcessing.process()).

params dict structure on the unit config:

{
  "voltage":     { "isLatest": false },
  "current":     { "isLatest": true },
  "powerFactor": { "isLatest": false }
}

isLatest: true — use the last sample's paramValue dict for the day. Used for instantaneous readings where averaging doesn't make sense.

isLatest: false — average all paramValue[param] across the day using pandas mean().

Extra document fields:

• paramValue: { "voltage": <avg>, "current": <latest>, ... } — at daily, monthly, and yearly level
• Hourly entries include py: { param: <hourly_avg> }

Same IR/FR delta and energy savings logic as EnergyProcess.

---

QualityProcess — QUALITY_CATEGORY

File: app/services/batch_processing/types/quality_process.py

Water quality sensors. Reads instantaneous concentration values — not a cumulative counter.

value field = dict, not a scalar. Each key is a param name, value is the daily mean:

{ "pH": 7.2, "turbidity": 0.8, "tds": 340.5 }

Hourly y = dict of per-param means for that hour.

Monthly/yearly value = mean of daily means per param (average of averages — not a sum, because concentrations don't accumulate).

No-data document: value: { "pH": null, ... } one null per param. meta.offlineCount: 288, offlineDuration: 1440.

---

GroundWaterLevelProcess — GROUND_WATER_LEVEL

File: app/services/batch_processing/types/ground_water_level_process.py

Snapshot sensors — the reading is the current level, not a delta.

Daily value = the last sample's value from the last hour processed (most recent reading of the day).

Hourly y = last sample's value for that hour (not a sum).

Monthly/yearly value = the last day's/month's level (most recent at end of period).

meta fields:

Field	Description
average	Mean of all hourly y values (mean level across the day)
offlineCount	Accumulated offline sample count
offlineDuration	Accumulated offline minutes
lastEpoch	createdOn of the last processed sample (daily only)

---

StockProcess — STOCK_CATEGORY

File: app/services/batch_processing/types/stock_process.py

Tank/reservoir level. Raw data is a % reading (0–100). Converted to absolute volume:

level = maxCapacity × (percentRemaining / 100)

Daily value = level from the last sample of the last hour (closing level).

Opening stock = last sample from the previous day, fetched via DatabaseSupporter.get_yesterday_last_value() before processing begins. Stored in meta.openingStock.

Hourly y1 = level at the start of the hour (used by virtual formulas via #meta#openingStock).

meta fields:

Field	Description
maxCapacity	Max tank volume from unit config
percentage	(level / maxCapacity) × 100 truncated to 2dp
openingStock	Level at start of day (last reading of previous day)
offlineCount	Accumulated offline count
offlineDuration	Accumulated offline minutes

Monthly meta adds closingStock = last day's level, openingStock = first day's opening stock.

---

VirtualProcess — VIRTUAL_CATEGORY

File: app/services/batch_processing/types/virtual_process.py

Virtual units have no physical meters. Their value is computed from a formula that references other units' daily values.

Execution order: always last. Receives the full result_data list from all physical processors.

Formula string

Stored at unit.meta.calculations. A Python arithmetic expression with curly-brace unit ID placeholders:

{U001} + {U002} - {U003}
{U001} / ({U002} + {U003}) * 100
ADD_ALL
{SOURCE_INLET} - {SOURCE_OUTLET}

ADD_ALL expands to {uid} + {uid} + ... for every unit in units_data.

Sub-category ID (e.g. {SOURCE_INLET}) expands to {uid} + {uid} + ... for every unit in that sub-category, looked up from industryData['categories']['SOURCE_CATEGORY']['subCategories'].

Formula evaluation

class DefaultDict(dict):
    def __missing__(self, key): return 0

result = eval(calculation.format_map(DefaultDict(units_data)))

All values passed to format_map are strings. Missing unit IDs default to '0' via DefaultDict.__missing__ — the formula never raises KeyError for missing units.

Negative result: clamped to 0 by default. If industryData.meta.virtualNegativeCalc = true, negative results are preserved as-is.

What units_data contains

For daily process_daily():

• All physical unit daily docs from self.allUnitsData
• Key = unitId, value = str(doc['value'] or 0)
• Stock units additionally expose {unitId}#meta#openingStock = str(doc['meta']['openingStock'] or 0) — used for formulas like {U005} - {U005#meta#openingStock} to compute tank consumption

For hourly rows: formula re-evaluated at each hour using source units' hourly[i]['y'] values merged by unitId.

Monthly and yearly: read back daily/monthly docs from Cosmos DB and sum value — no formula re-evaluation.

Selecting source units for hourly merge (meta.units)

• ['ALL'] — use every unit in units_daily_data_map
• [unitId, ...] — use only the listed unit IDs
• meta.subCategories — units from those sub-categories are appended to selected_units

No-data and formula errors

If evaluate_calculation() throws (division by zero, bad formula), it returns None — the document is written with value: null, not a crash.

If a unit has no meta.calculations, it is skipped entirely — no document is written.

---

Offline tracking — how it works

File: app/services/batch_processing/types/offline_period_calculator.py

Each raw sample in devices_data can carry an offlinePeriod field (minutes) — set by the IoT ingestion pipeline when it detects a gap since the previous sample. The OfflinePeriodCalculator reads these to reconstruct offline intervals.

Algorithm:

1. For each sample with offlinePeriod > 0, create an interval: (createdOn - offlinePeriod, createdOn)
2. Merge overlapping intervals
3. Intersect merged intervals with the current hour window to get hourly_offline_duration (minutes)
4. Count samples with offlinePeriod > 0 → offline_record_count
5. For flow/energy units, sum value of interpolated=True samples within offline intervals → offline_y_value (estimated volume during gap)

Offline sentinel values for no-data units:

Field	Value	Meaning
offlineCount	288	All 5-minute intervals in 24 hours
offlineDuration	1440	All 24 × 60 minutes
offlineValue	288	Estimated offline contribution (flow only)

To change the offline threshold for a unit: set unit.dataUpdateFrequency (seconds) — used by the device data formatter and granular data formatter to determine whether a unit is currently online. The batch processor itself uses offlinePeriod from raw samples, not dataUpdateFrequency.

---

Unit config keys reference

These keys are read from the unit document (stored under industries.units[] in Cosmos DB) and affect processor and formatter behaviour.

Core identity

Key	Type	Used by	Description
unitId	str	All	Unique unit identifier
unitName	str	All	Display name shown in reports and UI
unitDesc	str	Device data formatters	Secondary description / connection number shown below the unit name in the device data panel and notifications. Optional — null if absent
standardCategoryId	str	Dispatcher	Determines which processor class handles this unit (SOURCE_CATEGORY, ENERGY_CATEGORY, etc.)
isDeployed	bool	IndustryDataFormatter	Only deployed units are included in unitsMapping. Units with isDeployed: false are completely excluded from all API responses, batch processing, and formatters. Use this to disable a unit without deleting it

Flow / energy processing

Key	Type	Used by	Description
flowFactor	float	ConsumptionProcess, EnergyProcess, EnergyProcessV2	Multiplier applied to raw register values to convert to the industry's SI unit. flowFactor: 1000 means raw value in litres, output in kL
minNonZeroThreshold	float	ConsumptionProcess, EnergyProcess, EnergyProcessV2	Samples with value ≤ minNonZeroThreshold are counted as "zero" (motor off). Affects meta.nonZero count and energy savings (motor-off intervals don't contribute savings). Default 0
pRef	float	EnergyProcess, EnergyProcessV2	Reference power draw in kW for energy savings calculation. If absent, energySaved = 0. Set in unit.meta.pRef
eneThreshold	float	EnergyProcessV2	Energy-specific threshold read during V2 processing. Configures per-unit energy alarm sensitivity

Stock / tank

Key	Type	Used by	Description
maxCapacity	float	StockProcess, IndustryDataFormatter	Maximum tank volume. Raw % readings are converted to absolute volume via level = maxCapacity × (percent / 100). Merged from alerts_config.units[unitId].configs.maxCapacity by IndustryDataFormatter

Quality / multi-param

Key	Type	Used by	Description
params	dict	QualityProcess, EnergyProcessV2, AlertsConfig	Dict of parameter names → config. For quality units: { "pH": {}, "turbidity": {} }. For energy V2 units: { "voltage": { "isLatest": false }, "current": { "isLatest": true } }. Presence of params is what triggers EnergyProcessV2 instead of EnergyProcess

Virtual units

Key	Type	Used by	Description
meta.calculations	str	VirtualProcess	Formula string for virtual unit value. Supports {unitId} placeholders, ADD_ALL, and sub-category ID expansion. If absent, unit is skipped
meta.units	list	VirtualProcess	Which source unit IDs to include when building the hourly merge for this virtual unit. ['ALL'] = all units
meta.subCategories	list	VirtualProcess	Sub-category IDs whose units are appended to meta.units for the hourly merge

Online / offline detection

Key	Type	Used by	Description
dataUpdateFrequency	int (seconds)	Device data formatters, granular formatter, health function, physical node	How often this unit is expected to send data. If (now - lastUpdated) < dataUpdateFrequency, the unit is online. Default threshold when absent is 3600 (1 hour). To change online/offline detection threshold for a unit, set this key

Manual / virtual nodes

Key	Type	Used by	Description
unitType	str	VirtualDeviceDataService	Set to MANUAL_FLOW_CATEGORY for manually-entered flow units. Determines which units appear in the manual node UI
manualMeterType	str	VirtualDeviceDataService	'RS485' (cumulative register, uses summedValue) or 'PULSE' (incremental, adds to last summed value before sending to EventHub). Default 'RS485'

---

Industry-level meta flags

These keys are read from industryData.meta and affect processing globally for the industry.

Key	Type	Used by	Description
timezone	str	All	IANA timezone string (e.g. 'Asia/Kolkata'). Used by IndustryDataFormatter and DateTimeUtil for all date/time conversions
shift.startAt	str (HH:MM:SS)	Batch processor, formatters, reports	Industry's operational shift start time in local time. Determines daysOffset and the BP trigger window
shift.endAt	str (HH:MM:SS)	Reports, virtual device data	Shift end time
siUnit	dict	Reports, formatters	Per-category SI unit labels shown in reports (e.g. { "source": "kL", "energy": "kWh" })
virtualNegativeCalc	bool	VirtualProcess	If true, virtual formula results are allowed to be negative. Default: negative results are clamped to 0
logo	str (URL)	Report templates	Industry logo URL injected into all report headers
autoReportConfigs	list	Automated report service	List of scheduled report configs for this industry

---

updateOnlyVirtual — end-to-end behaviour

When triggered with updateOnlyVirtual=true:

1. All category groups still iterate
2. Physical processors run daily aggregation (needed so virtual formulas have fresh daily docs to read)
3. For no-data units in physical processors, monthly/yearly no-data placeholder writes are skipped (guarded by if not self.requestData.updateOnlyVirtual)
4. VirtualProcess runs normally and writes results

note

Physical daily docs are still written even with updateOnlyVirtual=true — they feed the virtual formula evaluation. Only the no-data monthly/yearly placeholder writes for physical units are suppressed.

---

ir and fr — initial and final register

ir (initial reading) and fr (final reading) are the cumulative odometer positions at period boundaries:

• Daily ir = first hour's first sample summedValue × flowFactor
• Daily fr = last hour's last sample summedValue × flowFactor
• Monthly ir = ir from the first daily doc (carried forward unchanged)
• Monthly fr = fr from the last daily doc

These are stored for reports that display absolute meter positions, not used in value calculation.

---

pRef — energy savings configuration

unit.meta.pRef is the reference power in kW (nameplate or benchmark). Required for energy savings calculation.

• If absent: energySaved = 0, energySavedCoverage = 0 — always written to the document, just zeroed
• Savings are signed — negative if the unit used more energy than the reference

---

Offline sentinel values

Field	No-data value	Meaning
offlineCount	288	24 h × 12 five-minute intervals = all intervals offline
offlineDuration	1440	All 24 × 60 minutes offline
offlineValue	288	Estimated offline volume (flow processors only)