Skip to content

Developer & Technical Reference Guide

Jump to bottom
acdcnow edited this page Sep 21, 2026 · 3 revisions

ARCHIVED - historical reference only. This page documents a superseded version of the integration and is no longer maintained. For current documentation see the Home page and the HA 2026.09 Development Branch page. Full index: Archived Documentation.

📘 Developer & Technical Reference: Crow/AAP IP Module Integration

This document provides a deep dive into the codebase, intended for developers maintaining the integration or users wanting to understand the "magic" behind the scenes.

1. System Architecture

This integration acts as a bridge between Home Assistant and the Crow Alarm system. It relies on the Local Push model, meaning the alarm panel "pushes" updates to Home Assistant instantly via a TCP socket, rather than Home Assistant constantly asking "Are you open?" (Polling).

Data Flow Diagram

  1. The TCP Socket: The pycrowipmodule library opens a persistent TCP connection to the IP Module (port 5002).
  2. The Listener Thread: The library runs a background thread that constantly listens for incoming ASCII data from the alarm.
  3. The Bridge (__init__.py): When data arrives, the library triggers a callback. Since this callback happens in the background thread, we must "bridge" it over to the Home Assistant Main Loop to update entities safely.
  4. The Dispatcher: Once on the main loop, a signal (e.g., crowipmodule.zones_updated) is broadcast.
  5. The Entities: Specific entities (like binary_sensor.kitchen_window) listen for their specific signal and update their state in the UI.

2. File-by-File Code Explanation

A. manifest.json (The Identity Card)

This file tells Home Assistant who we are and what we need.

{
  "domain": "crowipmodule",
  "requirements": ["pycrowipmodule>=0.32"],  // <--- PIP Dependency
  "config_flow": true,                        // <--- Tells HA we have a UI setup
  "iot_class": "local_push"                   // <--- Quality Scale: We are fast!
}
  • requirements: When you install this integration, Home Assistant automatically downloads pycrowipmodule version 0.32 or higher from PyPI. This library handles the complex ASCII protocol parsing.

B. config_flow.py (The Setup Wizard)

Role: Handles the UI forms when you click "Add Integration" or "Configure".

Key Concept: The State Machine

The flow is designed as a sequence of steps. Data is collected in temporary variables (self.areas_config, etc.) and only written to disk in the final step.

Code Breakdown: The Zone Step (async_step_zones)

async def async_step_zones(self, user_input=None):
    # 1. If form is submitted (user_input has data)
    if user_input is not None:
        self.zones_config = user_input
        return await self.async_step_connection() # Move to next step

    # 2. Build the form schema
    schema = {}
    for i in range(1, 17):
        # Create a text field for the Name
        schema[vol.Optional(f"zone_{i}_name")] = str
        
        # Create a Dropdown for the Type
        # vol.In(ZONE_TYPES) validates that the input MUST be one of the list items
        # This renders as a Dropdown menu in the UI
        schema[vol.Optional(f"zone_{i}_type", default="motion")] = vol.In(ZONE_TYPES)

    # 3. Show the form to the user
    return self.async_show_form(step_id="zones", data_schema=vol.Schema(schema))
  • vol.Optional: Means the user can leave the field empty.
  • vol.In(...): This is the magic that creates the dropdown list in the UI.

Key Concept: Robust Options Handling

In CrowOptionsFlowHandler, we use defensive coding to prevent the "500 Internal Server Error".

# Unsafe way (Causes crashes if "areas" doesn't exist):
# current = self.config_entry.options["areas"]

# Safe way (Used in our code):
current = self.config_entry.options.get(CONF_AREAS) or {}

Explanation: options.get() returns None if the key is missing. The or {} part ensures that current becomes an empty dictionary {} instead of None, allowing subsequent code like current.get("1") to work without crashing.

C. __init__.py (The Brain)

Role: Initializes the connection and handles thread safety.

1. The Thread-Safe Wrapper This is the most critical function in the entire integration.

def _thread_safe_send(signal, data):
    """Helper to send dispatcher signals thread-safely."""
    # hass.loop.call_soon_threadsafe tells HA:
    # "Please run this function (async_dispatcher_send) on your main loop
    # as soon as you have free time."
    hass.loop.call_soon_threadsafe(async_dispatcher_send, hass, signal, data)
  • Why? The pycrowipmodule library runs in a separate thread. If that thread tries to touch Home Assistant objects directly, Python will crash with a RuntimeError because HA objects are not thread-safe.

2. The Non-Blocking Start

# OLD (Blocking - BAD):
# await hass.async_add_executor_job(controller.start)
# Problem: If the alarm IP is wrong, HA waits 60s here and fails to boot.

# NEW (Non-Blocking - GOOD):
hass.async_add_executor_job(controller.start)
  • Explanation: We start the connection task but do not await it. This creates a "Fire and Forget" task. Home Assistant considers the setup "Complete" immediately and finishes booting. The controller quietly tries to connect in the background.

D. alarm_control_panel.py (The Operator)

Role: Represents the partition (Area A/B) and handles arming logic.

1. Keypad Logic

@property
def code_format(self) -> CodeFormat | None:
    # Always return NUMBER. This forces the UI to show the keypad.
    return CodeFormat.NUMBER

@property
def code_arm_required(self) -> bool:
    # If a code is hardcoded in settings (self._code), return False.
    # False means: "User input is allowed but NOT required".
    if self._code:
        return False
    return self._code_arm_required_config

2. The Arming Sequence

Crow alarms are particular, and there is no single "arm with code" command. Some panels arm on the bare command; others send ARM / STAY and then wait for the user code followed by Enter. 2.0.0 always followed up with the code, and 2.1.0-beta.2 makes that a per-entry choice — the arm_sequence option.

async def _async_arm(self, *, stay: bool, code: str | None) -> None:
    command = self._controller.arm_stay if stay else self._controller.arm_away
    if self._arm_sequence == ARM_SEQUENCE_COMMAND_ONLY:
        command()
        return
    command()

    keypad_code = self._resolve_code(code)

    if keypad_code:
        self._controller.send_keypress(keypad_code)

send_keypress(code) writes KEYS <code>E. The trailing E is the ENTER key, so that one line is the keypad equivalent of typing the code and pressing Enter — and it is byte-identical to what disarm() sends (KEYS <code>E, then STATUS ). The panel decides from its current state, which is why the same sequence both finishes an arming and performs a disarm.

E. binary_sensor.py (The Eyes)

Role: Monitors Zones and System Health.

1. Creating Entities Dynamically In async_setup_entry, we loop through the configuration to create sensors.

# 1. Get the list of zones configured in Step 3 of Config Flow
configured_zones = options.get(CONF_ZONES, {})

# 2. Create an entity for each zone
for zone_num_str, zone_info in configured_zones.items():
    entities.append(CrowZoneSensor(..., zone_info["name"], ...))

2. Logic Inversion (Diagnostic Sensors) Diagnostic sensors often need to invert the raw logic from the hardware.

# Example: Battery Sensor
# Hardware sends: True = Battery OK
# Home Assistant expects: On = Problem (Low Battery)

if self._attr_device_class == BinarySensorDeviceClass.BATTERY:
    return not val  # Invert True to False (Off = OK)

F. switch.py (The Hands)

Role: Controls outputs (e.g., Garage Door).

Optimistic State Updates When you toggle a switch in HA, we update the UI immediately without waiting for the hardware to confirm. This makes the UI feel snappy.

async def async_turn_on(self, **kwargs):
    # 1. Send command to hardware
    self._controller.command_output(str(self._output_number))
    
    # 2. Update internal state immediately (Optimistic)
    self._is_on = True
    
    # 3. Tell HA to refresh the UI
    self.async_write_ha_state()

3. Data Structures & Signals

The Signal Map

How does data get from __init__.py to the specific sensor? We use unique strings (Signals).

Signal Name Payload Who Listens?
crowipmodule.zones_updated zone_id (int) CrowZoneSensor
crowipmodule.areas_updated area_id (int) CrowAlarmPanel
crowipmodule.output_updated output_id (int) CrowOutput
crowipmodule.system_updated Full Data Dict CrowSystemStatusSensor

Example of Listening: In binary_sensor.py:

async_dispatcher_connect(self.hass, SIGNAL_ZONE_UPDATE, self._update_callback)

This line effectively subscribes the sensor to the zones_updated "radio station".

Device Registry

We link all entities to one device using device_info.

@property
def device_info(self) -> DeviceInfo:
    return DeviceInfo(
        # This tuple uniquely identifies the device in HA's registry
        identifiers={(DOMAIN, "crow_alarm_panel")}, 
        name="Crow Alarm System",
        # ...
    )

Because every entity (Sensor, Switch, Panel) returns this exact same identifiers tuple, Home Assistant knows they all belong to the same physical box.

Clone this wiki locally