External Device writing to one of My Objects about bacpypes

[kheldaroz402]

it appears that I need to use these objects
from bacpypes.local.object import (
AnalogValueCmdObject,
AnalogOutputCmdObject,
BinaryOutputCmdObject,
BinaryValueCmdObject,
DateValueCmdObject,
MultiStateValueCmdObject,
MultiStateOutputCmdObject

)

according to CommandableMixin.py

apparently there is a commandable_feedback.py in the sandbox that might help :)

[kheldaroz402]

updated class with objectIdentifier

@bacpypes_debugging
@register_object_type(vendor_id=47)
class AnalogOutputFeedbackObject(AnalogOutputCmdObject):
    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', [])
        self._property_monitors['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}")

    def check_feedback(self, old_value, new_value):
        # 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]

        logging.info(f"{color.GREEN}AnalogOutputFeedbackObject (ID: {object_identifier_value}, Name: {self.objectName}): Detected change from {old_value} to {new_value}{color.END}")

        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' "
                "WHERE Object_Identifier = %s"
            )

            mycursorProgram.execute(sql, (new_value, object_identifier_value))
            mydbProgram.commit()
            logging.info(f"{color.GREEN}AnalogOutputFeedbackObject: Database updated for Object_Identifier {object_identifier_value} with Present_Value {new_value}{color.END}")

        except (MySQLError, 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()

[JoelBender]

I'm glad that you got this to work! When it becomes time to move this over to BACpypes3 I'm hoping that you'll find it a lot easier.

[kheldaroz402]

So I've made progress but not quiet there yet - it appear to be updating the bacnet stack when i use YABe to change a value, but then it goes back to the previous database value, so I think the issue is in how I check for the value being updated

creation of the points (AV example)

`# Function to create Analog Value Object
def create_analog_value(row):
try:
ArrayOfPropertyIdentifier = ArrayOf(PropertyIdentifier)
property_list = ArrayOfPropertyIdentifier([
'presentValue', 'statusFlags', 'eventState', 'outOfService', 'units',
'minPresValue', 'maxPresValue', 'resolution', 'covIncrement', 'description',
'relinquishDefault'
])
statusFlags = calculate_status_flags(row)
analog_value = AnalogValueCmdObject(
objectName=row['Object_Name'],
objectIdentifier=(AnalogValueObject.objectType, row['Object_Identifier']),
presentValue=row['Present_Value'],
statusFlags=statusFlags,
eventState=row.get('Event_State', 0),
outOfService=True if row['Out_Of_Service'] == 1 else False,
covIncrement=row.get('COV_Increment', 1),
units=row['BACnet_Engineering_Units'],
description=row['Description'],
minPresValue=row['Min_Pres_Value'],
maxPresValue=row['Max_Pres_Value'],
resolution=row.get('Resolution', None),
relinquishDefault=row.get('Relinquish_Default', 0),
propertyList=property_list,
)

    # Initialize the priority array with default values
    priority_array = PriorityArray()

    # Set the Present_Value at the specified priority position if within range
    if 'Priority' in row and 1 <= row['Priority'] <= 16:
        priority_index = row['Priority']   # Don't need to Adjust for 0-based index
        priority_array[priority_index] = PriorityValue(real=Real(row['Present_Value']))

    # Assign the priority array to the object
    analog_value.priorityArray = priority_array

    # Print out the priority array for debugging
    logging.info(f"Priority Array Contents: Current Priority is {row['Priority']} with a value of {row['Present_Value']}")
    # for i, priority_value in enumerate(priority_array):
    #     if priority_value:
    #         logging.info(f"Priority {i + 1}: {priority_value.real}")
    #     else:
    #         logging.info(f"Priority {i + 1}: None")

    logging.debug(f"Created AnalogValueObject: {analog_value}")

    return analog_value

except Exception as e:
    logging.error(f"{color.RED}An error occurred in create_analog_value: {e}{color.END}")
    return None`

`def main():
global vendor_id, this_application, this_device, device_address, object_identifier, property_list
reset_subscribed()
# register the classes
register_object_type(LocalDeviceObject, vendor_id=47)

register_object_type(AnalogValueCmdObject, vendor_id=47)
register_object_type(AnalogOutputCmdObject, vendor_id=47)
register_object_type(BinaryValueCmdObject, vendor_id=47)
register_object_type(BinaryOutputCmdObject, vendor_id=47)
register_object_type(MultiStateOutputCmdObject, vendor_id=47)
register_object_type(MultiStateValueCmdObject, vendor_id=47)
# make a simple application
this_application = WebAIApplication(this_device, address)
# create the database objects and add them to the application
create_objects_database(this_application)
# convenience definition

deferred(update_WebAI_data)
deferred(WhoIsRequest)
deferred(device_discovered)
deferred(bacnet_subscribe_import)
deferred(bacnet_subscribe_screens)
deferred(bacnet_subscribe_logic)
deferred(bacnet_logging)
deferred(bacnet_write)
deferred(bacnet_write_release)
deferred(bacnet_time)
deferred(bacnet_time_get_logic)
deferred(bacnet_time_get_screens)
deferred(enumerate_device)
deferred(updateCOV)

# enable sleeping will help with threads
enable_sleeping()

if _debug:
    _log.debug("running")

run()

if _debug:
    _log.debug("fini")

if name == "main":
main()
`

`@bacpypes_debugging
@register_object_type(vendor_id=47)
class AnalogValueFeedbackObject(AnalogValueCmdObject):
def init(self, *args, **kwargs):
if _debug:
AnalogValueFeedbackObject._debug("init %r %r", args, kwargs)
super().init(*args, **kwargs)

    # listen for changes to the present value
    self._property_monitors["presentValue"].append(self.check_feedback)

def check_feedback(self, old_value, new_value):
    if _debug:
        AnalogValueFeedbackObject._debug("check_feedback %r %r", old_value, new_value)
        logging.info(f"AnalogValueFeedbackObject: Detected change from {old_value} to {new_value}")

    mydbProgram = None
    mycursorProgram = None
    try:
        mydbProgram = poolProgram.get_connection()
        mycursorProgram = mydbProgram.cursor(dictionary=True, buffered=True)

        object_identifier = self._property_monitors['objectIdentifier'][0]

        sql = (
            "UPDATE Objects "
            "SET Present_Value = %s, Updated_By = 'Bacnet Feedback' "
            "WHERE Object_Identifier = %s"
        )

        mycursorProgram.execute(sql, (new_value, object_identifier))
        mydbProgram.commit()

        if _debug:
            AnalogValueFeedbackObject._debug("Database updated: %r -> %r", object_identifier, new_value)
            logging.info(f"AnalogValueFeedbackObject: Database updated for Object_Identifier {object_identifier} with Present_Value {new_value}")

    except (MySQLError, Exception) as e:
        logging.error(f"AnalogValueFeedbackObject: An error occurred while updating the database: {e}")
    finally:
        if mycursorProgram:
            mycursorProgram.close()
        if mydbProgram:
            mydbProgram.close()

`

[kheldaroz402]

ok - so I needed to use the feedback class in the creation process like so

`# Function to create Analog Output Object
def create_analog_output(row):
try:
ArrayOfPropertyIdentifier = ArrayOf(PropertyIdentifier)
property_list = ArrayOfPropertyIdentifier([
'presentValue', 'statusFlags', 'eventState', 'outOfService', 'units',
'minPresValue', 'maxPresValue', 'resolution', 'covIncrement', 'description',
'highLimit', 'lowLimit', 'limitEnable'
])

    analog_output = AnalogOutputFeedbackObject(
        objectName=row['Object_Name'],
        objectIdentifier=(AnalogOutputObject.objectType, row['Object_Identifier']),
        presentValue=row['Present_Value'],
        statusFlags=calculate_status_flags(row),
        eventState=row.get('Event_State', 0),  # Assuming default value is 0
        outOfService=bool(row['Out_Of_Service']),  # Convert to bool
        units=row['BACnet_Engineering_Units'],
        minPresValue=row['Min_Pres_Value'],
        maxPresValue=row['Max_Pres_Value'],
        resolution=row.get('Resolution', None),
        covIncrement=row.get('COV_Increment', 1),
        description=row['Description'],
        highLimit=row.get('High_Limit', None),
        lowLimit=row.get('Low_Limit', None),
        limitEnable=row.get('Limit_Enable', [False, False]),
        propertyList=property_list,
    )

    # Initialize the priority array with default values
    priority_array = PriorityArray()

    # Set the Present_Value at the specified priority position if within range
    if 'Priority' in row and 1 <= row['Priority'] <= 16:
        priority_index = row['Priority']   # Don't need to Adjust for 0-based index
        priority_array[priority_index] = PriorityValue(real=Real(row['Present_Value']))

    # Assign the priority array to the object
    analog_output.priorityArray = priority_array

    # Print out the priority array
    logging.info(f"Priority Array Contents: Current Priority is {row['Priority']} with a value of {row['Present_Value']}")
    # for i, priority_value in enumerate(priority_array):
    #     # Log the actual value contained in the PriorityValue object
    #     if priority_value:
    #         logging.info(f"Priority {i+1}: {priority_value.real}")
    #     else:
    #         logging.info(f"Priority {i+1}: None")

    logging.debug(f"Created AnalogOutputObject: {analog_output}")

    return analog_output

except Exception as e:
    logging.error(f"{color.RED}An error occurred in create_analog_output: {e}{color.END}")
    return None`

[kheldaroz402]

that's on my todo list - how mature is bacpypes3 ?

[JoelBender]

It's more mature, for client-like applications value = await app.read_property(...) is vastly simpler to use, and for server applications the ability to customize implementations with @property def presentValue(self) is much easier. The most important change is that asyncore is no longer supported in Python 3.12.