kheldaroz402
commented
1 month ago Got the debounce working but would still like to know if i can detect where a change is coming from
`class AnalogOutputFeedbackObject(AnalogOutputCmdObject): debounce_interval = 3 # 500 milliseconds debounce interval
def __init__(self, *args, **kwargs):
if _debug:
AnalogOutputFeedbackObject._debug("__init__ %r %r", args, kwargs)
super().__init__(*args, **kwargs)
# Initialize the objectIdentifier monitor
self._property_monitors.setdefault('objectIdentifier', []).append(self.objectIdentifier)
# Listen for changes to the present value
self._property_monitors["presentValue"].append(self.check_feedback)
logging.info(f"{color.GREEN}AnalogOutputFeedbackObject initialized for {self.objectName}{color.END}")
# Initialize debounce timer attribute using a private name to avoid conflicts
self._last_change_time = time.time()
def get_lowest_priority_value(self):
"""
Get the lowest non-null priority value from the priorityArray.
"""
for i, value in enumerate(self.priorityArray):
if value is not None:
# Handle different BACnet value types
if hasattr(value, 'real'):
value = value.real
elif hasattr(value, 'integer'):
value = value.integer
elif hasattr(value, 'unsigned'):
value = value.unsigned
elif hasattr(value, 'boolean'):
value = value.boolean
if value is not None:
return i + 1, value # BACnet priorities are 1-indexed
return 16, None
def check_feedback(self, old_value, new_value):
current_time = time.time()
if new_value == old_value or (current_time - self._last_change_time) < self.debounce_interval:
return
self._last_change_time = current_time
# Ensure the objectIdentifier is available
if 'objectIdentifier' not in self._property_monitors or not self._property_monitors['objectIdentifier']:
logging.error(f"{color.RED}AnalogOutputFeedbackObject: objectIdentifier not available in property monitors{color.END}")
return
object_identifier = self._property_monitors['objectIdentifier'][0]
object_identifier_value = object_identifier[1]
# Get the lowest non-null priority value
priority, priority_value = self.get_lowest_priority_value()
logging.info(f"{color.GREEN}AnalogOutputFeedbackObject (ID: {object_identifier_value}, Name: {self.objectName}): Detected change from {old_value} to {new_value} with priority {priority} and priority value {priority_value}{color.END}")
# Determine the value to update in the database
if priority != 16:
present_value_to_update = priority_value
else:
present_value_to_update = new_value
# Set Override_Value if priority is not 16
override_value = present_value_to_update if priority != 16 else None
# Set Override to 1 in the database if priority is 8
override_flag = 1 if priority == 8 else 0
mydbProgram = None
mycursorProgram = None
try:
mydbProgram = poolProgram.get_connection()
mycursorProgram = mydbProgram.cursor(dictionary=True, buffered=True)
sql = (
"UPDATE Objects "
"SET Present_Value = %s, Updated_By = 'Bacnet Feedback', Priority = %s, Override_Value = %s, Override = %s "
"WHERE Object_Identifier = %s"
)
mycursorProgram.execute(sql, (present_value_to_update, priority, override_value, override_flag, object_identifier_value))
mydbProgram.commit()
logging.info(f"{color.YELLOW}AnalogOutputFeedbackObject: Database updated for Object_Identifier {object_identifier_value} with Present_Value {present_value_to_update}, Priority {priority}, Override_Value {override_value}, Override {override_flag}{color.END}")
except (Exception) as e:
logging.error(f"{color.RED}AnalogOutputFeedbackObject: An error occurred while updating the database: {e}{color.END}")
finally:
if mycursorProgram:
mycursorProgram.close()
if mydbProgram:
mydbProgram.close()`
bbartling
The issue that I'm having is that an external device makes a change to a Point and before the database has been updated, the BACnet stack has been updated by the process that checks the database, so the external change gets "lost" (its intermitent of course, since sometime the database is updated)
`# Function to create Binary Output Object def create_binary_output(row): try: ArrayOfPropertyIdentifier = ArrayOf(PropertyIdentifier) property_list = ArrayOfPropertyIdentifier([ 'presentValue', 'statusFlags', 'eventState', 'inactiveText', 'activeText', 'outOfService', 'description', 'reliability', 'polarity' ])
@bacpypes_debugging @register_object_type(vendor_id=47) class BinaryOutputFeedbackObject(BinaryOutputCmdObject): def init(self, *args, *kwargs): super().init(args, **kwargs)
`
(green text)
I've tried to implement a "debounce" feature, but that just complains that its unknown property
`from bacpypes.local.object import AnalogOutputCmdObject import time import logging
class AnalogOutputFeedbackObject(AnalogOutputCmdObject): def init(self, *args, *kwargs): self.last_update_time = 0 self.debounce_interval = 1 # seconds if _debug: AnalogOutputFeedbackObject._debug("init %r %r", args, kwargs) super().init(args, **kwargs)
`