Realarm and Reallocate¶

`JobRealarmReallocate(baze)` ¶

Class used for handling realarm and reallocate jobs.

Source code in echo_baze/baze_root.py

def __init__(self, baze: e_bz.Baze) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    baze : Baze
        Top level object carrying all functionality and the connection handler.

    """
    # check inputs
    if not isinstance(baze, e_bz.Baze):
        raise ValueError(f"baze must be of type Baze, not {type(baze)}")

    self.baze: e_bz.Baze = baze

`insert(period, object_names, realarm=True, realarm_settings=None, reallocate=True, reallocate_settings=None, time_zone='local')` ¶

Schedule realarm and/or reallocate jobs for the desired objects.

Parameters:

period ¶
(DateTimeRange) –

The desired period
object_names ¶
(list[str]) –

List of object names to realarm and/or reallocate
realarm ¶
(bool, default: True ) –

If realarm should be done, by default True
realarm_settings ¶
(RealarmSettings | None, default: None ) –

Dict with settings for the realarm job. If None, the default settings (below) will be used. By default None

Default settings: - validateAlarms: True - truncateAlarms: True - truncateAlarmTolerance: 300, - truncateTagShortName: "ActivePower" - truncateThreshold: 100
reallocate ¶
(bool, default: True ) –

If reallocate should be done, by default True
reallocate_settings ¶
(dict[str, Any] | None, default: None ) –

Dict with settings for the reallocate job. If None, the default settings (below) will be used. By default None

Default settings: - calculateLoss: True - validateAllocations: True - allocationTypes: None (will set to "" if None, else will be replaced by the Acronym of the allocation type)
time_zone ¶
( TimeZone, default: 'local' ) –
In which time zone we assume the inputs are. Also used to define in which time zone the output is. There are three options:
- If "UTC" is used, we assume time already is in UTC.
- If local is used, the default time zone defined in echo_baze will be used.
- If an int, must be between -12 and +12
By default "local"

Returns:

dict[str, Any] –

The response of the request. Most useful keys are orderId and pkId

Source code in echo_baze/job_realarmreallocate.py

@validate_call
def insert(
    self,
    period: DateTimeRange,
    object_names: list[str],
    realarm: bool = True,
    realarm_settings: RealarmSettings | None = None,
    reallocate: bool = True,
    reallocate_settings: ReallocateSettings | None = None,
    time_zone: TimeZone = "local",
) -> dict[str, Any]:
    """
    Schedule realarm and/or reallocate jobs for the desired objects.

    Parameters
    ----------
    period : DateTimeRange
        The desired period
    object_names : list[str]
        List of object names to realarm and/or reallocate
    realarm : bool, optional
        If realarm should be done, by default True
    realarm_settings : RealarmSettings | None, optional
        Dict with settings for the realarm job. If None, the default settings (below) will be used. By default None

        Default settings:
        - `validateAlarms`: True
        - `truncateAlarms`: True
        - `truncateAlarmTolerance`: 300,
        - `truncateTagShortName`: "ActivePower"
        - `truncateThreshold`: 100

    reallocate : bool, optional
        If reallocate should be done, by default True
    reallocate_settings : dict[str, Any] | None, optional
        Dict with settings for the reallocate job. If None, the default settings (below) will be used. By default None

        Default settings:
        - `calculateLoss`: True
        - `validateAllocations`: True
        - `allocationTypes`: None (will set to "" if None, else will be replaced by the Acronym of the allocation type)

    time_zone :  TimeZone, optional
        In which time zone we assume the inputs are. Also used to define in which time zone the output is. There are three options:

        - If "UTC" is used, we assume time already is in UTC.
        - If local is used, the default time zone defined in echo_baze will be used.
        - If an int, must be between -12 and +12

        By default "local"

    Returns
    -------
    dict[str, Any]
        The response of the request. Most useful keys are orderId and pkId
    """
    t0 = time.perf_counter()

    if realarm_settings is None:
        realarm_settings = RealarmSettings()
    if reallocate_settings is None:
        reallocate_settings = ReallocateSettings()

    # converting to dict
    realarm_settings = realarm_settings.model_dump()
    reallocate_settings = reallocate_settings.model_dump()

    # checking input
    if reallocate:
        # removing allocationTypes if None
        if reallocate_settings["allocationTypes"] is None:
            reallocate_settings["allocationTypes"] = ""
        else:
            # replacing allocationTypes with the Acronym of the allocation type
            alloc_types = self.baze.allocations.types.get()
            if reallocate_settings["allocationTypes"] not in alloc_types:
                raise ValueError(f"Allocation type not found: {reallocate_settings['allocationTypes']}")
            reallocate_settings["allocationTypes"] = alloc_types[reallocate_settings["allocationTypes"]]["acronym"]

    # making sure at least one of the two is True
    if not realarm and not reallocate:
        raise ValueError("At least one of realarm and reallocate must be True")

    endpoint = "statesyncorders"

    # validating if all objects exist
    objs_def = self.baze.objects.instances.get(object_names=object_names)
    wrong_objs = set(object_names) - set(objs_def.keys())
    if len(wrong_objs) > 0:
        raise ValueError(f"The following objects were not found: {wrong_objs}")

    # defining the order mask
    mask_mapping = {
        "realarm": 4,
        "reallocate": 8,
        "both": 12,
    }
    order_mask = mask_mapping["both" if realarm and reallocate else "realarm" if realarm else "reallocate"]

    payload = {
        "from": convert_time_zone(dt=period.start, old_tz=time_zone, new_tz="UTC").strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
        "to": convert_time_zone(dt=period.end, old_tz=time_zone, new_tz="UTC").strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
        "orderMask": order_mask,  # just replicating what the portal does
        "settings": {},
        "assetIds": [],
    }

    # adding realarm and reallocate keys depending on the input
    if realarm:
        payload["realarm"] = True
    if reallocate:
        payload["reallocate"] = True

    # adding realarm settings
    if realarm:
        # adding truncateTagFilter
        realarm_settings["truncateTagFilter"] = (
            f"(IsQualityGood eq true) and (ValueAsDouble gt {realarm_settings['truncateThreshold']})"
        )
        # adding truncateTagNamePattern
        realarm_settings["truncateTagNamePattern"] = None
        # adding "reimportAlarms"
        realarm_settings["reimportAlarms"] = True
        payload["settings"] |= realarm_settings
    else:
        payload["settings"] |= {
            "validateAlarms": False,
            "truncateAlarms": False,
            "truncateAlarmTolerance": 300,
            "truncateTagShortName": None,
            "truncateThreshold": False,
            "truncateTagFilter": None,
            "truncateTagNamePattern": None,
            "reimportAlarms": False,
        }

    # adding reallocate settings
    if reallocate:
        # adding reallocateAlarms
        reallocate_settings["reallocateAlarms"] = True
        payload["settings"] |= reallocate_settings
    else:
        payload["settings"] |= {
            "calculateLoss": False,
            "validateAllocations": False,
            "reallocateAlarms": False,
        }

    # adding assets and assetIds
    for obj_name in object_names:
        payload["assetIds"].append(objs_def[obj_name]["attributes"]["objectHandle"])

    t1 = time.perf_counter()
    result = self.baze.conn.post(endpoint, json=payload)
    self._handle_http_errors(result)
    logger.debug(
        f"Request response for {object_names} {period.start:%Y-%m-%d %H:%M:%S} to {period.end:%Y-%m-%d %H:%M:%S} in {time.perf_counter() - t1:.3f} s",
    )

    logger.debug(f"Created realarm/reallocate job in {time.perf_counter() - t0:.3f} s")

    result = result.json()

    if result["state"] != "Created":
        raise ValueError(f"Error in realarm/reallocate job creation: {result}")

    return result

Realarm and Reallocate¶

`JobRealarmReallocate(baze)` ¶

`insert(period, object_names, realarm=True, realarm_settings=None, reallocate=True, reallocate_settings=None, time_zone='local')` ¶

`period` ¶

`object_names` ¶

`realarm` ¶

`realarm_settings` ¶

`reallocate` ¶

`reallocate_settings` ¶

`time_zone` ¶

Realarm and Reallocate¶

JobRealarmReallocate(baze) ¶

insert(period, object_names, realarm=True, realarm_settings=None, reallocate=True, reallocate_settings=None, time_zone='local') ¶

period ¶

object_names ¶

realarm ¶

realarm_settings ¶

reallocate ¶

reallocate_settings ¶

time_zone ¶

`JobRealarmReallocate(baze)` ¶

`insert(period, object_names, realarm=True, realarm_settings=None, reallocate=True, reallocate_settings=None, time_zone='local')` ¶

`period` ¶

`object_names` ¶

`realarm` ¶

`realarm_settings` ¶

`reallocate` ¶

`reallocate_settings` ¶

`time_zone` ¶