New Page

Verarbeitung von Ladevorgängen via API-Synchronisation

Dokumentation zur Verarbeitung von Gridware Usage Daten über die Gridware API und deren Aggregation zu Customer Usage Data und Supplier Usage Data für die Abrechnung.

---

1. Workflow-Überblick

Gridware API
    ↓
Gridware Usage (Status: Complete)
    ↓
Auto-Fetch + Aggregation
    ↓
Customer Usage Data / Supplier Usage Data
    ↓
Sales Invoice / Purchase Invoice

---

2. Gridware Usage - Das Rohformat

Quelle: becharged/becharged/doctype/gridware_usage/gridware_usage.py

Status-Lebenszyklus

1. Pending
   └─ Rohformat mit gridware_usage_id
   └─ Details werden async abgerufen (Background Job)
   
2. Complete
   └─ Alle Details verfügbar (Kosten, Verbrauch, Zeitstempel)
   └─ Bereit für Aggregation
   
3. Failed
   └─ API-Fehler beim Detail-Abruf
   └─ Wird täglich automatisch erneut versucht

Wichtige Felder

Feld	Bedeutung
gridware_usage_id	Eindeutige ID aus Gridware API
gridware_contract	Verknüpfter Vertrag
start, end	Zeitraum des Ladevorgangs
consumption_value_in_kwh	Stromverbrauch in kWh
usage_price_payment_net_amount	Nettobetrag (ohne MwSt.)
usage_price_payment_amount	Bruttobetrag (mit MwSt.)
rental_object_display_name	Ladesäulen-Name
user_usage_medium_label	Nutzer-ID (z.B. NFC-Card)

Kostenaufschlüsselung (aus API)

total_fix_costs_excl_vat        # Verbindungsgebühr netto
total_fix_costs_incl_vat        # Verbindungsgebühr brutto
total_energy_costs_excl_vat     # €/kWh * Verbrauch (netto)
total_energy_costs_incl_vat     # €/kWh * Verbrauch (brutto)
total_time_costs_excl_vat       # €/min * Dauer (netto)
total_time_costs_incl_vat       # €/min * Dauer (brutto)
total_parking_costs_excl_vat    # Standgebühr (netto)
total_parking_costs_incl_vat    # Standgebühr (brutto)

---

3. Auto-Fetch Mechanismus

Quelle: becharged/becharged/doctype/customer_usage_data/customer_usage_data.py

Was ist Auto-Fetch?

Beim Öffnen von Customer Usage Data oder Supplier Usage Data werden automatisch alle zugehörigen Complete-Status Usages abgerufen.

Ablauf

class CustomerUsageData(BaseUsageData):
    def validate(self):
        # 1. Auto-fetch
        self.auto_fetch_completed_usages()
        
        # 2. Aggregation (hinzufügen zu usage_details Tabelle)
        self.aggregate_completed_usages()
        
        # 3. Berechnung der Summen
        self.calculate_totals()
        
        # 4. Status setzen
        self.set_status()

Auto-Fetch Bedingungen

def auto_fetch_completed_usages(self):
    # Nur wenn gesetzt:
    if not self.customer or not self.billing_period:
        return
    
    # Nicht wenn _completed_usages bereits extern gesetzt (Bulk-API)
    if hasattr(self, '_completed_usages'):
        return
    
    # Query:
    usages = get_customer_usages(
        customer=self.customer,
        billing_period=self.billing_period,
        exclude_usage_ids=[bereits_aggregierte_usages]
    )
    
    self._completed_usages = usages  # Wird dann aggregiert

---

4. Query-Logik für Customer Usages

Quelle: becharged/becharged/utils.py (Funktion: get_customer_usages)

Abfrage-Struktur

def get_customer_usages(
    customer: str = None,                    # Customer Name
    billing_period: str = None,              # "MM/YYYY" z.B. "01/2026"
    from_date: str = None,                   # Optionaler Start
    to_date: str = None,                     # Optionales Ende
    exclude_usage_ids: List[str] = None      # Bereits aggregierte
) -> List[Dict]:

SQL-Logik (vereinfacht)

SELECT 
    gu.name,
    gu.gridware_contract,
    gu.end,
    gc.contract_type,
    gc.vat,
    gu.status
FROM `tabGridware Usage` gu
INNER JOIN `tabGridware Contract` gc 
    ON gu.gridware_contract = gc.name
INNER JOIN `tabGridware Contract Customer` gcc
    ON (gcc.parent = gc.name 
        AND gcc.gridware_party = customer.custom_gridware_party)
WHERE
    gu.status IN ("Complete", "Pending")
    AND NOT gu.name IN (SELECT gridware_usage FROM `tabCustomer Usage Detail`)
    AND MONTH(gu.end) = billing_month
    AND YEAR(gu.end) = billing_year
ORDER BY gu.end

Filtermechanismus

1. Status-Filter: Nur Complete oder Pending
2. Kunde: Über custom_gridware_party auf Customer verknüpft
3. Periode: end Zeitstempel liegt im Abrechnungsmonat
4. Duplikat-Check: Keine bereits in Customer Usage Detail aggr eg rierten Usages
5. Ausschluss-Liste: Weitere anzugebende IDs ausschließen

---

5. Aggregation - Vom Usage zur Usage Data

Quelle: becharged/becharged/utils.py (Klasse: BaseUsageData)

Was ist Aggregation?

Eine Aggregation ist das Hinzufügen eines Gridware Usage zu einem Customer Usage Data Datensatz, indem eine Zeile in der usage_details Tabelle erstellt wird.

Aggregations-Prozess

def aggregate_completed_usages(self):
    """Nur ausgeführt wenn _completed_usages gesetzt ist"""
    
    for usage_data in self._completed_usages:
        usage = frappe.get_doc("Gridware Usage", usage_data["name"])
        contract = frappe.get_doc("Gridware Contract", usage.gridware_contract)
        
        # Item-Mapping: Contract Type + VAT → Item Code
        item_code = get_item_code_from_mapping(
            contract.contract_type,
            contract.vat
        )
        
        if not item_code:
            frappe.log_error(f"No mapping for {contract.contract_type} + {contract.vat}%")
            continue
        
        # Zeile hinzufügen
        self.append("usage_details", {
            "gridware_usage": usage.name,
            "gridware_contract": usage.gridware_contract,
            "contract_type": contract.contract_type,
            "vat": contract.vat,
            "item_code": item_code,
            "usage_start": usage.start,
            "usage_end": usage.end,
            "duration": usage.duration,
            "consumption_kwh": usage.consumption_value_in_kwh,
            "price_excl_vat": usage.usage_price_payment_net_amount,
            "price_incl_vat": usage.usage_price_payment_amount,
            "rental_object_display_name": usage.rental_object_display_name,
            # Kostenaufschlüsselung:
            "usage_price_total_fix_costs_excl_vat": usage.usage_price_total_fix_costs_excl_vat,
            "usage_price_total_fix_costs_incl_vat": usage.usage_price_total_fix_costs_incl_vat,
            "usage_price_total_energy_costs_excl_vat": usage.usage_price_total_energy_costs_excl_vat,
            "usage_price_total_energy_costs_incl_vat": usage.usage_price_total_energy_costs_incl_vat,
            "usage_price_total_time_costs_excl_vat": usage.usage_price_total_time_costs_excl_vat,
            "usage_price_total_time_costs_incl_vat": usage.usage_price_total_time_costs_incl_vat,
            "usage_price_total_parking_costs_excl_vat": usage.usage_price_total_parking_costs_excl_vat,
            "usage_price_total_parking_costs_incl_vat": usage.usage_price_total_parking_costs_incl_vat,
        })

Duplikat-Schutz

def validate_usages_not_aggregated(usage_list: List[str], detail_table_name: str):
    """
    Prüft, dass keine Usage bereits aggregiert wurde.
    detail_table_name: "Customer Usage Detail" oder "Supplier Usage Detail"
    """
    existing = frappe.qb.from_(detail_table)
        .select(detail.gridware_usage)
        .where(detail.gridware_usage.isin(usage_list))
        .run(as_dict=True)
    
    if existing:
        frappe.throw(
            f"These usages are already aggregated: {existing}"
        )

---

6. Berechnung von Gesamtzahlen

Quelle: becharged/becharged/utils.py (Methode: calculate_totals)

Nach der Aggregation werden Summen berechnet:

def calculate_totals(self):
    """Summiert alle Werte aus usage_details"""
    self.total_consumption = 0
    self.total_usage_time = 0
    self.total_charging_sessions = 0
    self.total_price_excl_vat = 0
    self.total_price_incl_vat = 0
    self.total_fix_costs_excl_vat = 0
    self.total_fix_costs_incl_vat = 0
    # ... weitere Kosten-Felder
    
    for detail in self.usage_details:
        if not detail.gridware_usage:
            continue
        
        # Verbrauch & Zeit
        self.total_consumption += detail.consumption_kwh
        self.total_usage_time += detail.duration
        self.total_charging_sessions += 1
        
        # Preise
        self.total_price_excl_vat += detail.price_excl_vat
        self.total_price_incl_vat += detail.price_incl_vat
        
        # Kostenaufschlüsselung
        self.total_fix_costs_excl_vat += detail.usage_price_total_fix_costs_excl_vat
        self.total_energy_costs_excl_vat += detail.usage_price_total_energy_costs_excl_vat
        # ... weitere Kosten
    
    # Eindeutige Ladesäulen zählen
    self.total_charging_points = len(
        set(d.rental_object_display_name for d in self.usage_details)
    )

---

7. Status-Management

Status-Workflow

Uninvoiced
    ↓
    [Knopf: Create Sales/Purchase Invoice]
    ↓
Draft Invoice
    ↓
    [Submit Invoice]
    ↓
Invoiced

Automatische Status-Synchronisation

def _set_status_from_invoice(self, invoice_doctype: str, link_field: str):
    """
    Status abgeleitet vom verknüpften Invoice
    """
    invoice = frappe.db.get_value(
        invoice_doctype,                    # "Sales Invoice" oder "Purchase Invoice"
        {link_field: self.name},            # "custom_customer_usage_data" etc.
        ["name", "docstatus"],
        as_dict=True
    )
    
    if invoice:
        if invoice.docstatus == 0:
            self.status = "Draft Invoice"
        elif invoice.docstatus == 1:
            self.status = "Invoiced"
        elif invoice.docstatus == 2:
            self.status = "Cancelled Invoice"
    else:
        self.status = "Uninvoiced"

Diese Methode wird aufgerufen beim:

• Speichern der Usage Data
• Update/Submit/Cancel des Invoices

---

8. Customer Usage Data vs. Supplier Usage Data

Customer Usage Data

# Query-Beispiel
from becharged.becharged.utils import get_customer_usages

usages = get_customer_usages(
    customer="Customer A",
    billing_period="01/2026"
)

# Erstelle/Update
cud = frappe.get_doc({
    "doctype": "Customer Usage Data",
    "customer": "Customer A",
    "billing_period": "01/2026",
    "month": 1,
    "year": 2026
})
cud._completed_usages = usages  # Auto-fetch wird übersprungen
cud.save()  # validate() aggregiert

Supplier Usage Data

from becharged.becharged.utils import get_supplier_usages

usages = get_supplier_usages(
    supplier="Supplier A",
    billing_period="01/2026"
)

sud = frappe.get_doc({
    "doctype": "Supplier Usage Data",
    "supplier": "Supplier A",
    "billing_period": "01/2026",
    "month": 1,
    "year": 2026
})
sud._completed_usages = usages
sud.save()

Unterschiede

Aspekt	Customer	Supplier
Abfrage	Über custom_gridware_party auf Customer	Über gridware_supplier_party auf Contract
Invoice	Sales Invoice (Verkauf)	Purchase Invoice (Einkauf)
Duplikat-Check	Customer Usage Detail	Supplier Usage Detail
Filter-Feld	gcc.gridware_party	gc.supplier
Nutzer-ID	Ladendes Elektrofahrzeug	CPO (Charging Point Operator)

---

9. Item-Mapping und Preisgruppen

Quelle: becharged/becharged/utils.py (Funktion: get_item_code_from_mapping)

Mapping-Logik

def get_item_code_from_mapping(contract_type: str, vat_rate: float) -> Optional[str]:
    """
    Mappt Contract Type + VAT zu Item Code
    """
    settings = frappe.get_single("Becharged Settings")
    
    # Normalisiere VAT (z.B. 19 für 19%)
    normalized_vat = flt(vat_rate)
    if normalized_vat > 100:
        normalized_vat = normalized_vat / 100
    
    for mapping in settings.contract_type_mappings:
        if (mapping.contract_type == contract_type and 
            abs(flt(mapping.vat_rate) - normalized_vat) < 0.01):
            return mapping.item_code
    
    return None

Mapping-Beispiel (in Becharged Settings)

Contract Type           | VAT % | Item Code
─────────────────────────────────────────
USAGE_POSTPAID          | 19    | CHARGING-POSTPAID-19
USAGE_POSTPAID          | 0     | CHARGING-POSTPAID-0
AUTHORIZATION           | 19    | CHARGING-AUTH-19
USAGE_HUBJECT_CPO_ROAMING | 19  | (wird übersprungen)

Nutzung in Rechnungserstellung

# Gruppiere nach Contract Type + VAT + Item Code
item_groups = {}
for detail in usage_data.usage_details:
    # Überspringe bestimmte Typen
    if detail.contract_type in ["USAGE_ADHOC", "USAGE_HUBJECT_CPO_ROAMING"]:
        continue
    
    group_key = f"{detail.contract_type}|{detail.vat}|{detail.item_code}"
    
    if group_key not in item_groups:
        item_groups[group_key] = {
            "item_code": detail.item_code,
            "contract_type": detail.contract_type,
            "vat": detail.vat,
            "total_price_excl_vat": 0,
            "count": 0
        }
    
    item_groups[group_key]["total_price_excl_vat"] += detail.price_excl_vat
    item_groups[group_key]["count"] += 1

# Jede Gruppe wird 1 Rechnungszeile
for group in item_groups.values():
    invoice.append("items", {
        "item_code": group["item_code"],
        "qty": 1,
        "rate": group["total_price_excl_vat"],
        "description": f"Charging sessions for {period} ({group['contract_type']}, {group['vat']}% VAT) - {group['count']} sessions"
    })

---

10. Fehlerfälle & Logging

Gültige Fehler-Szenarien

# 1. Keine Customer/Supplier-Party gefunden
if not customer:
    frappe.log_error(
        title=f"No Customer found for Gridware Usage {usage.name}",
        message=f"Person: {usage.gridware_party_person}, Org: {usage.gridware_party_organization}"
    )
    continue

# 2. Contract Type Mapping fehlt
if not item_code:
    frappe.log_error(
        title=f"No item mapping found for {contract_type} + {vat}% VAT",
        message=f"Customer: {customer}, Usage: {usage.name}, Period: {billing_period}"
    )
    continue

# 3. Billing Period kann nicht bestimmt werden
if not billing_period:
    frappe.log_error(
        title=f"Could not determine billing period for usage {usage.name}",
        message=f"End datetime: {usage.end}"
    )
    continue

# 4. Pending Usages vorhanden
if pending_count > 0:
    frappe.throw(_(
        f"Found {pending_count} pending Gridware Usage record(s). "
        f"Wait for all usages to complete before creating Usage Data."
    ))

# 5. Duplikat-Aggregation
if usage_already_aggregated:
    frappe.throw(
        f"The following Gridware Usage records are already aggregated: {usage_ids}"
    )

---

11. Beispiel-Workflow: Von API zu Rechnung

Szenario

Kunde "Max Müller" lädt am 15.01.2026 sein Elektroauto

Schritt 1: Gridware Usage erstellt (Pending)

Gridware Usage GW-2026-001
├─ gridware_usage_id: 12345
├─ gridware_contract: Contract-001
├─ start: 2026-01-15 10:00
├─ end: 2026-01-15 11:30
├─ status: Pending
└─ Details: werden async abgerufen

Schritt 2: Background Job holt Details (Complete)

Gridware Usage GW-2026-001
├─ consumption_value_in_kwh: 45.0
├─ usage_price_payment_net_amount: 105.04
├─ usage_price_payment_amount: 125.00
├─ rental_object_display_name: "Charging Point Tesla 1"
├─ user_usage_medium_label: "NFC-7890"
└─ status: Complete ✓

Schritt 3: Benutzer öffnet Customer Usage Data

Customer Usage Data: Max Müller - 01/2026
├─ customer: Max Müller
├─ billing_period: 01/2026
├─ [auto_fetch_completed_usages]
│  └─ Query: get_customer_usages(customer="Max Müller", billing_period="01/2026")
│  └─ Findet: GW-2026-001 (Complete, nicht aggregiert)
├─ [aggregate_completed_usages]
│  └─ Item Mapping: USAGE_POSTPAID + 19% → CHARGING-POSTPAID-19
│  └─ Neue Zeile in usage_details
├─ [calculate_totals]
│  ├─ total_consumption: 45.0 kWh
│  ├─ total_price_excl_vat: 105.04 €
│  ├─ total_price_incl_vat: 125.00 €
│  └─ total_charging_sessions: 1
└─ status: Uninvoiced

Schritt 4: Benutzer erstellt Sales Invoice

Sales Invoice SI-2026-001
├─ customer: Max Müller
├─ custom_customer_usage_data: "Max Müller - 01/2026"
├─ items:
│  └─ CHARGING-POSTPAID-19 | qty=1 | rate=105.04 €
└─ docstatus: 0 (Draft)
     ↓ [Submit]
     └─ docstatus: 1 (Submitted)

Schritt 5: Status-Synchronisation

Customer Usage Data: Max Müller - 01/2026
├─ [on_update von Sales Invoice]
├─ [_set_status_from_invoice]
└─ status: Invoiced ✓

---

12. Best Practices

✓ DO's

# 1. Immer Status prüfen vor Aggregation
if usage.status != "Complete":
    continue

# 2. Duplikate checken
validate_usages_not_aggregated(usage_list)

# 3. Auto-Fetch nutzen statt manuell zu laden
# Usage Data öffnen → auto_fetch_completed_usages wird automatisch aufgerufen

# 4. Fehler loggieren mit vollem Kontext
frappe.log_error(
    title="Usage Aggregation Error - GW-001",
    message=f"Customer: {customer}, Period: {billing_period}, Error: {str(e)}"
)

# 5. Background Jobs für lange Operationen
frappe.enqueue(
    create_customer_usage_data,
    filters=filters,
    queue="long",
    timeout=1200
)

✗ DON'Ts

# 1. Nicht direkt SQL schreiben
# ✗ Schlecht:
frappe.db.sql("SELECT ... WHERE status='Complete'")

# ✓ Gut:
get_customer_usages(customer="...", billing_period="...")

# 2. Nicht ohne Duplikat-Check aggregieren
# ✗ Schlecht:
for usage in usages:
    doc.append("usage_details", ...)

# ✓ Gut:
validate_usages_not_aggregated(usage_list)
for usage in usages:
    doc.append("usage_details", ...)

# 3. Nicht auf Pending Usages ignorieren
# ✗ Schlecht:
for usage in all_usages:  # Auch Pending!
    aggregate(usage)

# ✓ Gut:
complete_usages = [u for u in all_usages if u.status == "Complete"]
for usage in complete_usages:
    aggregate(usage)

---

13. Testing-Checkliste

• Gridware Usage mit Status "Complete" vorhanden
• get_customer_usages() findet korrekte Usages für Kunde + Periode
• Auto-Fetch wird ausgelöst beim Öffnen von Customer Usage Data
• Aggregation fügt Usage zu usage_details hinzu
• Duplikat-Check blockiert bereits aggregierte Usages
• Item-Mapping findet korrekten Item Code
• Totals berechnet sich korrekt
• Status wechselt zu "Draft Invoice" nach Sales Invoice-Erstellung
• Status wechselt zu "Invoiced" nach Submit
• Kostenaufschlüsselung (Fix, Energy, Time, Parking) addiert sich korrekt
• Pending Usages blockieren Aggregation mit Fehlermeldung

---

Ressourcen

• Customer Usage Data: becharged/becharged/doctype/customer_usage_data/customer_usage_data.py
• Supplier Usage Data: becharged/becharged/doctype/supplier_usage_data/supplier_usage_data.py
• Utils: becharged/becharged/utils.py
• Gridware Usage: becharged/becharged/doctype/gridware_usage/gridware_usage.py