The `apex` object

The root of the ApexOS runtime is a single instance of the ApexOS class, defined in apexos/core.py ("Root object of the ApexOS home automation"). By convention it is passed to and named apex everywhere — setup functions, helpers, and config flows all use this name:

from apexos.config_entries import ConfigEntry
from apexos.core import ApexOS

async def async_setup_entry(
    apex: ApexOS,
    config_entry: ConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    ...

Inside config flows it is available as self.apex; inside entities as self.apex.

Core attributes

Attribute	Type	Purpose
`apex.data`	`ApexDict`	Dictionary any integration can store data on
`apex.loop`	`asyncio.AbstractEventLoop`	The event loop
`apex.bus`	`EventBus`	Fire and listen for events
`apex.services`	`ServiceRegistry`	Register and call services
`apex.states`	`StateMachine`	Read and write entity states
`apex.config`	`Config`	Core configuration (incl. `config_dir`)
`apex.config_entries`	`ConfigEntries`	Config-entry manager
`apex.auth`	`AuthManager`	Authentication
`apex.http`	`ApexOSHTTP`	HTTP component API
`apex.state`	`CoreState`	Lifecycle state; see `apex.is_running` / `apex.is_stopping`

States — `apex.states`

state = apex.states.get("sensor.temperature")     # State | None (thread-safe)
apex.states.is_state("sun.sun", "above_horizon")  # bool
apex.states.async_all("light")                    # all states, optional domain filter
apex.states.async_set(
    entity_id, new_state,
    attributes=None, force_update=False, context=None,
)
apex.states.async_remove(entity_id)

async_set adds the entity if it does not exist; if only attributes change (not the state), last-changed is untouched. All async_* state methods must run in the event loop. Note that integrations providing entities normally never call async_set directly — they implement entities instead.

Services — `apex.services`

apex.services.async_register(
    domain, service, service_func,
    schema=None,                          # voluptuous schema for service data
    supports_response=SupportsResponse.NONE,
)
apex.services.async_remove(domain, service)

response = await apex.services.async_call(
    domain, service,
    service_data=None,
    blocking=False,       # True to wait for completion
    context=None,
    target=None,
    return_response=False,  # only valid with blocking=True
)

service_func receives a ServiceCall and may be sync, a callback, or a coroutine function; it can return a response dict when supports_response allows it. apex.services.services returns the current {domain: {service: Service}} mapping. Registration must run in the event loop (apex.verify_event_loop_thread guards this).

Events — `apex.bus`

unsub = apex.bus.async_listen(event_type, listener, event_filter=None)
unsub()  # stop listening

apex.bus.async_listen_once(event_type, listener)
apex.bus.async_fire(event_type, event_data)

Use the constant MATCH_ALL as event_type to listen to every event. An optional event_filter — a @callback-decorated callable returning a bool — decides whether the listener runs. Events carry a Context (user id, parent id) which you can propagate when reacting to them.

Async patterns

ApexOS is asyncio-based, and the naming convention is strict: functions prefixed async_ must be called from the event loop thread; unprefixed variants (e.g. ServiceRegistry.register, StateMachine.set) are the thread-safe wrappers.

@callback (from apexos.core) marks a function as safe to run in the event loop without being a coroutine.
apex.async_add_executor_job(func, *args) — run blocking (sync) code, such as a non-async client library, in the executor pool:
```
result = await apex.async_add_executor_job(client.fetch_data)
```
apex.async_create_task(coro) — create a task in the event loop.
apex.async_add_job(...) / apex.async_run_job(...) — schedule a callback, coroutine function, or plain function appropriately.
apex.verify_event_loop_thread(what) — raise/report when called from the wrong thread.

Helpers

The apexos.helpers package provides the higher-level APIs integrations use daily:

Entity platform

Platform modules (e.g. sensor.py) implement async_setup_entry and receive an async_add_entities callback (AddConfigEntryEntitiesCallback from apexos.helpers.entity_platform):

from apexos.helpers.entity_platform import AddConfigEntryEntitiesCallback

async def async_setup_entry(
    apex: ApexOS,
    config_entry: ConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    async_add_entities([MySensor(...)])

Device registry

from apexos.helpers import device_registry as dr

registry = dr.async_get(apex)
device = registry.async_get_or_create(config_entry_id=..., identifiers={...}, ...)

Entities attach device metadata declaratively via DeviceInfo (apexos.helpers.device_registry.DeviceInfo):

device_info = DeviceInfo(
    identifiers={(DOMAIN, board)},
    manufacturer="Raspberry Pi",
    model=MODELS.get(board),
    name=BOARD_NAMES.get(board, "Raspberry Pi"),
)

Entity registry

from apexos.helpers import entity_registry as er

registry = er.async_get(apex)

The entity registry tracks unique IDs, user customizations (name, disabled, hidden), and the mapping to entity_ids.

HTTP sessions

from apexos.helpers.aiohttp_client import async_get_clientsession

session = async_get_clientsession(apex)

Share the core's aiohttp session instead of creating your own.

Raising setup problems

From apexos.exceptions, raise PlatformNotReady (platform setup will be retried) or ApexOSError (base error class) — for example when a backing service is not available yet:

from apexos.exceptions import PlatformNotReady

raise PlatformNotReady(
    translation_domain=DOMAIN,
    translation_key="supervisor_not_ready",
)

Core attributes​

States — apex.states​

Services — apex.services​

Events — apex.bus​

Async patterns​

Helpers​

Entity platform​

Device registry​

Entity registry​

HTTP sessions​

Raising setup problems​