main.py

#!/usr/bin/env python3
# SPDX-License-Identifier: CC0-1.0
# Public-domain example code (CC0) - see LICENSE. The CAS BACnet Stack itself is
# a separate, commercially licensed product and is not covered by CC0.
# =============================================================================
# BACnet Profile Example - B-SS (BACnet Smart Sensor) - Python
#
# This example shows the *minimum* needed to implement the BACnet "B-SS"
# (Smart Sensor) device profile with the CAS BACnet Stack.
#
# "A B-SS is a simple sensing device with very limited resources." (ANSI/ASHRAE
# 135-2024, Annex L.4.5). Its Profile Function is the one Data Sharing line of
# that clause: "Ability to provide values for any of its BACnet objects upon
# request" - a client reads the sensor and gets what it is measuring NOW. Its
# only mandatory service is:
#
#     DS-RP-B  - respond to ReadProperty requests
#
# plus the ability to be discovered: it answers Who-Is with I-Am (handled
# automatically by the stack) and announces itself with an unsolicited I-Am on
# start-up.
#
# SIMULATION. This example has no real sensor, so section 2 below is a sensor
# Simulation that stands in for the physical world: the temperature ramps up
# and down, the binary input toggles and the multi-state input steps through
# its states, all on a timer. It is NOT part of the BACnet Surface - on a real
# device you replace sensor_simulation_tick() with your hardware sampling. The
# Simulation also has one Failure Mode: a broken sensor wire, which this device
# reports the BACnet way, as Reliability OPEN_LOOP and the FAULT bit of
# Status_Flags (cl. 12.2.7, 12.2.9). A Wire Test breaks the "wire" through a
# test-only hook (SENSOR_FAULT_FILE_VARIABLE below); nothing on the network can.
#
# A B-SS does NOT require WriteProperty, COV, alarms/events, scheduling, or
# trending - so this example deliberately does NOT enable any of those. That is
# the whole point of a profile example: implement exactly what the profile
# asks for, nothing more.
#
# This is the Python edition of BACnetProfileExample-B-SS-CPP - same objects,
# same names, same section layout, so you can diff the two side by side. The
# systematic difference is the binding mechanism: Python talks to the native
# CAS BACnet Stack library through `ctypes`, using the stack's own generated
# Python adapter IN PLACE from submodules/cas-bacnet-stack/adapters/python/
# (nothing of it is copied into this repository), which means:
#
#   - an EXPLICIT load step, like the C++ edition's LoadBACnetFunctions():
#     `ctypes.CDLL(<run directory or submodule bin>/<libname>)` then
#     `bind(library)`, both before any BACnetStack_* call (see "Load the CAS
#     BACnet Stack" below). Without the library the example prints why and
#     exits 1.
#   - every Get*Property callback is a plain Python function wrapped in a
#     ctypes.CFUNCTYPE, receiving ctypes POINTER out-parameters it indexes
#     like a list (`value[0] = 21.5`).
#   - THE STACK IS SINGLE-THREADED BY CONTRACT. Nothing in this file spawns a
#     thread that touches the stack; all networking is done through a
#     non-blocking socket (common/simple_udp.py) polled once per
#     BACnetStack_Tick() in the same loop - see "Run the stack" at the bottom.
#
# The device exposes three read-only sensor objects plus the Network Port
# object that every BACnet device must have. Each object is given a colour name
# (a convention shared across this example series):
#
#     Device 389001          "Chipkin Example B-SS"    (instance configurable with --deviceID,
#                                                       Object_Name with --deviceName)
#     Analog Input  1        "Bronze"     (REAL, degrees Celsius; ramps 18.0 .. 26.0)
#     Binary Input  1        "Emerald"    (active / inactive; toggles every 3 s)
#     Multi-State Input 1    "Hot Pink"   (an enumerated state, 1..3; steps every 4 s)
#     Network Port 1         "Vermilion"  (the BACnet/IP port - required)
#
# To be a conformant BACnet device (Protocol_Revision 30) each object must
# expose its full set of REQUIRED properties. Most are generated by the stack
# (Object_Identifier, Object_Type, Status_Flags, Object_List, Protocol_*).
# Event_State is NOT generated - with no alarming configured it simply reads its
# datatype default of normal(0), correct by coincidence, not because the stack
# computes it. The handful that the application must supply are served by the
# Get*Property callbacks below, and a few are turned on with SetPropertyEnabled.
#
# Interactive keys (handled by the shared helper): h = help, q = quit,
# up/down = nudge the simulated Analog Input 1 reading by +/-1.1. Command line:
# --port <n>, --deviceID <n>, --deviceName <text>.
#
# All the UDP/stack plumbing lives in common/ so this file can stay focused on
# the BACnet logic.
# =============================================================================

import ctypes
import os
import signal
import sys
import time

# Where this example finds its code. It is run as `python main.py`, so the
# script's own directory (the RUN DIRECTORY) goes first on sys.path (for
# `common`, and for the stack's adapter files if a licensed customer placed them
# there). A development checkout falls back to the CAS BACnet Stack's own Python
# adapter, used in place from the stack submodule (`CASBACnetStackAdapter` and
# the generated `CASBACnetStackAdapterBindings`). The stack itself is loaded in
# main() - see "Load the CAS BACnet Stack" - never at import time, so that
# --help / --version and the missing-library message work without it.
REPOSITORY = os.path.dirname(os.path.abspath(__file__))
CAS_BACNET_STACK = os.path.join(REPOSITORY, "submodules", "cas-bacnet-stack")
ADAPTER_DIRECTORIES = [REPOSITORY, os.path.join(CAS_BACNET_STACK, "adapters", "python")]
LIBRARY_DIRECTORIES = [REPOSITORY, os.path.join(CAS_BACNET_STACK, "bin")]
sys.path[:0] = [d for d in ADAPTER_DIRECTORIES if d not in sys.path]

# Show print() output immediately, even when stdout is redirected to a file or
# pipe (a background run, a Wire Test): Python fully buffers stdout when it is
# not a terminal - the C++ edition's setvbuf(stdout, NULL, _IONBF, 0) does the
# same job.
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(line_buffering=True)

from common.simple_udp import SimpleUDP  # noqa: E402
from common import cas_example_helper as helper  # noqa: E402
from common import cas_bacnet_stack_example_constants as c  # noqa: E402

# -----------------------------------------------------------------------------
# 1. Example + device configuration
# -----------------------------------------------------------------------------
APP_NAME = "BACnet B-SS (Smart Sensor) Example - Python"
APP_VERSION = "1.2.0"

# The device instance. BACnet requires this to be configurable, so it defaults
# to 389001 and can be overridden on the command line with --deviceID. Keep it
# configurable in your product: it must be unique across the internetwork.
g_device_instance = 389001

# ---- Device identity: CHANGE ALL OF THIS BEFORE YOU SHIP --------------------
# Everything in this block is read by clients and shown to the operator in
# every discovery tool on the network. Left as-is, your product will appear on
# a real site announcing itself as a Chipkin demo. None of it is cosmetic:
# Object_Name must be unique across the BACnet internetwork, and Model_Name /
# Vendor_Identifier are what a building operator uses to identify your device.
#
# This block is the ship checklist. Every constant below has a note saying
# what to change it to; nothing here is safe to leave at its example value.
# -----------------------------------------------------------------------------

# Your BACnet Vendor Identifier. 389 = Chipkin Automation Systems; change this
# to YOUR company's vendor ID before shipping a product. Vendor IDs are
# assigned by ASHRAE - request one (free) at
# https://bacnet.org/assigned-vendor-ids/. Update VENDOR_NAME below to match.
VENDOR_IDENTIFIER = 389

# The Device object's Object_Name - the DEFAULT one.
#
# THIS IS THE ONE THAT WILL BITE YOU. Object_Name must be unique across the
# whole BACnet internetwork (cl. 12.1.1.4), so it must be per-unit
# configurable, exactly like the instance. This example takes it at start-up
# with --deviceName <text> (common/ parse_device_name_arg(), read into
# g_device_name in main()); DEVICE_NAME is only what a unit announces when
# nobody configured it. Ship two units without --deviceName and BOTH announce
# the same Object_Name - a spec violation, and a hard BTL failure. In your
# product, configure the name per unit (a serial number, DIP switches, a
# config file, or this argument) and change the default below to something
# that says "not configured yet".
DEVICE_NAME = "Chipkin Example B-SS"
g_device_name = DEVICE_NAME  # --deviceName, or the default

# The Device object's Description. Change it to what YOUR device actually is;
# this string describes this tutorial.
DEVICE_DESCRIPTION = (
    "Chipkin CAS BACnet Stack example - B-SS (Smart Sensor) profile. "
    "Demonstrates DS-RP-B: ReadProperty plus Who-Is/I-Am with read-only sensor objects."
)

# Device identity strings (read by clients, and used to populate I-Am).
#   VENDOR_NAME - your company name; it must match VENDOR_IDENTIFIER above.
#   MODEL_NAME  - your model designation. This is what a building operator
#                 reads to identify your device in a discovery tool.
# Application_Software_Version (12) is just APP_VERSION - one source of truth,
# so it can never drift from what --version / the start-up banner prints.
#
# Firmware_Revision (44) names the underlying platform/stack, not this
# example's own version - built at run time from the CAS BACnet Stack's own
# BACnetStack_GetAPIMajorVersion()/etc. (the same 4 calls
# common/cas_example_helper.py's print_version() uses for the start-up
# banner's "CAS BACnet Stack version: X.Y.Z.W" line), so it can never go stale
# either - see g_firmware_revision below, set once right after the stack
# loads.
VENDOR_NAME = "Chipkin Automation Systems"
MODEL_NAME = "CAS BACnet Stack Example - B-SS"
g_firmware_revision = ""

# The sensor objects (all instance 1) and their colour names.
ANALOG_INPUT_INSTANCE = 1       # "Bronze"
BINARY_INPUT_INSTANCE = 1       # "Emerald"
MULTI_STATE_INPUT_INSTANCE = 1  # "Hot Pink"
MULTI_STATE_INPUT_NUMBER_OF_STATES = 3

# The Network Port object - every BACnet device must have one. It represents
# the BACnet/IP port this device communicates on.
NETWORK_PORT_INSTANCE = 1  # "Vermilion"
MAX_APDU_LENGTH = 1476     # BACnet/IP APDU length

# BACnet/IP addressing the Network Port reports. The IP address and subnet
# mask are filled in at start-up from the host's primary interface; the
# gateway is left unset (0.0.0.0) for this example. The stack also uses
# IP_Address + BACnet_IP_UDP_Port to build the port's MAC_Address
# automatically.
g_ip_address = [0, 0, 0, 0]
g_ip_subnet_mask = [0, 0, 0, 0]
g_ip_default_gateway = [0, 0, 0, 0]
g_bacnet_ip_udp_port = 47808

# The sensor objects' live values. The callbacks in section 3 only ever hand
# these back; section 2 (the Simulation) is the only thing that changes them.
# On a real device your hardware sampling updates them instead.
g_analog_input_1_value = 21.5     # degrees Celsius
g_binary_input_1_value = 0        # 0 = inactive, 1 = active
g_multi_state_input_1_value = 1   # 1..MULTI_STATE_INPUT_NUMBER_OF_STATES

# Analog Input 1's Reliability (cl. 12.2.9). NO_FAULT_DETECTED while the
# sensor is healthy; OPEN_LOOP while its (simulated) wire is broken. The stack
# builds the FAULT bit of Status_Flags from this property by itself
# (cl. 12.2.7: FAULT is "Logical TRUE (1) if the Reliability property is
# present and does not have a value of NO_FAULT_DETECTED") - the example only
# has to serve it.
PROPERTY_IDENTIFIER_RELIABILITY = 103
RELIABILITY_NO_FAULT_DETECTED = 0
RELIABILITY_OPEN_LOOP = 4
g_analog_input_1_reliability = RELIABILITY_NO_FAULT_DETECTED

udp = SimpleUDP()
bacnet = None  # the stack's ctypes bindings, once load_cas_bacnet_stack() has bound the library

# -----------------------------------------------------------------------------
# 2. The sensor Simulation (NOT part of the BACnet Surface)
#
# Stands in for the physical world. Once a second:
#   - Analog Input 1 moves 0.5 C, ramping up to 26.0 and back down to 18.0
#   - Binary Input 1 toggles every 3 s
#   - Multi-State Input 1 steps 1 -> 2 -> 3 -> 1 every 4 s
# and the sensor wire is checked.
#
# The broken wire is a TEST-ONLY hook, not a feature: when the environment
# variable named by SENSOR_FAULT_FILE_VARIABLE names a file, the wire counts as
# broken for as long as that file exists. A Wire Test creates and deletes the
# file to break and repair the sensor. While the wire is broken the reading
# freezes at its last good value (a dead wire measures nothing new) and
# Reliability reads OPEN_LOOP; once the file is gone the reading resumes and
# Reliability returns to NO_FAULT_DETECTED. On a real device, set Reliability
# from your hardware's own diagnostics (an ADC reading on the rails, a sensor
# that stopped answering, ...).
# -----------------------------------------------------------------------------
SENSOR_FAULT_FILE_VARIABLE = "CAS_EXAMPLE_TEST_SENSOR_FAULT_FILE"
g_sensor_fault_file = ""   # read once at start-up; empty = no hook
ANALOG_INPUT_LOWEST = 18.0
ANALOG_INPUT_HIGHEST = 26.0
ANALOG_INPUT_STEP = 0.5
BINARY_INPUT_TOGGLE_SECONDS = 3
MULTI_STATE_INPUT_STEP_SECONDS = 4

# The Simulation's own state (the C++ edition's function-local statics).
_simulation_started = None
_simulation_last_second = 0
_analog_input_direction = 1.0


def sensor_wire_broken():
    if not g_sensor_fault_file:
        return False
    try:
        with open(g_sensor_fault_file, "rb"):
            return True
    except OSError:
        return False


def sensor_simulation_tick():
    global _simulation_started, _simulation_last_second, _analog_input_direction
    global g_analog_input_1_value, g_binary_input_1_value, g_multi_state_input_1_value
    global g_analog_input_1_reliability
    if _simulation_started is None:
        _simulation_started = time.monotonic()
    seconds_running = int(time.monotonic() - _simulation_started)
    if seconds_running == _simulation_last_second:
        return
    _simulation_last_second = seconds_running

    reliability = RELIABILITY_OPEN_LOOP if sensor_wire_broken() else RELIABILITY_NO_FAULT_DETECTED
    if reliability != g_analog_input_1_reliability:
        g_analog_input_1_reliability = reliability
        broken = reliability == RELIABILITY_OPEN_LOOP
        print(f"Simulation: Analog Input 1 (Bronze) sensor wire {'broken' if broken else 'repaired'} - "
              f"Reliability {'OPEN_LOOP' if broken else 'NO_FAULT_DETECTED'}")
    if g_analog_input_1_reliability == RELIABILITY_NO_FAULT_DETECTED:
        if g_analog_input_1_value >= ANALOG_INPUT_HIGHEST:
            _analog_input_direction = -1.0
        elif g_analog_input_1_value <= ANALOG_INPUT_LOWEST:
            _analog_input_direction = 1.0
        g_analog_input_1_value += _analog_input_direction * ANALOG_INPUT_STEP
    if seconds_running % BINARY_INPUT_TOGGLE_SECONDS == 0:
        g_binary_input_1_value = 0 if g_binary_input_1_value else 1
    if seconds_running % MULTI_STATE_INPUT_STEP_SECONDS == 0:
        g_multi_state_input_1_value = g_multi_state_input_1_value % MULTI_STATE_INPUT_NUMBER_OF_STATES + 1


# -----------------------------------------------------------------------------
# 3. Property "get" callbacks
#
# The stack calls these when a client reads a property. For each data type the
# stack uses a separate callback. We return True (and fill the out-param) when
# we recognise the (object, property) pair, and False otherwise.
#
# THE errorCode OUT-PARAMETER. Every Get callback ends with a trailing
# error_code pointer. The stack PRESETS it to success (84) before the call, and
# reads it only if you return False. That gives a declining callback two
# distinct meanings:
#
#   1. return False and LEAVE error_code ALONE -> "I have no opinion on this
#      property." The stack falls back to its own handling (see below).
#   2. return False and SET error_code[0]      -> "This read fails, with THIS
#      BACnet error." The client gets exactly that Error-PDU.
#
# Option 2 is new (CAS BACnet Stack issue #974); before it, a Get callback had
# no way to name an error at all. Do not reach for it reflexively - option 1 is
# still the right answer most of the time, for the reason in the next
# paragraph.
#
# WHAT False-WITHOUT-AN-ERROR-CODE ACTUALLY DOES - the most important
# paragraph in this file, and the opposite of what most people assume. It
# does NOT reliably produce a BACnet error. The stack errors only for the
# handful of properties it refuses to invent: Present_Value,
# Number_Of_States, Relinquish_Default, Local_Date, Local_Time, and a Network
# Port's APDU_Length (declining one of those reads back as Error:
# read-access-denied). For EVERYTHING ELSE, a False return means the stack
# SILENTLY SUBSTITUTES a default:
#     Object_Name -> the literal string "undefined"
#     Units       -> no-units (95)
#     otherwise   -> a datatype zero-value
#
# AND THAT FALLBACK IS LORE-BEARING, WHICH IS WHY WE DO NOT "FIX" IT HERE. It
# is tempting to end every callback with error_code[0] = unknown-property so
# nothing is ever silently invented. That breaks the device. The stack relies
# on the decline-and-fabricate path to answer required properties the
# application is not expected to serve - the Device's
# Max_APDU_Length_Accepted, APDU_Timeout and Number_Of_APDU_Retries among
# them. Name an error on the catch-all return and those required properties
# start failing instead of answering. So: set error_code[0] ONLY where THIS
# device knows the read is wrong. There is exactly one such case below
# (State_Text with an out-of-range array index); the catch-all `return False`
# at the end of each callback deliberately leaves error_code alone.
#
# ADDING AN OBJECT? READ THIS FIRST.
# The consequence is the opposite of reassuring. These callbacks are not
# uniformly strict:
#   - get_property_real / get_property_enumerated /
#     get_property_unsigned_integer match on object type AND INSTANCE (e.g.
#     object_instance == ANALOG_INPUT_INSTANCE). A new instance falls through
#     every one of those checks.
#   - get_property_bool matches on TYPE ONLY, so a new instance gets
#     Out_Of_Service for free.
# So a half-added object does NOT fail loudly. Its Present_Value errors (that
# one is in the list above) - but its Object_Name reads back as "undefined"
# and its Units as no-units, with no error at all. Add two objects that way
# and BOTH report Object_Name "undefined": duplicate object names within one
# device, which is a spec violation and a hard BTL failure, and which every
# scan tool will show you as a healthy object. The device looks fine and is
# non-conformant.
#
# So: when you add an instance, walk EVERY callback below, then read back
# every required property of the new object and DIFF IT against the existing
# one. Do not trust "it scanned OK" - that is exactly the failure mode. See
# TUTORIAL.md "Add a second analog input" for the edits.
# -----------------------------------------------------------------------------


# REAL (floating point) - the Analog Input's Present_Value.
def get_property_real(device_instance, object_type, object_instance, property_identifier,
                      value, use_array_index, property_array_index, error_code):
    # error_code: see "THE errorCode OUT-PARAMETER" above - we decline without naming an error.
    if (device_instance == g_device_instance and
            object_type == c.OBJECT_TYPE_ANALOG_INPUT and
            object_instance == ANALOG_INPUT_INSTANCE and
            property_identifier == c.PROPERTY_IDENTIFIER_PRESENT_VALUE):
        # ON REAL HARDWARE: return the live sensor reading here. Read it from a
        # cached variable that your hardware updates (as g_analog_input_1_value
        # is), NOT directly from a slow/blocking device (I2C, SPI, ADC
        # conversion): this callback runs inside BACnetStack_Tick(), so
        # blocking it delays all BACnet processing. Sample the sensor on a
        # timer and just hand back the latest value from here.
        value[0] = g_analog_input_1_value
        return True
    return False


# ENUMERATED - the Binary Input's Present_Value (0 = inactive, 1 = active),
# the Analog Input's Units and Reliability, and the Network Port's
# BACnet_IP_Mode.
def get_property_enumerated(device_instance, object_type, object_instance, property_identifier,
                            value, use_array_index, property_array_index, error_code):
    if device_instance != g_device_instance:
        return False
    if object_type == c.OBJECT_TYPE_BINARY_INPUT and object_instance == BINARY_INPUT_INSTANCE:
        if property_identifier == c.PROPERTY_IDENTIFIER_PRESENT_VALUE:
            # ON REAL HARDWARE: return your cached input state here - the same
            # rule as get_property_real above applies (never block this
            # callback on slow I/O; sample on a timer and hand back the latest).
            value[0] = g_binary_input_1_value
            return True
        if property_identifier == c.PROPERTY_IDENTIFIER_POLARITY:
            value[0] = c.POLARITY_NORMAL  # required property of a Binary Input
            return True
    if object_type == c.OBJECT_TYPE_ANALOG_INPUT and object_instance == ANALOG_INPUT_INSTANCE:
        if property_identifier == c.PROPERTY_IDENTIFIER_UNITS:
            value[0] = c.ENGINEERING_UNITS_DEGREES_CELSIUS
            return True
        # Reliability (enabled in main()). The stack also reads it through this
        # callback to build Status_Flags, so once enabled it must always answer:
        # if it declines, the stack silently substitutes 0 (no-fault-detected),
        # and the sensor reports itself healthy forever.
        if property_identifier == PROPERTY_IDENTIFIER_RELIABILITY:
            value[0] = g_analog_input_1_reliability
            return True
    if (object_type == c.OBJECT_TYPE_NETWORK_PORT and object_instance == NETWORK_PORT_INSTANCE and
            property_identifier == c.PROPERTY_IDENTIFIER_BACNET_IP_MODE):
        value[0] = c.BACNET_IP_MODE_NORMAL  # not foreign-device, not BBMD
        return True
    return False


# UNSIGNED INTEGER - the Multi-State Input's Present_Value, and the Device's
# Vendor_Identifier (the stack also uses Vendor_Identifier to build I-Am).
def get_property_unsigned_integer(device_instance, object_type, object_instance, property_identifier,
                                  value, use_array_index, property_array_index, error_code):
    if device_instance != g_device_instance:
        return False
    if object_type == c.OBJECT_TYPE_MULTI_STATE_INPUT and object_instance == MULTI_STATE_INPUT_INSTANCE:
        if property_identifier == c.PROPERTY_IDENTIFIER_PRESENT_VALUE:
            value[0] = g_multi_state_input_1_value  # valid range is 1..Number_Of_States
            return True
        if property_identifier == c.PROPERTY_IDENTIFIER_NUMBER_OF_STATES:
            value[0] = MULTI_STATE_INPUT_NUMBER_OF_STATES  # required property
            return True
        # State_Text is an array. The stack asks for its LENGTH here (array
        # index 0) before reading each element via get_property_character_string.
        if (property_identifier == c.PROPERTY_IDENTIFIER_STATE_TEXT and
                use_array_index and property_array_index == 0):
            value[0] = MULTI_STATE_INPUT_NUMBER_OF_STATES
            return True
    if (object_type == c.OBJECT_TYPE_DEVICE and object_instance == g_device_instance and
            property_identifier == c.PROPERTY_IDENTIFIER_VENDOR_IDENTIFIER):
        value[0] = VENDOR_IDENTIFIER
        return True
    if object_type == c.OBJECT_TYPE_NETWORK_PORT and object_instance == NETWORK_PORT_INSTANCE:
        if property_identifier == c.PROPERTY_IDENTIFIER_APDU_LENGTH:
            value[0] = MAX_APDU_LENGTH
            return True
        if property_identifier == c.PROPERTY_IDENTIFIER_REFERENCE_PORT:
            value[0] = c.NETWORK_PORT_REFERENCE_PORT_NONE
            return True
        if property_identifier == c.PROPERTY_IDENTIFIER_BACNET_IP_UDP_PORT:
            value[0] = g_bacnet_ip_udp_port
            return True
    return False


# BOOLEAN - Out_Of_Service is a required property of every input object and of
# the Network Port. This is a read-only sensor, so nothing is ever out of
# service: always False.
def get_property_bool(device_instance, object_type, object_instance, property_identifier,
                      value, use_array_index, property_array_index, error_code):
    if device_instance != g_device_instance:
        return False
    if property_identifier == c.PROPERTY_IDENTIFIER_OUT_OF_SERVICE and object_type in (
            c.OBJECT_TYPE_ANALOG_INPUT, c.OBJECT_TYPE_BINARY_INPUT,
            c.OBJECT_TYPE_MULTI_STATE_INPUT, c.OBJECT_TYPE_NETWORK_PORT):
        value[0] = False
        return True
    return False


# OCTET STRING - the Network Port's BACnet/IP addressing. The stack cannot
# know the host's IP, so the application must supply IP_Address and
# IP_Subnet_Mask (and IP_Default_Gateway). Each is four octets. The stack also
# reads IP_Address (with BACnet_IP_UDP_Port) to build the port's six-octet
# MAC_Address.
def get_property_octet_string(device_instance, object_type, object_instance, property_identifier,
                              value, value_element_count, max_element_count,
                              use_array_index, property_array_index, error_code):
    if (device_instance != g_device_instance or
            object_type != c.OBJECT_TYPE_NETWORK_PORT or
            object_instance != NETWORK_PORT_INSTANCE or
            max_element_count < 4):
        return False
    if property_identifier == c.PROPERTY_IDENTIFIER_IP_ADDRESS:
        source = g_ip_address
    elif property_identifier == c.PROPERTY_IDENTIFIER_IP_SUBNET_MASK:
        source = g_ip_subnet_mask
    elif property_identifier == c.PROPERTY_IDENTIFIER_IP_DEFAULT_GATEWAY:
        source = g_ip_default_gateway
    else:
        return False
    for i in range(4):
        value[i] = source[i]
    value_element_count[0] = 4
    return True


# Small helper: copy a Python str into the stack's character-string buffer as
# UTF-8 and set the element count + encoding. Returns True (so callers can
# `return`).
def return_character_string(text, value, value_element_count, max_element_count, encoding_type):
    encoded = text.encode("utf-8")
    if len(encoded) > max_element_count:
        # Truncate SILENTLY to fit the stack's buffer. max_element_count is
        # MAX_CHARACTER_STRING_SIZE (256 in this build), and our longest
        # strings fit: DEVICE_DESCRIPTION with room to spare, and a
        # --deviceName, which parse_device_name_arg() caps at 255 bytes - so
        # this never trips here. But a stack built with STACK_OPTION_TARGET_EMBEDDED drops
        # that limit to 64, and a long Object_Name or Description would be
        # clipped mid-word with nothing on the wire or console to tell you. If
        # you lengthen any served string, check it against
        # MAX_CHARACTER_STRING_SIZE for your target, or make this truncation
        # loud.
        encoded = encoded[:max_element_count]
    ctypes.memmove(value, encoded, len(encoded))
    value_element_count[0] = len(encoded)
    encoding_type[0] = c.CHARACTER_STRING_ENCODING_UTF8
    return True


# CHARACTER STRING - Object_Name for each object, and the device Description.
def get_property_character_string(device_instance, object_type, object_instance, property_identifier,
                                  value, value_element_count, max_element_count, encoding_type,
                                  use_array_index, property_array_index, error_code):
    if device_instance != g_device_instance:
        return False

    # State_Text (optional) - one label per state of the Multi-State Input. It
    # is a BACnet array, so the stack asks for one element at a time by index
    # (1..Number_Of_States). Present_Value 1 -> "On", 2 -> "Off", 3 -> "Auto".
    if (object_type == c.OBJECT_TYPE_MULTI_STATE_INPUT and
            object_instance == MULTI_STATE_INPUT_INSTANCE and
            property_identifier == c.PROPERTY_IDENTIFIER_STATE_TEXT and use_array_index):
        state_text = ("On", "Off", "Auto")
        if 1 <= property_array_index <= MULTI_STATE_INPUT_NUMBER_OF_STATES:
            return return_character_string(state_text[property_array_index - 1], value,
                                           value_element_count, max_element_count, encoding_type)
        # The one place in this file where naming an error is clearly right:
        # the client asked for State_Text[n] and this object has no element n.
        # That is not "no opinion" - it is a wrong read, and the spec has a
        # code for it. (At this Stack Pin the stack also checks n against the
        # length get_property_unsigned_integer serves for index 0 and answers
        # invalid-array-index before it gets here; this line is what still
        # answers correctly if that length and this table ever disagree -
        # without it the client would silently receive an empty string.)
        error_code[0] = c.ERROR_CODE_INVALID_ARRAY_INDEX
        return False

    # Object_Name - the colour name for each object.
    if property_identifier == c.PROPERTY_IDENTIFIER_OBJECT_NAME:
        if object_type == c.OBJECT_TYPE_DEVICE and object_instance == g_device_instance:
            return return_character_string(g_device_name, value, value_element_count, max_element_count,
                                           encoding_type)
        if object_type == c.OBJECT_TYPE_ANALOG_INPUT and object_instance == ANALOG_INPUT_INSTANCE:
            return return_character_string("Bronze", value, value_element_count, max_element_count,
                                           encoding_type)
        if object_type == c.OBJECT_TYPE_BINARY_INPUT and object_instance == BINARY_INPUT_INSTANCE:
            return return_character_string("Emerald", value, value_element_count, max_element_count,
                                           encoding_type)
        if object_type == c.OBJECT_TYPE_MULTI_STATE_INPUT and object_instance == MULTI_STATE_INPUT_INSTANCE:
            return return_character_string("Hot Pink", value, value_element_count, max_element_count,
                                           encoding_type)
        if object_type == c.OBJECT_TYPE_NETWORK_PORT and object_instance == NETWORK_PORT_INSTANCE:
            return return_character_string("Vermilion", value, value_element_count, max_element_count,
                                           encoding_type)

    # The remaining strings are all on the Device object - its identity, read
    # by clients and used to populate the device's I-Am / object list.
    if object_type == c.OBJECT_TYPE_DEVICE and object_instance == g_device_instance:
        device_strings = {
            c.PROPERTY_IDENTIFIER_DESCRIPTION: DEVICE_DESCRIPTION,
            c.PROPERTY_IDENTIFIER_VENDOR_NAME: VENDOR_NAME,
            c.PROPERTY_IDENTIFIER_MODEL_NAME: MODEL_NAME,
            c.PROPERTY_IDENTIFIER_FIRMWARE_REVISION: g_firmware_revision,
            c.PROPERTY_IDENTIFIER_APPLICATION_SOFTWARE_VERSION: APP_VERSION,
        }
        if property_identifier in device_strings:
            return return_character_string(device_strings[property_identifier], value,
                                           value_element_count, max_element_count, encoding_type)

    return False


# -----------------------------------------------------------------------------
# 4. main()
# -----------------------------------------------------------------------------

# The stack library's file name when the adapter (which names it) is not there
# to ask - only used to fill in the missing-library message.
FALLBACK_LIBRARY_NAME = "CASBACnetStack_x64_Release.dll" if os.name == "nt" else "libCASBACnetStack_x64_Release.so"


def load_cas_bacnet_stack():
    """Find, load and bind the CAS BACnet Stack. Returns None on success, or
    (missing file, [folders searched]) when it cannot: the adapter's two files
    or the library are in neither the run directory nor the development
    checkout's submodule, or ctypes cannot load the library it found."""
    global bacnet
    try:
        from CASBACnetStackAdapter import libname
        import CASBACnetStackAdapterBindings
    except ImportError:
        # No adapter either (a release has neither): name the library first -
        # it is the part a customer licenses and builds.
        if not any(os.path.isfile(os.path.join(d, FALLBACK_LIBRARY_NAME)) for d in LIBRARY_DIRECTORIES):
            return FALLBACK_LIBRARY_NAME, LIBRARY_DIRECTORIES
        return "CASBACnetStackAdapterBindings.py", ADAPTER_DIRECTORIES
    if libname is None:
        return FALLBACK_LIBRARY_NAME, LIBRARY_DIRECTORIES
    for directory in LIBRARY_DIRECTORIES:
        library_path = os.path.join(directory, libname)
        if os.path.isfile(library_path):
            try:
                CASBACnetStackAdapterBindings.bind(ctypes.CDLL(library_path))
            except (OSError, AttributeError):
                return libname, LIBRARY_DIRECTORIES
            bacnet = CASBACnetStackAdapterBindings
            return None
    return libname, LIBRARY_DIRECTORIES


def main(argv):
    global g_device_instance, g_device_name, g_bacnet_ip_udp_port, g_analog_input_1_value
    global g_firmware_revision, g_sensor_fault_file

    # --- Load the CAS BACnet Stack -------------------------------------------
    # Required before any other BACnetStack_* call, like the C++ edition's
    # LoadBACnetFunctions(). The library is looked for in the run directory
    # (the folder that holds main.py), then in a development checkout's
    # submodules/cas-bacnet-stack/bin/, where building the stack puts it
    # (README.md "Build the CAS BACnet Stack library"). A release of this
    # example does not include it: it is a separately licensed product.
    missing = load_cas_bacnet_stack()

    # --- Command line + version --------------------------------------------
    # --help / --version print and exit, so handle them before we bind a socket.
    # They work without the stack (the banner then says it is not loaded).
    # show_device_name_cli_option: --help lists --deviceName, which this
    # example reads.
    if helper.handle_help_and_version_args(argv, APP_NAME, APP_VERSION, show_device_name_cli_option=True):
        return 0
    if missing is not None:
        helper.print_missing_stack_library(*missing)
        return 1

    # g_firmware_revision (Device object property 44) - see its own comment
    # above for why this is the STACK's version, not this example's own (that
    # is Application_Software_Version / APP_VERSION instead). Must happen after
    # the stack loads (these getters are some of the functions it binds) and
    # before the Device object is ever readable.
    g_firmware_revision = "{}.{}.{}.{}".format(
        bacnet.BACnetStack_GetAPIMajorVersion(), bacnet.BACnetStack_GetAPIMinorVersion(),
        bacnet.BACnetStack_GetAPIPatchVersion(), bacnet.BACnetStack_GetAPIBuildVersion())

    port = helper.parse_port_arg(argv, 47808)
    g_device_instance = helper.parse_device_id_arg(argv, g_device_instance)
    g_device_name = helper.parse_device_name_arg(argv, DEVICE_NAME)
    helper.print_version(APP_NAME, APP_VERSION)

    # --- Bind the BACnet/IP socket -----------------------------------------
    if not helper.setup_udp(udp, port, NETWORK_PORT_INSTANCE):
        return 1

    # Capture the BACnet/IP addressing the Network Port object will report.
    g_bacnet_ip_udp_port = port
    local = helper.get_local_ipv4()
    g_ip_address[:] = [int(part) for part in local.address.split(".")]
    g_ip_subnet_mask[:] = [int(part) for part in local.netmask.split(".")]

    # --- Register callbacks -------------------------------------------------
    # The transport + time callbacks are shared boilerplate, keyed to the
    # Network Port instance that owns the socket just bound (the stack
    # identifies a link by its Network Port INSTANCE, so the transport
    # callbacks and the start-up I-Am name the one added below).
    helper.register_common_callbacks(udp, NETWORK_PORT_INSTANCE)
    # The property callbacks are specific to this example. Wrap each Python
    # function in the matching ctypes.CFUNCTYPE and keep the wrapper object
    # alive for the life of the process - see cas_example_helper.py's
    # "CALLBACKS MUST BE ROOTED" note; the same rule applies here.
    property_callbacks = [
        (bacnet.BACnetStack_RegisterCallbackGetPropertyReal,
         bacnet.FPCallbackGetPropertyReal(get_property_real)),
        (bacnet.BACnetStack_RegisterCallbackGetPropertyEnumerated,
         bacnet.FPCallbackGetPropertyEnumerated(get_property_enumerated)),
        (bacnet.BACnetStack_RegisterCallbackGetPropertyUnsignedInteger,
         bacnet.FPCallbackGetPropertyUnsignedInteger(get_property_unsigned_integer)),
        (bacnet.BACnetStack_RegisterCallbackGetPropertyCharacterString,
         bacnet.FPCallbackGetPropertyCharacterString(get_property_character_string)),
        (bacnet.BACnetStack_RegisterCallbackGetPropertyBool,
         bacnet.FPCallbackGetPropertyBool(get_property_bool)),
        (bacnet.BACnetStack_RegisterCallbackGetPropertyOctetString,
         bacnet.FPCallbackGetPropertyOctetString(get_property_octet_string)),
    ]
    for register, callback in property_callbacks:
        helper._callback_refs.append(callback)  # rooted for the life of the process
        register(callback)

    # --- Create the device --------------------------------------------------
    if not bacnet.BACnetStack_AddDevice(g_device_instance):
        print(f"Error: Failed to add the Device {g_device_instance}.")
        return 1

    # Enable ReadProperty (DS-RP-B) - the one mandatory service for B-SS.
    # We set it explicitly to make the profile requirement obvious.
    #
    # We deliberately do NOT enable WriteProperty, ReadPropertyMultiple,
    # SubscribeCOV, or any alarm/event service: a Smart Sensor does not
    # require them, so a faithful B-SS example leaves them off.
    if not bacnet.BACnetStack_SetServiceEnabled(g_device_instance, c.SERVICE_READ_PROPERTY, True):
        print("Error: Failed to enable the ReadProperty service.")
        return 1

    # Discovery: Who-Is/I-Am (DM-DDB-B) and Who-Has/I-Have (DM-DOB-B).
    #
    # These need enabling even though the device already ANSWERS them. The
    # stack's service defaults are whoIs + whoHas + readProperty only - iAm
    # and iHave are left False. Who-Is is answered and the start-up I-Am is
    # sent regardless, because neither is gated on the bit; but
    # Protocol_Services_Supported is emitted verbatim from that bitstring, so
    # without these calls the device DOES I-Am and I-Have while telling every
    # client it supports neither. The README claims DM-DDB-B and DM-DOB-B;
    # this is what makes the claim true on the wire.
    if not (bacnet.BACnetStack_SetServiceEnabled(g_device_instance, c.SERVICE_WHO_IS, True) and
            bacnet.BACnetStack_SetServiceEnabled(g_device_instance, c.SERVICE_I_AM, True) and
            bacnet.BACnetStack_SetServiceEnabled(g_device_instance, c.SERVICE_WHO_HAS, True) and
            bacnet.BACnetStack_SetServiceEnabled(g_device_instance, c.SERVICE_I_HAVE, True)):
        print("Error: Failed to enable the discovery services (Who-Is/I-Am, Who-Has/I-Have).")
        return 1

    # --- Add the read-only sensor objects -----------------------------------
    # Every stack setup call returns a bool; a real device should always check
    # it, so this example does too.
    if not bacnet.BACnetStack_AddObject(g_device_instance, c.OBJECT_TYPE_ANALOG_INPUT, ANALOG_INPUT_INSTANCE):
        print(f"Error: Failed to add Analog Input {ANALOG_INPUT_INSTANCE} (Bronze).")
        return 1
    if not bacnet.BACnetStack_AddObject(g_device_instance, c.OBJECT_TYPE_BINARY_INPUT, BINARY_INPUT_INSTANCE):
        print(f"Error: Failed to add Binary Input {BINARY_INPUT_INSTANCE} (Emerald).")
        return 1
    if not bacnet.BACnetStack_AddObject(g_device_instance, c.OBJECT_TYPE_MULTI_STATE_INPUT,
                                        MULTI_STATE_INPUT_INSTANCE):
        print(f"Error: Failed to add Multi-State Input {MULTI_STATE_INPUT_INSTANCE} (Hot Pink).")
        return 1

    # --- Add the Network Port object ----------------------------------------
    # Every BACnet device (Protocol_Revision 17+) must have at least one
    # Network Port object describing the port it talks on. This one is the
    # BACnet/IP application port; it is the lowest layer, so its reference
    # port is "none". networkNumber 0 with quality "unknown" describes a
    # local port that has not learned its network number - the right answer
    # for a device that is not a router and has not been told one.
    if not bacnet.BACnetStack_AddNetworkPortObject(
            g_device_instance, NETWORK_PORT_INSTANCE,
            c.NETWORK_PORT_NETWORK_TYPE_IPV4,
            c.NETWORK_PORT_PROTOCOL_LEVEL_BACNET_APPLICATION,
            0,  # networkNumber: not configured
            c.NETWORK_NUMBER_QUALITY_UNKNOWN,
            c.NETWORK_PORT_REFERENCE_PORT_NONE):
        print("Error: Failed to add Network Port 1 (Vermilion).")
        return 1

    # --- Enable the OPTIONAL properties we choose to expose -----------------
    # The stack automatically enables an object's REQUIRED properties when the
    # object is added (AddObject / AddNetworkPortObject) - so Units, Polarity,
    # Number_Of_States, Out_Of_Service, and the Network Port's BACnet/IP
    # addressing (IP_Address, IP_Subnet_Mask, BACnet_IP_UDP_Port, ...) are
    # already enabled; our Get* callbacks just supply their values. Only
    # OPTIONAL properties need SetPropertyEnabled. State_Text is optional on a
    # Multi-State Input, so we enable it here (and serve it in
    # get_property_character_string).
    #
    # The Device's Description is optional too, and it is an easy one to get
    # wrong: serving it from a Get callback is NOT enough. The stack checks
    # IsPropertyEnabled BEFORE it ever reaches the callbacks, and for an
    # optional property that check falls back to "is it required?" - which is
    # False. So a Description branch in the callback without this enable is
    # DEAD CODE, and the client reads back Error: unknown-property.
    if not bacnet.BACnetStack_SetPropertyEnabled(g_device_instance, c.OBJECT_TYPE_DEVICE,
                                                 g_device_instance, c.PROPERTY_IDENTIFIER_DESCRIPTION, True):
        print("Error: Failed to enable Description on the Device object.")
        return 1
    if not bacnet.BACnetStack_SetPropertyEnabled(g_device_instance, c.OBJECT_TYPE_MULTI_STATE_INPUT,
                                                 MULTI_STATE_INPUT_INSTANCE,
                                                 c.PROPERTY_IDENTIFIER_STATE_TEXT, True):
        print("Error: Failed to enable State_Text on Multi-State Input 1 (Hot Pink).")
        return 1

    # Reliability is optional on an Analog Input; it is how a sensor says its
    # reading cannot be trusted. Enabling it also gives the FAULT bit of
    # Status_Flags its meaning - the stack derives that bit from it.
    if not bacnet.BACnetStack_SetPropertyEnabled(g_device_instance, c.OBJECT_TYPE_ANALOG_INPUT,
                                                 ANALOG_INPUT_INSTANCE, PROPERTY_IDENTIFIER_RELIABILITY, True):
        print("Error: Failed to enable Reliability on Analog Input 1 (Bronze).")
        return 1

    # The Simulation's test-only broken-wire hook (section 2).
    g_sensor_fault_file = os.environ.get(SENSOR_FAULT_FILE_VARIABLE, "")

    # Who-Is is answered automatically. The spec also requires a device to
    # announce itself on start-up, so broadcast an unsolicited I-Am now (to the
    # local subnet broadcast - the Network Port's own network).
    if not helper.send_i_am(g_device_instance, g_bacnet_ip_udp_port, NETWORK_PORT_INSTANCE):
        print("Warning: the start-up I-Am was not sent.")

    print(f"FYI: Device {g_device_instance} (\"{g_device_name}\") ready. "
          f"Vendor ID {VENDOR_IDENTIFIER}. Press 'h' for help.")

    # --- Graceful Ctrl+C shutdown -------------------------------------------
    running = [True]

    def stop_running(signum, frame):
        running[0] = False  # finish this tick and exit, instead of dying mid-tick

    signal.signal(signal.SIGINT, stop_running)
    if hasattr(signal, "SIGTERM"):
        signal.signal(signal.SIGTERM, stop_running)

    # --- Run the stack ------------------------------------------------------
    # BACnetStack_Tick() processes incoming messages and timers. Call it
    # continuously, step the sensor Simulation beside it, and poll the keyboard
    # for interactive commands. The stack is single-threaded by contract -
    # everything below runs in this one loop.
    try:
        while running[0]:
            bacnet.BACnetStack_Tick()
            sensor_simulation_tick()

            key = helper.poll_key()
            if key == helper.KeyCommand.HELP:
                helper.print_help(APP_NAME, APP_VERSION)
            elif key == helper.KeyCommand.QUIT:
                running[0] = False
            elif key == helper.KeyCommand.ARROW_UP:
                g_analog_input_1_value += 1.1
                print(f"Analog Input 1 (Bronze) = {g_analog_input_1_value:.1f} C")
            elif key == helper.KeyCommand.ARROW_DOWN:
                g_analog_input_1_value -= 1.1
                print(f"Analog Input 1 (Bronze) = {g_analog_input_1_value:.1f} C")

            time.sleep(0.001)  # 1 ms - be a good citizen, don't spin the CPU
    finally:
        helper.restore_input()
        udp.shutdown()
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))