Batch Processing

Batch processing aggregates raw 5-minute granular IoT readings into daily, monthly, and yearly summaries stored in the processed_data Cosmos DB container. It runs per-industry and is triggered in two ways: automatically by the Azure Event Hub consumer (bpTrigger event) or manually via the API endpoints documented here.

The processor is organised around standardCategoryId groups — each category type dispatches to a specialised processor class (ConsumptionProcess, EnergyProcess, StockProcess, etc.) that knows how to derive meaningful aggregates for that sensor class.

---

Concepts

Aggregation Periods

Each batch processing run can independently enable or disable three aggregation windows:

Flag	Period	Description
dailyEnabled	Daily	Aggregates all readings for yesterdayDate into one daily summary document
monthlyEnabled	Monthly	Updates the month-to-date summary for the month containing yesterdayDate
yearlyEnabled	Yearly	Updates the year-to-date summary for the year containing yesterdayDate

Date and Shift Offset

The processor derives its working date from the time parameter (ISO format), not the system clock:

todayDate     = datetime.fromisoformat(args['time'])
shift_time    = datetime.strptime(industryData["meta"]["shift"]["startAt"], "%H:%M:%S")
daysOffset    = 1 if shift_time.hour >= 12 else 0
yesterdayDate = todayDate - timedelta(days=1)
offsetMinutes = tz.utcoffset(now).seconds / 60

note

daysOffset shifts the aggregation boundary by one day when the industry's shift starts at or after noon — ensuring that the "day" boundary matches the operational shift rather than midnight.

Category-to-Processor Mapping

standardCategoryId	Processor class	Notes
FLOW, CONSUMPTION (and related flow types)	ConsumptionProcess	Sums interval readings into daily/monthly/yearly totals
ENERGY	EnergyProcess or EnergyProcessV2	Version selected based on industry config
STOCK	StockProcess	Derives closing stock values
GROUND_WATER_LEVEL	GroundWaterLevelProcess	Level-based aggregation
QUALITY	QualityProcess	Averages or samples quality parameters
VIRTUAL	VirtualProcess	Computed from other unit values; only runs when updateOnlyVirtual=true

Dry Run Mode

tip

When updateData=false, the processor runs the full pipeline — fetches raw data, computes aggregates, builds documents — but skips the final Cosmos DB write step. This is useful for validating that a reprocess will produce correct results before committing changes.

---

GET /api/user/batchProcess/ — Run Batch Processing

Triggers a full batch processing run for one industry and one point in time. This endpoint does not require a JWT — it is not decorated with @jwt_required.

GET /api/user/batchProcess/?industryId=IND001&time=2023-08-29T01:30:00.000000&dailyEnabled=true&monthlyEnabled=true&yearlyEnabled=true

Parameters

Parameter	Required	Default	Description
time	Yes	—	ISO format timestamp (e.g. 2023-08-29T01:30:00.000000) — used to derive todayDate and yesterdayDate
industryId	Yes	—	The industry to process
dailyEnabled	No	true	Run daily aggregation
monthlyEnabled	No	true	Run monthly aggregation
yearlyEnabled	No	true	Run yearly aggregation
dailyOldDataEnabled	No	false	Re-aggregate historical daily data
updateOnlyVirtual	No	false	When true, only virtual units are processed — physical unit aggregates are not touched
updateData	No	true	When false, pipeline runs but no writes occur (dry run)
unitIds	No	all units	Comma-separated unit IDs — restricts processing to the listed units only

Validation

Check	Error code	Message
industryId absent	1401	"`industryId` missing"
time absent	1402	"`time` missing"

Execution flow — BatchProcessing.__init__()

info

1. If industryData is not already in args, it is loaded from DatabaseSupporter.get_industry_details_by_id(industryId).
2. If unitIds is provided, industryData['units'] is filtered to only the listed IDs before any further processing.
3. Working dates are computed from time and the industry's meta.shift.startAt (see Concepts above).

Execution flow — BatchProcessing.process()

info

The processor iterates industryData['units'] and groups them by standardCategoryId. For each group:

1. BatchProcessRequestData is built with the group's units and the full units_mapping.
2. The appropriate processor class is instantiated and called with the group data.
3. The processor reads raw readings from Cosmos DB for the relevant window, computes aggregates, and upserts summary documents into processed_data.
4. If updateOnlyVirtual=true, only groups whose standardCategoryId is VIRTUAL are dispatched — all other groups are skipped.
5. Each of dailyEnabled, monthlyEnabled, yearlyEnabled, dailyOldDataEnabled is checked before the corresponding aggregation step is executed within each processor. Disabling a flag skips that aggregation period without affecting the others.

Response

HTTP 200
{
  "status": "success"
}

---

GET /api/user/batchProcess/range — Range Reprocessing

Reprocesses an entire date range day by day. Requires a valid JWT.

GET /api/user/batchProcess/range?startDate=01/11/2024&endDate=09/11/2024&updateData=true
Authorization: Bearer <access_token>
targetIndustryId: IND001        # optional — falls back to g.industry_id from JWT

Parameters

Parameter	Required	Default	Description
startDate	Yes	—	DD/MM/YYYY — first day to reprocess
endDate	Yes	—	DD/MM/YYYY — last day to reprocess (inclusive)
targetIndustryId	No (header)	g.industry_id	Override the industry — useful for admins reprocessing another industry's data
updateOnlyVirtual	No	false	Restrict to virtual unit processing only
updateData	No	true	Set to false for a dry run
unitIds	No	all units	Comma-separated unit IDs

Industry ID resolution

industry_id = args.get('targetIndustryId') or g.industry_id

warning

If neither source yields an industry_id — for example, the header is absent and the JWT does not carry an industry ID — the endpoint returns 400 "Industry ID not found. Please provide targetIndustryId header or ensure valid authentication.".

Execution flow

info

BpTriggerDateRangeService().process_date_range(args, industry_id, current_user) iterates every calendar day in the inclusive [startDate, endDate] window. For each day it calls BatchProcessing with the same flags (updateOnlyVirtual, updateData, unitIds) as the single-run endpoint. The daily calls are sequential — the range service does not parallelise across days.

Response

HTTP 200
{
  "status": "success"
}

Edge cases

warning

Scenario	Behaviour
targetIndustryId header absent and JWT has no industry_id	400 with message prompting the header
startDate is after endDate	Processed with no iterations — the date range loop exits immediately; returns 200
updateData=false	Full pipeline runs for every day in the range, no writes committed
unitIds restricts to units not belonging to the industry	Industry's unit filter produces an empty list; processor runs but produces no output documents

---

Processor Template Pattern

All processors extend a shared base class ProcessTemplate (app/services/batch_processing/types/process_template.py). This enforces a consistent aggregation contract across every category type.

ProcessTemplate — Base Class

@dataclass
class BatchProcessRequestData:
    unitsList: list
    unitsMapping: dict
    industryId: str
    yesterdayDate: datetime
    offsetMinutes: int
    industryData: dict
    todayDate: datetime
    oldData: bool = False
    dailyEnabled: bool = True
    monthlyEnabled: bool = True
    yearlyEnabled: bool = True
    updateData: bool = True
    updateOnlyVirtual: bool = False
    daysOffset: int = 0

Every processor receives a BatchProcessRequestData instance. The base class exposes these shared methods:

Method	What it does
process()	Calls process_daily(), process_monthly(), process_yearly() in order, guarded by the enabled flags. Returns self.resultsData.
process_daily()	Stub — overridden by each processor. Reads raw hourly samples, computes aggregates, upserts daily doc.
process_monthly()	Stub — reads daily docs for all days in the month, aggregates, upserts monthly doc.
process_yearly()	Stub — reads monthly docs for all months in the year, aggregates, upserts yearly doc.
get_daily_common_data(unit_id)	Returns the standard daily document skeleton with id, unitId, industryId, date, hourly: [], tag.
get_monthly_common_data(unit_id)	Returns the standard monthly skeleton with id, unitId, industryId, month, daily: [], tag.
get_yearly_common_data(unit_id)	Returns the standard yearly skeleton with id, unitId, industryId, year, monthly: [], tag.
get_days_list()	Generates UTC timestamps for every day in the processing month. Each day is shifted to the industry's shift.startAt time, then converted to UTC and adjusted by daysOffset.
get_months_list()	Returns ["01/YYYY", "02/YYYY", ..., "12/YYYY"] for yearly aggregation.

note

Document ID format — guaranteed unique per unit per period:

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

No-data handling — every processor maintains a non_data_units list initialised to all units. As data arrives for a unit, it is removed from the list. After the main loop, any unit still in non_data_units gets a fallback document written with value: None and full-day offline stats (offlineCount: 288, offlineDuration: 1440 — 288 = 24h × 12 five-minute slots).

---

Processor Classes

ConsumptionProcess — FLOW / CONSUMPTION

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

Calculates cumulative water consumption using an initial reading (IR) / final reading (FR) approach on summed meter values.

Daily aggregation — step by step:

info

1. Fetch all raw 5-minute samples for yesterdayDate grouped by unitId and hour
2. Within each hour, sort samples by createdOn
3. Compute IR and FR from summedValue (the cumulative meter register):

   ir = value[0]['summedValue'] * flow_factor - value[0]['value'] * flow_factor
   fr = value[-1]['summedValue'] * flow_factor
   total = fr - ir

4. Negative total fallback: If total < 0 (meter rollover or reset), fall back to summing individual value fields:

   total = sum(item['value'] for item in value) * flow_factor
   ir = fr - total

5. Track nonZero — count of samples where value > minNonZeroThreshold (configurable per unit, defaults to 0)
6. Run OfflinePeriodCalculator per hour to get offlineCount, offlineDuration, offlineIntervals, offlineValue
7. Append hourly entry: {x, y: total, fr, ir, offlineCount, offlineDuration, offlineIntervals, offlineValue}
8. Accumulate data['value'] += total
9. Upsert to processed_data_container if updateData=true

Monthly aggregation:

• Reads all daily documents for the month via get_unit_data_by_month()
• Sums value across days; carries forward ir from the first day and fr from the last day with a non-null value
• Accumulates nonZero, offlineCount, offlineDuration, offlineValue from each day's meta

Yearly aggregation: Same pattern over monthly documents.

Written document shape (daily):

{
  "id": "IND001_09-11-2024_UNIT01_DAILY",
  "unitId": "UNIT01",
  "industryId": "IND001",
  "date": "09/11/2024",
  "value": 1240.5,
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 52.3, "fr": 10052.3, "ir": 10000.0, "offlineCount": 0, "offlineDuration": 0, "offlineIntervals": [], "offlineValue": null }],
  "meta": { "flowFactor": 1, "ir": 10000.0, "fr": 11240.5, "nonZero": 284, "offlineCount": 2, "offlineDuration": 10.5, "offlineValue": 3.2 },
  "tag": "09/11/2024_DAILY"
}

Edge cases:

warning

Scenario	Behaviour
total < 0 (meter rollover)	Falls back to summing raw value fields; recalculates ir = fr - total
Unit returns no data at all	Written with value: None, offlineDuration: 1440, offlineCount: 0
minNonZeroThreshold not set in unit config	Defaults to 0 — every non-zero reading counts
updateData=false	Document built in memory, added to resultsData, but upsert_item is never called

---

EnergyProcess — ENERGY (simple units)

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

Identical IR/FR aggregation to ConsumptionProcess, extended with energy savings analysis against a reference power baseline (pRef).

Daily aggregation — step by step:

info

1. Fetch raw 5-minute samples via get_unit_data_by_hour()
2. For each hour compute IR/FR and total exactly as ConsumptionProcess (including negative-total fallback)
3. Accumulate data['value'] += total per hour
4. Run OfflinePeriodCalculator per hour — same as ConsumptionProcess
5. After all hours, call calculate_energy_savings() over all samples for the day
6. Write meta.energySaved and meta.energySavedCoverage into the daily document

calculate_energy_savings() — interval-level algorithm:

for each consecutive sample pair (prev, curr):
    dt_hours = (curr.createdOn - prev.createdOn).total_seconds() / 3600
    if dt_hours <= 0: skip

    if curr.interpolated or curr.offlinePeriod: skip  # gap interval

    energy_delta_kwh = (curr.summedValue - prev.summedValue) * flow_factor / 1000
    p_act = energy_delta_kwh / dt_hours   # interval-average actual power (kW)

    if p_act < 0 or p_act > 1.5 * p_ref: skip  # bad meter / spike

    valid_intervals += 1

    if curr.value <= min_non_zero_threshold: continue  # motor off → zero saving

    e_act = p_act * dt_hours
    e_ref = p_ref * dt_hours
    s_total += (e_ref - e_act)   # signed: negative = overconsumption

expected_intervals = total_hours / (dataUpdateFrequency / 3600)
coverage = min(valid_intervals / expected_intervals, 1.0)

note

• pRef is read from unitsMapping[unitId].meta.pRef (kW). If absent or < 2 samples, returns {energySaved: 0, coverage: 0}
• dataUpdateFrequency defaults to 5 * 60 seconds (5 minutes) if not configured
• coverage is capped at 1.0 and rounded to 4 decimal places

Monthly/Yearly: Sums meta.energySaved across daily/monthly documents. Same IR/FR carry-forward as ConsumptionProcess.

Written document shape (daily):

{
  "value": 1240.5,
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 52.3, "fr": 10052.3, "ir": 10000.0, "offlineCount": 0, "offlineDuration": 0, "offlineIntervals": [], "offlineValue": null }],
  "meta": { "flowFactor": 1, "ir": 10000.0, "fr": 11240.5, "energySaved": 12.4, "energySavedCoverage": 0.9832, "nonZero": 284, "offlineCount": 0, "offlineDuration": 0, "offlineValue": 0 }
}

Edge cases:

warning

Scenario	Behaviour
pRef not set on unit	calculate_energy_savings() returns {energySaved: 0, coverage: 0} immediately
Less than 2 samples	Same early return — no pairs to iterate
p_act < 0 (meter rollover between samples)	Interval dropped; not counted in valid_intervals
p_act > 1.5 × pRef (spike)	Interval dropped as bad reading
value <= minNonZeroThreshold (motor off)	Valid interval counted toward coverage but contributes zero savings
Gap interval (interpolated or offlinePeriod > 0)	Skipped entirely — not counted in valid_intervals or coverage
No data for unit	Written with value: None, energySaved: None, energySavedCoverage: None, offlineDuration: 1440

---

EnergyProcessV2 — ENERGY (units with parameters)

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

Selected when the unit config has a params dict defined. Extends EnergyProcess with per-parameter tracking (e.g. voltage, current, power factor, temperature) stored alongside the energy value.

What's different from EnergyProcess:

info

1. Document gets a top-level paramValue: {} field in addition to value
2. Per hourly entry, a py (parameter values) field is written:

   for param in unitsMapping[unitId]['params']:
       param_values = pd.DataFrame(value)['paramValue'].apply(lambda x: x.get(param))
       hourly_py[param] = param_values.mean() or None

3. Daily paramValue is computed separately per parameter:
   • If params[param].isLatest == true → takes last_sample['paramValue'] (entire dict from last sample)
   • Otherwise → pandas mean across all samples' paramValue[param] for the day
4. Same calculate_energy_savings() method as EnergyProcess — the v2 version divides by 1000 (/ 1000) in the kWh delta calculation, matching Wh-scale summedValue entries

note

Monthly/Yearly paramValue: Recomputed via pd.DataFrame(units)['paramValue'].apply(lambda x: x.get(param)).mean() across the period's documents — not by averaging daily means.

Written document shape (daily):

{
  "value": 1240.5,
  "paramValue": { "voltage": 230.1, "current": 5.4, "powerFactor": 0.92 },
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 52.3, "fr": 10052.3, "ir": 10000.0, "py": { "voltage": 229.8, "current": 5.3, "powerFactor": 0.91 }, "offlineCount": 0, "offlineDuration": 0 }],
  "meta": { "flowFactor": 1, "ir": 10000.0, "fr": 11240.5, "energySaved": 12.4, "energySavedCoverage": 0.9832, "nonZero": 284, "offlineCount": 0, "offlineDuration": 0, "offlineValue": 0 }
}

Edge cases:

warning

Scenario	Behaviour
paramValue column absent from DataFrame	param set to None for that entry
isLatest: true on a param	Entire paramValue dict from the last sample is used — not per-param merge
Param present in config but missing from some samples	x.get(param) returns None; pandas mean excludes it
All param values NaN	Stored as None
No data for unit	Written with value: None, paramValue: {}, offlineDuration: 1440

---

StockProcess — STOCK

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

Tracks tank/reservoir inventory as an absolute level derived from a percentage reading.

Daily aggregation — step by step:

info

1. Fetch hourly percentage readings (0–100%) via get_unit_data_by_hour()
2. Fetch the previous day's closing value via get_yesterday_last_value() — used as openingStock
3. For each hour:

   level_opening = trunc(maxCapacity * (value[0]['value'] / 100), ndigits=2)  # first reading
   level = trunc(maxCapacity * (value[-1]['value'] / 100), ndigits=2)          # last reading (closing)
   percentage = trunc((level / maxCapacity) * 100, 2)

4. data['value'] is set to the last level seen — stock is a snapshot, not a cumulative
5. Hourly entry: {x, y: level (closing), y1: level_opening, percentage, offlineCount, offlineDuration, offlineIntervals}
6. Run OfflinePeriodCalculator (no flow_factor — offline value is not meaningful for stock)

Monthly/Yearly aggregation:

• Reads daily docs; openingStock is taken from the first day's meta.openingStock, closingStock from the last day with a non-null value
• data['value'] is set to the most recent daily level (last snapshot in the period), not a sum

Written document shape (daily):

{
  "value": 85000.0,
  "meta": { "maxCapacity": 100000, "percentage": 85.0, "openingStock": 82000.0, "offlineCount": 0, "offlineDuration": 0 },
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 85000.0, "y1": 82000.0, "percentage": 85.0 }]
}

Edge cases:

warning

Scenario	Behaviour
No previous day closing value	openingStock defaults to 0 via .get('value', 0)
Unit returns no data	Written with value: None, offlineCount: 288, offlineDuration: 1440
level is 0	percentage written as None (avoids divide-by-zero display)
updateOnlyVirtual=true	Yearly non-data units skip the upsert_item call

---

GroundWaterLevelProcess — GROUND_WATER_LEVEL

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

Tracks groundwater depth in metres. Unlike flow, this is a point measurement — only the last reading per hour is stored, and averages are computed across the period.

Daily aggregation — step by step:

info

1. Fetch hourly depth readings via get_unit_data_by_hour()
2. For each hour, take only the last sample: value[-1]['value']
3. data['value'] is updated to the last hour's last reading (final snapshot of the day)
4. Store lastEpoch = value[-1]['createdOn'] in both data['meta'] and each hourly entry's meta
5. Run OfflinePeriodCalculator (no flow_factor)
6. After all hours: compute meta.average = statistics.mean([h['y'] for h in hourly if y is not None])

Monthly/Yearly: data['value'] = last daily reading; meta.average = mean of non-null daily/monthly values across the period.

Written document shape (daily):

{
  "value": 12.4,
  "meta": { "average": 12.1, "offlineCount": 0, "offlineDuration": 0, "lastEpoch": "2024-11-09T22:55:00+00:00" },
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 12.4, "meta": { "lastEpoch": "2024-11-09T06:55:00+00:00" } }]
}

Edge cases:

warning

Scenario	Behaviour
All hourly y values are None	meta.average set to 0 (empty list guard)
Unit returns no data	Written with value: None, offlineCount: 288, offlineDuration: 1440
Multiple samples in one hour	Only the chronologically last sample is used — not an average

---

QualityProcess — QUALITY

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

Tracks multi-parameter water quality measurements. value is a dict ({param: mean}) rather than a scalar.

Daily aggregation — step by step:

info

1. Fetch hourly quality samples — each raw sample's value is a dict: {"pH": 7.1, "TDS": 408, "turbidity": 2.1}
2. For each hour, for each parameter defined in unitsMapping[unitId]['params']:

   param_values = pd.DataFrame(value)['value'].apply(lambda x: x.get(param))
   quality_data[param] = param_values.mean() if not np.nan else None

3. Append hourly entry: {x: hour, y: {param: mean, ...}}
4. After all hours, compute daily value dict using the same mean across all samples (not mean-of-means)
5. Run OfflinePeriodCalculator (no flow_factor)

note

Monthly/Yearly: Per-parameter means are recomputed from the daily value dicts using pandas, not by averaging the daily means.

Written document shape (daily):

{
  "value": { "pH": 7.2, "TDS": 410.5, "turbidity": 1.8 },
  "meta": { "offlineCount": 0, "offlineDuration": 0 },
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": { "pH": 7.1, "TDS": 408, "turbidity": 2.0 } }]
}

Edge cases:

warning

Scenario	Behaviour
A parameter missing from some samples	x.get(param) returns None; pandas treats as NaN; mean excludes it
All samples for a param are None	mean() returns NaN; stored as None in document
Unit returns no data	Written with value: {param: None} for all params, offlineCount: 288
New param added to unit config later	Old documents won't have it; only current params list is iterated

---

VirtualProcess — VIRTUAL

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

Computes derived values by evaluating formula expressions over other units' processed data. Always dispatched last so all physical unit daily results are available.

Constructor difference:

def __init__(self, request_data, all_units_data):
    super().__init__(request_data)
    self.allUnitsData = all_units_data  # daily results from all other processors this run

Daily aggregation — step by step:

info

1. Build units_daily_data_map = {unitId: daily_doc} from all_units_data
2. For each virtual unit, build a units_data dict of string values for expression substitution:

   units_data[u_id] = str(unit_data['value'] or 0)
   # For STOCK units also expose opening stock:
   units_data[f'{u_id}#meta#openingStock'] = str(unit_data['meta']['openingStock'] or 0)

3. Call evaluate_calculation(unit_id, units_data):
   • Reads meta.calculations expression from unitsMapping
   • Expands ADD_ALL → {unitA} + {unitB} + ... (all units in units_data)
   • Expands {subCategoryId} → sum of all units in that sub-category
   • Substitutes via str.format_map(DefaultDict(units_data)) — missing keys resolve to 0
   • Evaluates with eval()
   • If result < 0 and industryData.meta.virtualNegativeCalc is falsy → clamps to 0
   • On any exception → returns None
4. Build hourly virtual data by merging real units' hourly arrays:
   • merge_arrays(data): groups all referenced units' hourly entries by x (hour), produces {x, y: {unitId: value}}
   • For STOCK units: also injects {unitId}#meta#openingStock from y1
   • Evaluates the expression for each merged hour slot
5. Sort hourly entries by x before writing

note

Monthly/Yearly: Reads its own previously written daily documents and sums value — no re-evaluation of formulas.

Written document shape (daily):

{
  "value": 320.5,
  "hourly": [{ "x": "2024-11-09T06:00:00", "y": 14.2 }]
}

Edge cases:

warning

Scenario	Behaviour
Referenced unit has no data this run	DefaultDict substitutes 0 — calculation proceeds without error
Formula references a STOCK unit	openingStock is separately injected as {unitId}#meta#openingStock for use in expressions
eval() throws (syntax/division by zero)	Returns None; virtual unit written with value: None
Negative result + virtualNegativeCalc=false	Clamped to 0
updateOnlyVirtual=true	Batch processor skips all physical processors; VirtualProcess still runs with whatever data is in the DB from previous runs
ADD_ALL with no units	Expands to empty string — eval() will throw, returns None

---

OfflinePeriodCalculator

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

Shared utility used by all processors except QualityProcess. Detects and quantifies data transmission gaps within a processing hour.

How it works:

Each raw sample carries an offlinePeriod field — the number of minutes the device was offline before that sample was sent. The calculator builds explicit time intervals from this:

interval_start = createdOn - timedelta(minutes=offlinePeriod)
interval_end   = createdOn

info

1. Build all offline intervals from samples where offlinePeriod > 0
2. Merge overlapping intervals (sorted by start time)
3. Compute overlap of merged intervals with the current hour window (hour_start to hour_start + 1h)
4. Sum overlap durations → hourly_offline_duration (minutes)
5. For samples flagged interpolated: true that fall within a merged interval, sum value * flow_factor → offline_y_value (estimated consumption during the gap)
6. Mutate data['meta'] in place: increments offlineCount, offlineDuration, offlineValue

Return values:

Value	Description
formatted_intervals	List of {start, end, diff} dicts for each merged gap
hourly_offline_duration	Minutes of gap overlap within this hour
offline_record_count	Number of samples in this hour with offlinePeriod > 0
offline_y_value	Estimated flow during gaps (from interpolated samples)

Edge cases:

warning

Scenario	Behaviour
No samples have offlinePeriod	Returns empty intervals, zero duration and count
Overlapping offline intervals	Merged before duration calculation — no double-counting
flow_factor is None (Stock, GWL)	offline_y_value returns 0 — caller passes None explicitly
interpolated flag absent on sample	Treated as False — not counted toward offline_y_value

---

Status Code Reference

Code	Meaning
200	Batch processing completed successfully
400	industry_id could not be resolved (range endpoint only)
401	Missing or invalid JWT (range endpoint only)
1401	industryId missing (single-run endpoint)
1402	time missing (single-run endpoint)
500	Internal error during processing

---

For adding a new processor type to the codebase, see Adding a Batch Processor.