Task 06 — Add a New Batch Processor Type

Difficulty: Intermediate–Advanced
Goal: Add a new batch processor for a new category of units — from the StandardCategoryType enum through the ProcessTemplate subclass and dispatcher registration.

---

What you'll learn

• How the batch processing dispatcher groups units by standardCategoryId
• How ProcessTemplate base class helpers work (get_daily_common_data, get_monthly_common_data, etc.)
• Daily → monthly → yearly aggregation chain
• The no-data pattern and offline tracking

---

Your task: "Solar Energy Processor"

Add a new category SOLAR_CATEGORY that aggregates daily, monthly, and yearly energy generation data from solar panel units.

---

Files to create / edit

File	Action
app/enum/standard_category_type_enum.py	Add SOLAR_CATEGORY = 'SOLAR_CATEGORY'; add to all_categories()
app/services/batch_processing/types/solar_process.py	Create — extend ProcessTemplate
app/services/batch_processing/batch_processing.py	Import SolarProcess; add elif branch in process()

Existing categories for reference

The real StandardCategoryType values already in the codebase are: SOURCE_CATEGORY, ENERGY_CATEGORY, STOCK_CATEGORY, GROUND_WATER_LEVEL, QUALITY_CATEGORY, VIRTUAL_CATEGORY — check app/enum/standard_category_type_enum.py before adding yours to avoid conflicts.

---

Processor skeleton

# app/services/batch_processing/types/solar_process.py
from app.services.batch_processing.types.process_template import ProcessTemplate, BatchProcessRequestData
from app.database.database_supporter import DatabaseSupporter

class SolarProcess(ProcessTemplate):

    def __init__(self, request_data: BatchProcessRequestData):
        super().__init__(request_data)

    def process_daily(self):
        for unit_id in self.requestData.unitsList:
            raw_data = DatabaseSupporter.get_device_data_by_unit_and_date(
                self.requestData.industryId,
                unit_id,
                self.requestData.yesterdayDate,
            )

            daily_doc = self.get_daily_common_data(unit_id)

            if not raw_data:
                daily_doc['value'] = None
                daily_doc['meta'] = {'offlineCount': 288, 'offlineDuration': 1440}
                self.resultsData.append(daily_doc)
                continue

            # Solar: sum all positive generation readings for the day
            total_generation = sum(
                s.get('value', 0) for s in raw_data if s.get('value') is not None
            )

            hourly = []
            for hour in range(24):
                hour_samples = [s for s in raw_data if s.get('hour') == hour]
                y = sum(s.get('value', 0) for s in hour_samples) if hour_samples else None
                hourly.append({'hour': hour, 'y': y})

            daily_doc['value']  = total_generation
            daily_doc['hourly'] = hourly
            daily_doc['meta']   = {}
            self.resultsData.append(daily_doc)

    def process_monthly(self):
        for unit_id in self.requestData.unitsList:
            monthly_doc  = self.get_monthly_common_data(unit_id)
            daily_values = []

            for day in self.get_days_list():
                doc_id = f'{self.requestData.industryId}_{day.replace("/", "-")}_{unit_id}_DAILY'
                doc = DatabaseSupporter.get_processed_data_by_id(doc_id)
                if doc and doc.get('value') is not None:
                    daily_values.append(doc['value'])
                    monthly_doc['daily'].append({'date': day, 'y': doc['value']})

            monthly_doc['value'] = sum(daily_values) if daily_values else None
            self.resultsData.append(monthly_doc)

    def process_yearly(self):
        for unit_id in self.requestData.unitsList:
            yearly_doc     = self.get_yearly_common_data(unit_id)
            monthly_values = []

            for month in self.get_months_list():
                doc_id = f'{self.requestData.industryId}_{month.replace("/", "-")}_{unit_id}_MONTHLY'
                doc = DatabaseSupporter.get_processed_data_by_id(doc_id)
                if doc and doc.get('value') is not None:
                    monthly_values.append(doc['value'])
                    yearly_doc['monthly'].append({'month': month, 'y': doc['value']})

            yearly_doc['value'] = sum(monthly_values) if monthly_values else None
            self.resultsData.append(yearly_doc)

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

        if self.requestData.updateData:
            for doc in self.resultsData:
                DatabaseSupporter.upsert_processed_data(doc)

        return self.resultsData

---

Dispatcher branch

# app/services/batch_processing/batch_processing.py
from app.services.batch_processing.types.solar_process import SolarProcess

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

---

Done when

• SOLAR_CATEGORY is in the enum and in all_categories()
• SolarProcess processes daily data correctly (no-data pattern included)
• Monthly aggregation reads from daily docs written by process_daily()
• Yearly aggregation reads from monthly docs
• The elif branch in batch_processing.py correctly routes to SolarProcess

---

Edge cases to verify

Test these scenarios before shipping

Scenario	What to verify
Unit has zero samples for the day	process_daily() writes a placeholder doc with value: None, offlineCount: 288, offlineDuration: 1440 — monthly aggregation must still find a doc to read
A sample has value = None	Guarded by if s.get('value') is not None — only non-null readings are summed
process_monthly() runs at month start (day 1 only)	get_days_list() returns all calendar days including future ones; most docs will be missing or null — monthly_doc['value'] will be partial, which is correct
updateData=false (dry run)	upsert_processed_data is never called; resultsData is returned but nothing is written
updateOnlyVirtual=true	SolarProcess is skipped entirely by the dispatcher; only VirtualProcess runs
dailyEnabled=false	process_daily() is not called; only monthly/yearly steps run
Virtual formula references a solar unit	No extra work needed — your daily docs carry standard id/unitId fields and are automatically available to VirtualProcess
Running the trigger with unitIds=U001,U002	Only those two units appear in self.requestData.unitsList; other units are not processed