Source code for BAC0.core.devices.Points

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 by Christian Tremblay, P.Eng <christian.tremblay@servisys.com>
# Licensed under LGPLv3, see file LICENSE in this source tree.
#
"""
Points.py - Definition of points so operations on Read results are more convenient.
"""

import time
import typing as t
from collections import namedtuple

# --- standard Python modules ---
from datetime import datetime, timedelta

# --- 3rd party modules ---
from bacpypes.primitivedata import CharacterString

try:
    import pandas as pd
    from pandas.io import sql  # noqa E401

    try:
        from pandas import Timestamp
    except ImportError:
        from pandas.lib import Timestamp
    _PANDAS = True
except ImportError:
    _PANDAS = False

from ...tasks.Match import Match, Match_Value

# --- this application's modules ---
from ...tasks.Poll import SimplePoll as Poll
from ..io.IOExceptions import (
    NoResponseFromController,
    RemovedPointException,
    UnknownPropertyError,
    WritePropertyException,
)
from ..utils.notes import note_and_log

# ------------------------------------------------------------------------------


[docs]class PointProperties(object): """ A container for point properties. """ def __init__(self): self.device = None self.name = None self.type = "" self.address = -1 self.description = None self.units_state = None self.simulated: t.Tuple[bool, t.Optional[int]] = (False, None) self.overridden: t.Tuple[bool, t.Optional[int]] = (False, None) self.priority_array = None self.history_size = None self.bacnet_properties = {} def __repr__(self): return "{}".format(self.asdict) @property def asdict(self): return self.__dict__
# ------------------------------------------------------------------------------
[docs]@note_and_log class Point: """ Represents a device BACnet point. Used to NumericPoint, BooleanPoint and EnumPoints. Each point implements a history feature. Each time the point is read, its value (with timestamp) is added to a history table. Histories capture the changes to point values over time. """ _cache_delta = timedelta(seconds=1) def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None, tags=[], ): self._history = namedtuple("_history", ["timestamp", "value"]) self.properties = PointProperties() self._polling_task = namedtuple("_polling_task", ["task", "running"]) self._polling_task.task = None self._polling_task.running = False self._match_task = namedtuple("_match_task", ["task", "running"]) self._match_task.task = None self._match_task.running = False self._history.timestamp = [] self._history.value = [presentValue] self._history.timestamp.append(datetime.now().astimezone()) self.properties.history_size = history_size self.properties.device = device self.properties.name = pointName self.properties.type = pointType self.properties.address = pointAddress self.properties.description = description self.properties.units_state = units_state self.properties.simulated = (False, 0) self.properties.overridden = (False, 0) self.cov_registered = False self.tags = tags self._cache: t.Dict[str, t.Tuple[t.Optional[datetime], t.Any]] = { "_previous_read": (None, None) } @property def value(self): """ Retrieve value of the point """ if ( self._cache["_previous_read"][0] and datetime.now().astimezone() - self._cache["_previous_read"][0] < Point._cache_delta ): return self._cache["_previous_read"][1] try: res = self.properties.device.properties.network.read( "{} {} {} presentValue".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), ), vendor_id=self.properties.device.properties.vendor_id, ) # self._trend(res) except Exception: raise self._cache["_previous_read"] = (datetime.now().astimezone(), res) return res
[docs] def read_priority_array(self): """ Retrieve priority array of the point """ if self.properties.priority_array is not False: try: res = self.properties.device.properties.network.read( "{} {} {} priorityArray".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), ), vendor_id=self.properties.device.properties.vendor_id, ) self.properties.priority_array = res except (ValueError, UnknownPropertyError): self.properties.priority_array = False except Exception as e: raise Exception( "Problem reading : {} | {}".format(self.properties.name, e) )
[docs] def read_property(self, prop): try: return self.properties.device.properties.network.read( "{} {} {} {}".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), prop, ), vendor_id=self.properties.device.properties.vendor_id, ) except Exception as e: raise Exception("Problem reading : {} | {}".format(self.properties.name, e))
[docs] def update_bacnet_properties(self): """ Retrieve bacnet properties for this point To retrieve something general, forcing vendor id 0 """ try: res = self.properties.device.properties.network.readMultiple( "{} {} {} all".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), ), vendor_id=self.properties.device.properties.vendor_id, show_property_name=True, ) for each in res: if not each: continue v, prop = each self.properties.bacnet_properties[prop] = v except Exception as e: raise Exception("Problem reading : {} | {}".format(self.properties.name, e))
@property def bacnet_properties(self): if not self.properties.bacnet_properties: self.update_bacnet_properties() return self.properties.bacnet_properties @property def is_overridden(self): self.read_priority_array() if self.properties.priority_array is False: return False if self.priority(8) or self.priority(1): self.properties.overridden = (True, self.value) return True else: return False
[docs] def priority(self, priority=None): if self.properties.priority_array is False: return None self.read_priority_array() if not priority: return self.properties.priority_array.debug_contents() if priority < 1 or priority > 16: raise IndexError("Please provide priority to read (1-16)") else: pa = self.properties.priority_array.value[priority] try: key, value = zip(*pa.dict_contents().items()) if key[0] == "null": return None else: return (key[0], value[0]) except ValueError: return None
def _trend(self, res: float) -> None: now = datetime.now().astimezone() self._history.timestamp.append(now) self._history.value.append(res) if self.properties.device.properties.network.database: self.properties.device.properties.network.database.write_points_lastvalue_to_db( [self] ) if self.properties.history_size is None: return else: if self.properties.history_size < 1: self.properties.history_size = 1 if len(self._history.timestamp) >= self.properties.history_size: try: self._history.timestamp = self._history.timestamp[ -self.properties.history_size : # noqa E203 ] self._history.value = self._history.value[ -self.properties.history_size : # noqa E203 ] assert len(self._history.timestamp) == len(self._history.value) except Exception: self._log.exception("Can't append to history") @property def units(self): """ Should return units """ raise Exception("Must be overridden") @property def lastValue(self): """ returns: last value read """ if _PANDAS: last_val = self.history.dropna() last_val_clean = None if len(last_val) == 0 else last_val.iloc[-1] return last_val_clean else: return self._history.value[-1] @property def lastTimestamp(self): """ returns: last timestamp read """ if _PANDAS: last_val = self.history.dropna() last_val_clean = None if len(last_val) == 0 else last_val.index[-1] return last_val_clean else: return self._history.timestamp[-1] @property def history(self) -> t.Dict[datetime, t.Union[int, float, str]]: """ returns : (pd.Series) containing timestamp and value of all readings """ if not _PANDAS: return dict(zip(self._history.timestamp, self._history.value)) idx = self._history.timestamp.copy() his_table = pd.Series(index=idx, data=self._history.value[: len(idx)]) del idx his_table.name = ("{}/{}").format( self.properties.device.properties.name, self.properties.name ) his_table.units = self.properties.units_state if self.properties.name in self.properties.device.binary_states: his_table.states = "binary" elif self.properties.name in self.properties.device.multi_states: his_table.states = "multistates" else: his_table.states = "analog" his_table.description = self.properties.description his_table.datatype = self.properties.type return his_table
[docs] def clear_history(self): self._history.timestamp = [] self._history.value = []
[docs] def chart(self, remove=False): """ Add point to the bacnet trending list """ if remove: self.properties.device.properties.network.remove_trend(self) else: self.properties.device.properties.network.add_trend(self)
def __getitem__(self, key): """ Way to get points... presentValue, status, flags, etc... :param key: state :returns: list of enum states """ if str(key).lower() in ["unit", "units", "state", "states"]: key = "units_state" try: return getattr(self.properties, key) except AttributeError: try: if "@prop_" in key: key = key.split("prop_")[1] return self.read_property(key) except Exception: raise ValueError("Cannot find property named {}".format(key))
[docs] def write(self, value, *, prop="presentValue", priority=""): """ Write to present value of a point :param value: (float) numeric value :param prop: (str) property to write. Default = presentValue :param priority: (int) priority to which write. """ if prop == "description": self.update_description(value) else: if priority != "": if ( isinstance(float(priority), float) and float(priority) >= 1 and float(priority) <= 16 ): priority = "- {}".format(priority) else: raise ValueError("Priority must be a number between 1 and 16") try: self.properties.device.properties.network.write( "{} {} {} {} {} {}".format( self.properties.device.properties.address, self.properties.type, self.properties.address, prop, value, priority, ), vendor_id=self.properties.device.properties.vendor_id, ) except NoResponseFromController: raise # Read after the write so history gets updated. self.value
[docs] def default(self, value): self.write(value, prop="relinquishDefault")
[docs] def sim(self, value, *, force=False): """ Simulate a value. Sets the Out_Of_Service property- to disconnect the point from the controller's control. Then writes to the Present_Value. The point name is added to the list of simulated points (self.simPoints) :param value: (float) value to simulate """ if ( not self.properties.simulated[0] or self.properties.simulated[1] != value or force is not False ): self.properties.device.properties.network.sim( "{} {} {} presentValue {}".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), str(value), ) ) self.properties.simulated = (True, value)
[docs] def out_of_service(self): """ Sets the Out_Of_Service property [to True]. """ self.properties.device.properties.network.out_of_service( "{} {} {}".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), ) ) self.properties.simulated = (True, None)
[docs] def release(self): """ Clears the Out_Of_Service property [to False] - so the controller regains control of the point. """ self.properties.device.properties.network.release( "{} {} {}".format( self.properties.device.properties.address, self.properties.type, str(self.properties.address), ) ) self.properties.simulated = (False, None)
[docs] def ovr(self, value): self.write(value, priority=8) self.properties.overridden = (True, value)
[docs] def auto(self): self.write("null", priority=8) self.properties.overridden = (False, 0)
[docs] def release_ovr(self): self.write("null", priority=1) self.write("null", priority=8) self.properties.overridden = (False, None)
def _setitem(self, value): """ Called by _set, will trigger right function depending on point type to write to the value and make tests. This is default behaviour of the point : AnalogValue are written to AnalogOutput are overridden """ if "characterstring" in self.properties.type: self.write(value) elif "Value" in self.properties.type: if str(value).lower() == "auto": raise ValueError( "Value was not simulated or overridden, cannot release to auto" ) # analog value must be written to self.write(value) elif "Output" in self.properties.type: # analog output must be overridden if str(value).lower() == "auto": self.auto() else: self.ovr(value) else: # input are left... must be simulated if str(value).lower() == "auto": self.release() else: self.sim(value) def _set(self, value): """ Allows the syntax: device['point'] = value """ raise NotImplementedError("Must be overridden")
[docs] def poll(self, command="start", *, delay: int = 10) -> None: """ Poll a point every x seconds (delay=x sec) Stopped by using point.poll('stop') or .poll(0) or .poll(False) or by setting a delay = 0 """ if ( str(command).lower() == "stop" or command is False or command == 0 or delay == 0 ): if isinstance(self._polling_task.task, Poll): self._polling_task.task.stop() self._polling_task.task = None self._polling_task.running = False elif self._polling_task.task is None: self._polling_task.task = Poll(self, delay=delay) self._polling_task.task.start() self._polling_task.running = True elif self._polling_task.running: self._polling_task.task.stop() self._polling_task.running = False self._polling_task.task = Poll(self, delay=delay) self._polling_task.task.start() self._polling_task.running = True else: raise RuntimeError("Stop polling before redefining it")
[docs] def match(self, point, *, delay=5): """ This allow functions like : device['status'].match('command') A fan status for example will follow the command... """ if self._match_task.task is None: self._match_task.task = Match(command=point, status=self, delay=delay) self._match_task.task.start() self._match_task.running = True elif self._match_task.running and delay > 0: self._match_task.task.stop() self._match_task.running = False time.sleep(1) self._match_task.task = Match(command=point, status=self, delay=delay) self._match_task.task.start() self._match_task.running = True elif self._match_task.running and delay == 0: self._match_task.task.stop() self._match_task.running = False else: raise RuntimeError("Stop task before redefining it")
[docs] def match_value(self, value, *, delay=5, use_last_value=False): """ This allow functions like : device['point'].match('value') A sensor will follow a calculation... """ if self._match_task.task is None: self._match_task.task = Match_Value( value=value, point=self, delay=delay, use_last_value=use_last_value ) self._match_task.task.start() self._match_task.running = True elif self._match_task.running and delay > 0: self._match_task.task.stop() self._match_task.running = False time.sleep(1) self._match_task.task = Match_Value( value=value, point=self, delay=delay, use_last_value=use_last_value ) self._match_task.task.start() self._match_task.running = True elif self._match_task.running and delay == 0: self._match_task.task.stop() self._match_task.running = False else: raise RuntimeError("Stop task before redefining it")
def __len__(self): """ Length of a point = # of history records """ return len(self.history)
[docs] def subscribe_cov(self, confirmed=True, lifetime=None, callback=None): address = self.properties.device.properties.address obj_tuple = (self.properties.type, int(self.properties.address)) self.properties.device.properties.network.cov( address, obj_tuple, confirmed=confirmed, lifetime=lifetime, callback=callback, )
[docs] def cancel_cov(self, callback=None): address = self.properties.device.properties.address obj_tuple = (self.properties.type, int(self.properties.address)) self.properties.device.properties.network.cancel_cov( address, obj_tuple, callback=callback )
[docs] def update_description(self, value): """ This will write to the BACnet point and modify the description of the object """ self.properties.device.properties.network.send_text_write_request( addr=self.properties.device.properties.address, obj_type=self.properties.type, obj_inst=int(self.properties.address), value=value, prop_id="description", ) self.properties.description = self.read_property("description")
[docs] def tag(self, tag_id, tag_value, lst=None): """ Add tag to point. Those tags can be used to make queries, add information, etc. They will be included in InfluxDB is used. """ if lst is None: self.tag.append((tag_id, tag_value)) else: for each in lst: tag_id, tag_value = each self.tag.append((tag_id, tag_value))
# ------------------------------------------------------------------------------
[docs]class NumericPoint(Point): """ Representation of a Numeric value """ def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None, ): Point.__init__( self, device=device, pointType=pointType, pointAddress=pointAddress, pointName=pointName, description=description, presentValue=presentValue, units_state=units_state, history_size=history_size, ) @property def value(self): res = super().value self._trend(res) return res @property def units(self): return self.properties.units_state def _set(self, value): if str(value).lower() == "auto": self._setitem(value) else: try: if isinstance(value, Point): value = value.lastValue val = float(value) if isinstance(val, float): self._setitem(value) except Exception as error: raise WritePropertyException( "Problem writing to device : {}".format(error) ) def __repr__(self): try: polling = self.properties.device.properties.pollDelay if (polling < 90 and polling > 0) or self.cov_registered: val = float(self.lastValue) else: val = float(self.value) except ValueError: self._log.error( "Cannot convert value {}. Device probably disconnected or the response is inconsistent".format( self.value ) ) # Probably disconnected return "{}/{} : (n/a) {}".format( self.properties.device.properties.name, self.properties.name, self.properties.units_state, ) return "{}/{} : {:.2f} {}".format( self.properties.device.properties.name, self.properties.name, val, self.properties.units_state, ) def __add__(self, other): return self.value + other __radd__ = __add__ def __sub__(self, other): return self.value - other def __rsub__(self, other): return other - self.value def __mul__(self, other): return self.value * other __rmul__ = __mul__ def __truediv__(self, other): return self.value / other def __rtruediv__(self, other): return other / self.value def __lt__(self, other): return self.value < other def __le__(self, other): return self.value <= other def __eq__(self, other): return self.value == other def __gt__(self, other): return self.value > other def __ge__(self, other): return self.value >= other
# ------------------------------------------------------------------------------
[docs]class BooleanPoint(Point): """ Representation of a Boolean value """ def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None, ): Point.__init__( self, device=device, pointType=pointType, pointAddress=pointAddress, pointName=pointName, description=description, presentValue=presentValue, units_state=units_state, history_size=history_size, ) def _trend(self, res): res = "1: active" if res == "active" else "0: inactive" super()._trend(res) @property def value(self): """ Read the value from BACnet network """ res = super().value self._trend(res) if res == "inactive": self._key = 0 self._boolKey = False else: self._key = 1 self._boolKey = True return res @property def boolValue(self): """ returns : (boolean) Value """ if ":" in self.lastValue: _val = int(self.lastValue.split(":")[0]) else: _val = self.lastValue if _val in [1, "active"]: self._key = 1 self._boolKey = True else: self._key = 0 self._boolKey = False return self._boolKey @property def units(self): """ Boolean points don't have units """ return None def _set(self, value): try: if value is True: self._setitem("active") elif value is False: self._setitem("inactive") elif str(value) in ["inactive", "active"] or str(value).lower() == "auto": self._setitem(value) else: raise ValueError( 'Value must be boolean True, False or "active"/"inactive"' ) except (Exception, ValueError) as error: raise WritePropertyException("Problem writing to device : {}".format(error)) def __repr__(self): polling = self.properties.device.properties.pollDelay if (polling >= 90 or polling <= 0) and not self.cov_registered: # Force reading self.value return "{}/{} : {}".format( self.properties.device.properties.name, self.properties.name, self.boolValue ) def __or__(self, other): return self.boolValue | other def __and__(self, other): return self.boolValue & other def __xor__(self, other): return self.boolValue ^ other def __eq__(self, other): return self.boolValue == other
# ------------------------------------------------------------------------------
[docs]class EnumPoint(Point): """ Representation of an Enumerated (multiState) value """ def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, presentValue=None, units_state=None, history_size=None, ): Point.__init__( self, device=device, pointType=pointType, pointAddress=pointAddress, pointName=pointName, description=description, presentValue=presentValue, units_state=units_state, history_size=history_size, ) def _trend(self, res): res = "{}: {}".format(res, self.get_state(res)) super()._trend(res) @property def value(self): res = super().value # self._log.info("Value : {}".format(res)) # self._log.info("EnumValue : {}".format(self.get_state(res))) self._trend(res) return res
[docs] def get_state(self, v): try: # errors caught below return self.properties.units_state[v - 1] # type: ignore[index] except (TypeError, IndexError): return "n/a"
@property def enumValue(self): """ returns: (str) Enum state value """ try: if ":" in self.lastValue: _val = int(self.lastValue.split(":")[0]) return self.get_state(_val) except TypeError: # probably first occurence of history... retry return self.get_state(self.lastValue) except IndexError: value = "unknown" except ValueError: value = "NaN" return value @property def units(self): """ Enums have 'state text' instead of units. """ return None def _set(self, value): try: if isinstance(value, int): self._setitem(value) elif str(value) in self.properties.units_state: # type: ignore[operator] self._setitem(self.properties.units_state.index(value) + 1) elif str(value).lower() == "auto": self._setitem("auto") else: raise ValueError( "Value must be integer or correct enum state : {}".format( self.properties.units_state ) ) except (Exception, ValueError) as error: raise WritePropertyException("Problem writing to device : {}".format(error)) def __repr__(self): polling = self.properties.device.properties.pollDelay if (polling >= 90 or polling <= 0) and not self.cov_registered: # Force reading self.value return "{}/{} : {}".format( self.properties.device.properties.name, self.properties.name, self.enumValue ) def __eq__(self, other): return self.value == self.properties.units_state.index(other) + 1
[docs]class StringPoint(Point): """ Representation of CharacterString value """ def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, units_state=None, presentValue=None, history_size=None, ): Point.__init__( self, device=device, pointType=pointType, pointAddress=pointAddress, pointName=pointName, description=description, presentValue=presentValue, history_size=history_size, ) @property def units(self): """ Characterstring value do not have units or state text """ return None def _trend(self, res): super()._trend(res) @property def value(self): res = super().value self._trend(res) return res def _set(self, value): try: if isinstance(value, str): self._setitem(value) elif isinstance(value, CharacterString): self._setitem(value.value) else: raise ValueError("Value must be string or CharacterString") except (Exception, ValueError) as error: raise WritePropertyException("Problem writing to device : {}".format(error)) def __repr__(self): try: polling = self.properties.device.properties.pollDelay if (polling < 90 and polling > 0) or self.cov_registered: val = str(self.lastValue) else: val = str(self.value) except ValueError: self._log.error("Cannot convert value. Device probably disconnected") # Probably disconnected val = None return "{}/{} : {}".format( self.properties.device.properties.name, self.properties.name, val ) def __eq__(self, other): return self.value == other.value
[docs]class DateTimePoint(Point): """ Representation of DatetimeValue value """ def __init__( self, device=None, pointType=None, pointAddress=None, pointName=None, description=None, units_state=None, presentValue=None, history_size=None, ): Point.__init__( self, device=device, pointType=pointType, pointAddress=pointAddress, pointName=pointName, description=description, presentValue=presentValue, history_size=history_size, ) @property def units(self): """ Characterstring value do not have units or state text """ return None def _trend(self, res): # super()._trend(res) return @property def value(self): res = super().value # self._trend(res) year, month, day, dayofweek = res.date hour, minutes, seconds, ms = res.time res = datetime(year + 1900, month, day, hour, minutes, seconds) return res def _set(self, value): # try: # if isinstance(value, str): # self._setitem(value) # elif isinstance(value, CharacterString): # self._setitem(value.value) # else: # raise ValueError("Value must be string or CharacterString") # except (Exception, ValueError) as error: # raise WritePropertyException("Problem writing to device : {}".format(error))` raise NotImplementedError("Writing to Datetime is not supported yet") def __repr__(self): try: # polling = self.properties.device.properties.pollDelay # if (polling < 90 and polling > 0) or self.cov_registered: # val = str(self.lastValue) # else: val = self.value except ValueError: self._log.error("Cannot convert value. Device probably disconnected") # Probably disconnected val = None return "{}/{} : {}".format( self.properties.device.properties.name, self.properties.name, val ) def __eq__(self, other): return self.value == other.value def __ge__(self, other): return self.value >= other.value def __le__(self, other): return self.value <= other.value def __gt__(self, other): return self.value > other.value def __lt__(self, other): return self.value < other.value
# ------------------------------------------------------------------------------
[docs]class OfflinePoint(Point): """ When offline (DB state), points needs to behave in a particular way (we can't read on bacnet...) """ def __init__(self, device, name): self.properties = PointProperties() self.properties.device = device dev_name = self.properties.device.properties.db_name try: props = self.properties.device.read_point_prop(dev_name, name) except RemovedPointException: raise self.properties.name = props["name"] self.properties.type = props["type"] self.properties.address = props["address"] self.properties.description = props["description"] self.properties.units_state = props["units_state"] self.properties.simulated = (True, None) self.properties.overridden = (False, None) if "analog" in self.properties.type: self.new_state(NumericPointOffline) elif "multi" in self.properties.type: self.new_state(EnumPointOffline) elif "binary" in self.properties.type: self.new_state(BooleanPointOffline) elif "string" in self.properties.type: self.new_state(StringPointOffline) else: raise TypeError("Unknown point type")
[docs] def new_state(self, newstate): self.__class__ = newstate
[docs]class NumericPointOffline(NumericPoint): @property def history(self): his = self.properties.device._read_from_sql( 'select * from "{}"'.format("history"), self.properties.device.properties.db_name, ) his.index = his["index"].apply(Timestamp) return his.set_index("index")[self.properties.name] @property def value(self): """ Take last known value as the value """ try: value = self.lastValue except IndexError: value = 65535 return value
[docs] def write(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def sim(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def release(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
@property def units(self): return self.properties.units_state def _set(self, value): raise OfflineException("Must be online to write") def __repr__(self): return "{}/{} : {:.2f} {}".format( self.properties.device.properties.name, self.properties.name, self.value, self.properties.units_state, )
[docs]class BooleanPointOffline(BooleanPoint): @property def history(self): his = self.properties.device._read_from_sql( 'select * from "{}"'.format("history"), self.properties.device.properties.db_name, ) his.index = his["index"].apply(Timestamp) return his.set_index("index")[self.properties.name] @property def value(self): try: value = self.lastValue except IndexError: value = "NaN" return value def _set(self, value): raise OfflineException("Point must be online to write")
[docs] def write(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def sim(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def release(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs]class EnumPointOffline(EnumPoint): @property def history(self): his = self.properties.device._read_from_sql( 'select * from "{}"'.format("history"), self.properties.device.properties.db_name, ) his.index = his["index"].apply(Timestamp) return his.set_index("index")[self.properties.name] @property def value(self): """ Take last known value as the value """ try: value = self.lastValue except IndexError: value = "NaN" except ValueError: value = "NaN" return value @property def enumValue(self): """ returns: (str) Enum state value """ try: value = self.properties.units_state[int(self.lastValue) - 1] # type: ignore[index] except IndexError: value = "unknown" except ValueError: value = "NaN" return value def _set(self, value): raise OfflineException("Point must be online to write")
[docs] def write(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def sim(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def release(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs]class StringPointOffline(EnumPoint): @property def history(self): his = self.properties.device._read_from_sql( 'select * from "{}"'.format("history"), self.properties.device.properties.db_name, ) his.index = his["index"].apply(Timestamp) return his.set_index("index")[self.properties.name] @property def value(self): """ Take last known value as the value """ try: value = self.lastValue except IndexError: value = "NaN" except ValueError: value = "NaN" return value def _set(self, value): raise OfflineException("Point must be online to write")
[docs] def write(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def sim(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs] def release(self, value, *, prop="presentValue", priority=""): raise OfflineException("Must be online to write")
[docs]class OfflineException(Exception): pass