<commit_before>#!/usr/bin/env python

"""
This application presents a 'console' prompt to the user asking for
subscribe commands which create SubscribeCOVRequests.  The other commands are
for changing the type of reply to the confirmed COV notification that gets
sent.
"""

from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.consolecmd import ConsoleCmd

from bacpypes.core import run, enable_sleeping

from bacpypes.pdu import Address
from bacpypes.object import get_object_class
from bacpypes.apdu import SubscribeCOVRequest, \
    SimpleAckPDU, RejectPDU, AbortPDU

from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# globals
this_application = None

# how the application should respond
rsvp = (True, None, None)

#
#   SubscribeCOVApplication
#

@bacpypes_debugging
class SubscribeCOVApplication(BIPSimpleApplication):

    def __init__(self, *args):
        if _debug: SubscribeCOVApplication._debug("__init__ %r", args)
        BIPSimpleApplication.__init__(self, *args)

    def do_ConfirmedCOVNotificationRequest(self, apdu):
        if _debug: SubscribeCOVApplication._debug("do_ConfirmedCOVNotificationRequest %r", apdu)
        global rsvp

        if rsvp[0]:
            # success
            response = SimpleAckPDU(context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - simple_ack: %r", response)

        elif rsvp[1]:
            # reject
            response = RejectPDU(reason=rsvp[1], context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - reject: %r", response)

        elif rsvp[2]:
            # abort
            response = AbortPDU(reason=rsvp[2], context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - abort: %r", response)

        # return the result
        self.response(response)

    def do_UnconfirmedCOVNotificationRequest(self, apdu):
        if _debug: SubscribeCOVApplication._debug("do_UnconfirmedCOVNotificationRequest %r", apdu)


#
#   SubscribeCOVConsoleCmd
#

@bacpypes_debugging
class SubscribeCOVConsoleCmd(ConsoleCmd):

    def do_subscribe(self, args):
        """subscribe addr proc_id obj_type obj_inst [ confirmed ] [ lifetime ]

        Generate a SubscribeCOVRequest and wait for the response.
        """
        args = args.split()
        if _debug: SubscribeCOVConsoleCmd._debug("do_subscribe %r", args)

        try:
            addr, proc_id, obj_type, obj_inst = args[:4]

            proc_id = int(proc_id)
<commit_msg>put something out on the screen when a notification arrives<commit_after>#!/usr/bin/env python

"""
This application presents a 'console' prompt to the user asking for
subscribe commands which create SubscribeCOVRequests.  The other commands are
for changing the type of reply to the confirmed COV notification that gets
sent.
"""

from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.consolecmd import ConsoleCmd

from bacpypes.core import run, enable_sleeping

from bacpypes.pdu import Address
from bacpypes.object import get_object_class
from bacpypes.apdu import SubscribeCOVRequest, \
    SimpleAckPDU, RejectPDU, AbortPDU

from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# globals
this_application = None

# how the application should respond
rsvp = (True, None, None)

#
#   SubscribeCOVApplication
#

@bacpypes_debugging
class SubscribeCOVApplication(BIPSimpleApplication):

    def __init__(self, *args):
        if _debug: SubscribeCOVApplication._debug("__init__ %r", args)
        BIPSimpleApplication.__init__(self, *args)

    def do_ConfirmedCOVNotificationRequest(self, apdu):
        if _debug: SubscribeCOVApplication._debug("do_ConfirmedCOVNotificationRequest %r", apdu)
        global rsvp

        print("{} changed\n    {}".format(
            apdu.monitoredObjectIdentifier,
            ",\n    ".join("{} = {}".format(
                element.propertyIdentifier,
                str(element.value),
                ) for element in apdu.listOfValues),
            ))

        if rsvp[0]:
            # success
            response = SimpleAckPDU(context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - simple_ack: %r", response)

        elif rsvp[1]:
            # reject
            response = RejectPDU(reason=rsvp[1], context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - reject: %r", response)

        elif rsvp[2]:
            # abort
            response = AbortPDU(reason=rsvp[2], context=apdu)
            if _debug: SubscribeCOVApplication._debug("    - abort: %r", response)

        # return the result
        self.response(response)

    def do_UnconfirmedCOVNotificationRequest(self, apdu):
        if _debug: SubscribeCOVApplication._debug("do_UnconfirmedCOVNotificationRequest %r", apdu)

        print("{} changed\n    {}".format(
            apdu.monitoredObjectIdentifier,
            ",\n    ".join("{} is {}".format(
                element.propertyIdentifier,
                str(element.value),
                ) for element in apdu.listOfValues),
            ))

#
#   SubscribeCOVConsoleCmd
#

@bacpypes_debugging
class SubscribeCOVConsoleCmd(ConsoleCmd):

    def do_subscribe(self, args):
        """subscribe addr proc_id obj_type obj_inst [ confirmed ] [ lifetime ]

        Generate a SubscribeCOVRequest and wait for the response.
        """
        args = args.split()
        if _debug: SubscribeCOVConsoleCmd._debug("do_subscribe %r", args)

        try:
            addr, proc_id, obj_type, obj_inst = args[:4]

            proc_id = int(proc_id)

{'fraction_non_alphanumeric': 0.08418367346938775, 'fraction_numerical': 0.002380952380952381, 'mean_word_length': 3.1212333566923616, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '736262', 'n_tokens_mistral': 1755, 'n_tokens_neox': 1713, 'n_words': 442}