Adding a Batch Processor

Quick steps to add a new processor type. For full details — all existing processor types, virtual formula system, unit config keys, offline tracking, ir/fr, pRef, and updateOnlyVirtual — see Batch Processor Reference.

---

Files to create / modify

File	Action
app/enum/standard_category_type_enum.py	Add new enum value; add to all_categories()
app/services/batch_processing/types/my_process.py	Create — extend ProcessTemplate
app/services/batch_processing/batch_processing.py	Import and add elif dispatch branch

---

Step 1 — Add a StandardCategoryType enum value

File: app/enum/standard_category_type_enum.py

class StandardCategoryType(Enum):
    ...
    MY_CATEGORY = 'MY_CATEGORY'   # ← add this

@staticmethod
def all_categories():
    return [
        ...
        StandardCategoryType.MY_CATEGORY.value,
    ]

warning

This value must match the standardCategoryId field on unit documents in Cosmos DB.

---

Step 2 — Create the processor class

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

Extend ProcessTemplate. Implement process_daily(), process_monthly(), process_yearly(), and process().

from itertools import groupby
from app.database.database_config import processed_data_container
from app.database.database_supporter import DatabaseSupporter
from app.services.batch_processing.types.process_template import ProcessTemplate
from app.util.date_time_util import DateTimeUtil


class MyProcess(ProcessTemplate):

    def process_daily(self):
        units_hour_data = DatabaseSupporter.get_unit_data_by_hour(
            self.requestData.industryId,
            self.requestData.unitsList,
            self.requestData.yesterdayDate,
            self.requestData.offsetMinutes,
            old_data=self.requestData.oldData
        )
        non_data_units = [unit for unit in self.requestData.unitsList]

        for unitId, units in groupby(units_hour_data, lambda k: k['unitId']):
            units = list(units)
            non_data_units.remove(unitId)

            data = {
                **self.get_daily_common_data(unitId),
                'value': 0,
                'meta': { 'offlineCount': 0, 'offlineDuration': 0 }
            }

            for key, value in groupby(sorted(units, key=lambda k: k['createdOn']), lambda k: k['hour']):
                value = list(value)
                hourly_value = sum(s['value'] for s in value if s.get('value') is not None)
                data['value'] += hourly_value
                data['hourly'].append({'x': key, 'y': hourly_value})

            if self.requestData.updateData:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

        # No-data placeholder — always write so monthly/yearly aggregation finds a doc
        for unitId in non_data_units:
            data = {
                **self.get_daily_common_data(unitId),
                'value': None,
                'meta': { 'offlineCount': 288, 'offlineDuration': 1440 }
            }
            if self.requestData.updateData:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

    def process_monthly(self):
        units_daily_data = DatabaseSupporter.get_unit_data_by_month(
            self.requestData.industryId,
            self.requestData.unitsList,
            self.get_days_list()
        )
        non_data_units = [unit for unit in self.requestData.unitsList]

        for unitId, units in groupby(units_daily_data, lambda k: k['unitId']):
            units = list(units)
            non_data_units.remove(unitId)
            data = { **self.get_monthly_common_data(unitId), 'value': 0 }

            for value in sorted(units, key=lambda k: DateTimeUtil().strptime(k['date'], '%d/%m/%Y').timestamp()):
                total = value['value']
                data['value'] += total if total else 0
                data['daily'].append({'x': value['date'], 'y': total})

            if self.requestData.updateData:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

        for unitId in non_data_units:
            data = { **self.get_monthly_common_data(unitId), 'value': None }
            if self.requestData.updateData:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

    def process_yearly(self):
        units_monthly_data = DatabaseSupporter.get_unit_data_by_year(
            self.requestData.industryId,
            self.requestData.unitsList,
            self.get_months_list()
        )
        non_data_units = [unit for unit in self.requestData.unitsList]

        for unitId, units in groupby(units_monthly_data, lambda k: k['unitId']):
            units = list(units)
            non_data_units.remove(unitId)
            data = { **self.get_yearly_common_data(unitId), 'value': 0 }

            for value in sorted(units, key=lambda k: DateTimeUtil().strptime(k['month'], '%m/%Y').timestamp()):
                total = value['value']
                data['value'] += total if total else 0
                data['monthly'].append({'x': value['month'], 'y': total})

            if self.requestData.updateData:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

        for unitId in non_data_units:
            data = { **self.get_yearly_common_data(unitId), 'value': None }
            if self.requestData.updateData and not self.requestData.updateOnlyVirtual:
                processed_data_container.upsert_item(data)
            self.resultsData.append(data)

    def process(self):
        if self.requestData.dailyEnabled:
            self.process_daily()
        if self.requestData.monthlyEnabled:
            self.process_monthly()
        if self.requestData.yearlyEnabled:
            self.process_yearly()
        return self.resultsData

Always write a no-data placeholder

When a unit has no samples, write a placeholder document (value: null) so monthly/yearly aggregation always finds something to read. Never skip the unit entirely.

Base class helpers

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 days in the current month as DD/MM/YYYY (shift-adjusted)
get_months_list()	All 12 months of the current year as MM/YYYY

---

Step 3 — Register in BatchProcessing

File: app/services/batch_processing/batch_processing.py

from app.services.batch_processing.types.my_process import MyProcess

# Inside process(), after existing elif branches:
elif key == StandardCategoryType.MY_CATEGORY.value:
    result_data += MyProcess(request_data).process()

The groupby loop already builds request_data for your category — you only need to add the branch.

---

Edge cases to handle

Scenario	What to do
Unit has no raw samples	Write no-data placeholder — never skip the unit
value = None in a sample	Guard with if s.get('value') is not None before arithmetic
Monthly/yearly reads a None daily value	Skip in aggregation: data['value'] += total if total else 0
Month start — only day 1 exists	get_days_list() returns all days in the month; partial result is expected
updateOnlyVirtual = true	Guard yearly no-data writes: if not self.requestData.updateOnlyVirtual
updateData = false (dry run)	Skip all processed_data_container.upsert_item() calls — already done via the if self.requestData.updateData: guard