Source code for netdef.Sources.BaseSource

import copy
import datetime
import enum

from ..Interfaces.DefaultInterface import DefaultInterface


[docs]class StatusCode(enum.Enum): """ Used to indicate the quality of a value in BaseSource.status_code NONE: Value is not set yet. INITIAL: First value. you might have to update cashes with this value at application startup. GOOD: A normal value update. INVALID: A value update where the value is not to be trusted. """ NONE = 0 INITIAL = 1 GOOD = 2 INVALID = 3
[docs]class BaseSource: def __init__(self, key=None, value=None, controller=None, source=None, rule=None): # verdier # cache self.value = ( value ) # cachet verdi fra get_value eller set_value når disse er behandlet # verdiens statuskode, ved oppstart er den NONE # første innhentede verdi fra controller er INITIAL # alle innhentede verdier etter dette vil ha GOOD eller INVALID self.status_code = StatusCode.NONE # når verdien ble oppdatert. # skal alltid være av typen datetime.datetime # dersom verdi ikke har medfølgende tidsstempel skal # controller selv sette denne med datetime.datetime.utcnow() self.source_time = None self.source_datatype = None # Ny verdi inn fra driver self.get_value = None self.get_source_time = None self.get_status_code = None self.get_origin = None # Ny verdi ut fra uttrykk self.set_value = None self.set_source_time = None self.set_status_code = None self.set_origin = None self.set_callback = None # callback som skal kjøres når verdi settes fra utrykk # kilder self.rule = ( rule ) # kontroller må benytte denne for å sende RUN_EXPRESSION til riktig kø self.controller = ( controller ) # regelmotor må benytte denne for å sende ADD_SOURCE til riktig kontroller self.source = ( source ) # kan brukes av kontroller eller uttrykk for å verifisere kilde-typen self.key = ( key ) # unik identifikasjon for kilden. kommer typisk fra en konfigfil som er parset av regelmotor # dette er en valgfri "wrapper" til verdien fra kontrolleren. # hensikten er å gjøre manipulering av verdi så enkel som mulig i uttrykk self.interface = DefaultInterface def __str__(self): # brukes til søk og print av kildedata return self.get_reference() + " R: {} V:{}".format( self.rule, self.value_as_string )
[docs] def get_reference(self): """ Used to identify similar sources. if two instances return the same reference this means that one instance is redundant and can be replaced """ # Brukers av Rule. # Benyttes til å identifisere like kilder. hvis to instanser returnerer samme referanse # betyr dette at den ene instansen er overflødig og kan erstattes return "C:{} S:{} K:{}".format(self.controller, self.source, self.key)
@property def value_as_string(self): """ Is primarily used by web interfaces to display value in table. Can be overridden to limit the display of large data. Example:: @property def value_as_string(self): if self.value and isinstance(self.value, bytes): n = len(self.value) return "<{}...><data len:{}>".format(self.value[:10], n) else: return super().value_as_string """ # brukes primært av webgrensesnitt til å vise verdi i tabell. # bør overstyres for å begrense visning av store data return str(self.value)
[docs] @staticmethod def pack_subitems(value): """ Creates output that can be used to query for a list of inputs """ # Benyttes av controller. Lager utdata som kan benyttes til å spørre # etter etter en liste med inndata return None
[docs] @staticmethod def can_unpack_subitems(value): """ Function that confirms / decides on input data a known list. If so, then unpack_subitems can be used afterwards. Example:: def parse_response(self, response): for parser in self.get_parsers(): if parser.can_unpack_subitems(response): yield from parser.unpack_subitems(response) """ # """ Benyttes av controller. Funksjon som bekrefter/avkrefter om inndata er # en kjent liste av noe slag. hvis ja, så blir unpack_subitems brukt etterpå. # """ return False # kilder må overstyre denne. den er default av.
[docs] @staticmethod def unpack_subitems(value): """ Function that parses response from source and yield items found in value. This can be overridden and adapted to the controller it is to be used in. Example:: def parse_response(self, response): for parser in self.get_parsers(): if parser.can_unpack_subitems(response): yield from parser.unpack_subitems(response) """ # """ Benyttes av controller. Funksjon som parser svar fra kilde og # yielder items funnet i verdi. Denne kan overstyres og # tilpasses controller den skal brukes mot. # """ yield None
[docs] @staticmethod def can_unpack_value(value): """ Function that confirms / determines if the input data is compatible with this class. If so, unpack_value should be used afterwards. Example:: def parse_item(self, item): for parser in self.get_parsers(): if parser.can_unpack_value(item): key, source_time, value = parser.unpack_value(item) self.send_datachange(key, source_time, value) """ # """ Benyttes av controller. Funksjon som bekrefter/avkrefter om inndata er # kompatiblet med denne klassen. hvis ja, så blir unpack_value brukt etterpå. # """ return False # kilder må overstyre denne. den er default av.
[docs] @staticmethod def unpack_value(key, source_time, value): """ Function that parses response from source and returns following tuple: (key, source_time, value) Key can then be used to find the right instance and update values. Can be overridden and adapted to the controller it is to be used in. :returns: tuple(key, source_time, value) :rtype: tuple Example:: def parse_item(self, item): for parser in self.get_parsers(): if parser.can_unpack_value(item): key, source_time, value = parser.unpack_value(item) self.send_datachange(key, source_time, value) """ # """ Benyttes av controller. Funksjon som parser svar fra kilde og # returnererfølgende tuple: (key, source_time, value) # key kan så benyttes til å finne riktig instanse og oppdatere verdier # denne kan overstyres og tilpasses controller den skal brukes mot. # """ return key, source_time, value
[docs] def pack_value(self, value): """ Function that converts key and values into a format that the source uses. Can be overridden and adapted to the controller it is to be used in. Example:: def handle_write_source(self, incoming, value, source_time): data = incoming.pack_value(value, source_time) topic, payload = incoming.make_message(incoming.key, data) self.publish_data_item(topic, payload) """ # """ Benyttes av controller. Funksjon som gjør om key og verdier til et format # som kilden benytter. Denne kan overstyres og tilpasses controller den # skal brukes mot. # """ return self.key, value
[docs] def pack_add_source(self): """ Used if source must be added to external system. I.e. a subscription. Can be overridden and customized. """ # """ Benyttes av controller. Dersom kilde må legges til i eksternt system # for å endringer tilbake så kan denne funksjonen overstyres og tilpasses. # Den er default av. # """ return False
[docs] def copy_value(self): """ Shallow copy of the value """ return copy.copy(self.value)
[docs] def copy_get_value(self): """ Shallow copy of the value """ return copy.copy(self.get_value)
@property def get(self): """ Get the value that is updated by the controller """ return self.get_value @get.setter def get(self, val): """ Set the value """ # controller skriver til get_value med denne funksjonen self.get_value = val self.value = val @property def set(self): """ Get the value that is updated by expressions """ return self.set_value @set.setter def set(self, val): """ Set the value """ # Utrykk skriver til set_value med denne funsjonen if isinstance(val, DefaultInterface): val = val.value self._set_set_value(val, None, True, "expression") def _set_set_value(self, value, stime, status_ok, origin): """ Backend for `set` and `set_value_from_string` :param value: value to be set :param (None or datetime.datetime) stime: timestamp when the value was changed :param bool status_ok: True if value is good :param str origin: who set the value """ self.set_value = value self.set_status_code = StatusCode.GOOD if status_ok else StatusCode.INVALID self.set_source_time = datetime.datetime.utcnow() if stime is None else stime self.set_origin = origin self.value = value if self.set_callback: self.set_callback(self, value, self.set_source_time)
[docs] def register_set_callback(self, set_callback): """ Register the callback that sends WRITE_SOURCE message to the controller queue. """ # Regelmotor kaller denne funksjonen. # callback setter en WRITE_SOURCE melding til kontrollerens kø. self.set_callback = set_callback
[docs] def can_set_value_from_string(self): """ Returns True if the value can be converted from string to its given datatype. Only `builtins.int`, `str` and `float` have built-in support, but aditional types can be implemented by this function and `set_value_from_string` """ if isinstance(self.set_value, (int, str, float)): return True elif isinstance(self.get_value, (int, str, float)): return True else: return False
[docs] def set_value_from_string(self, value, stime=None, status_ok=True, origin=""): """ Converts given value to correct datatype and sends a WRITE_SOURCE message to controller. This function is called when a value change is triggered from :menuselection:`Webadmin --> Sources --> Edit` :param value: value to be set :param (None or datetime.datetime) stime: timestamp when the value was changed :param bool status_ok: True if value is good :param str origin: who set the value """ if isinstance(self.set_value, (int, str, float)): val = type(self.set_value)(value) elif isinstance(self.get_value, (int, str, float)): val = type(self.get_value)(value) else: val = str(value) self._set_set_value(val, stime, status_ok, origin)