pyopenadr/openleadr/objects.py

# SPDX-License-Identifier: Apache-2.0

# Copyright 2020 Contributors to OpenLEADR

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from dataclasses import dataclass
from typing import List
from datetime import datetime, timezone, timedelta

class objdict(dict):
    def __getattr__(self, name):
        if name in self:
            return self[name]
        else:
            raise AttributeError("No such attribute: " + name)

    def __setattr__(self, name, value):
        self[name] = value

    def __delattr__(self, name):
        if name in self:
            del self[name]
        else:
            raise AttributeError("No such attribute: " + name)

@dataclass
class AggregatedPNode(objdict):
    node: str

@dataclass
class EndDeviceAsset(objdict):
    mrid: str

@dataclass
class MeterAsset(objdict):
    mrid: str

@dataclass
class PNode(objdict):
    node: str

@dataclass
class FeatureCollection(objdict):
    id: str
    location: dict

@dataclass
class ServiceArea(objdict):
    feature_collection: FeatureCollection

@dataclass
class ServiceDeliveryPoint(objdict):
    node: str

@dataclass
class ServiceLocation(objdict):
    node: str

@dataclass
class TransportInterface(objdict):
    point_of_receipt: str
    point_of_delivery: str

@dataclass
class Target(objdict):
    aggregated_p_node: AggregatedPNode = None
    end_device_asset: EndDeviceAsset = None
    meter_asset: MeterAsset = None
    p_node: PNode = None
    service_area: ServiceArea = None
    service_delivery_point: ServiceDeliveryPoint = None
    service_location: ServiceLocation = None
    transport_interface: TransportInterface = None
    group_id: str = None
    group_name: str = None
    resource_id: str = None
    ven_id: str = None
    party_id: str = None

@dataclass
class EventDescriptor(objdict):
    event_id: int
    modification_number: int
    market_context: str
    event_status: str

    created_date_time: datetime = None
    modification_date_time: datetime = None
    priority: int = 0
    test_event: bool = False
    vtn_comment: str = None

    def __post_init__(self):
        if self.modification_date_time is None:
            self.modification_date_time = datetime.now(timezone.utc)
        if self.created_date_time is None:
            self.created_date_time = datetime.now(timezone.utc)
        if self.modification_number is None:
            self.modification_number = 0
        if not isinstance(self.test_event, bool):
            self.test_event = False

@dataclass
class ActivePeriod(objdict):
    dtstart: datetime
    duration: timedelta
    tolerance: dict = None
    notification: dict = None
    ramp_up: dict = None
    recovery: dict = None

@dataclass
class Interval(objdict):
    dtstart: datetime
    duration: timedelta
    signal_payload: float
    uid: int = None

@dataclass
class EventSignal(objdict):
    intervals: List[Interval]
    target: Target
    signal_name: str
    signal_type: str
    signal_id: str
    current_value: float

@dataclass
class Event(objdict):
    event_descriptor: EventDescriptor
    active_period: ActivePeriod
    event_signals: EventSignal
    target: Target

@dataclass
class Response(objdict):
    response_code: int
    response_description: str
    request_id: str