Services
The typed service layer is the supported write path. Services own business logic, enforce object scope, and accept Pydantic command objects.
micboard.services.core.hardware
Manufacturer-neutral normalization for hardware payloads.
NormalizedHardware
Bases: PydanticBaseDTO
Normalized hardware payload independent of manufacturer key names.
canonicalize_mac(value: Any) -> str
Canonicalize hardware identity on creation and assignment.
from_api(data: dict[str, Any]) -> NormalizedHardware | None
Best-effort normalization for heterogeneous vendor payloads.
micboard.services.core.hardware_lifecycle
Device lifecycle management service.
Handles all device state transitions, validation, and bi-directional sync with manufacturer APIs. Replaces signal-based state management with direct, testable method calls.
State Transitions: DISCOVERED → PROVISIONING → ONLINE ↓ DEGRADED → MAINTENANCE ↓ OFFLINE → RETIRED
HardwareStatus
Bases: StrEnum
Device lifecycle states.
HardwareLifecycleManager
Centralized manager for device lifecycle operations.
Handles:
- State transitions with validation
- Bi-directional sync with manufacturer APIs
- Health monitoring
Does NOT use signals for state management (signals only for broadcasts).
transition_device(device: WirelessChassis | WirelessUnit, to_status: str, reason: str = '', metadata: dict[str, Any] | None = None) -> bool
Transition a device on the database that supplied the instance.
mark_online(device: WirelessChassis | WirelessUnit, health_data: dict[str, Any] | None = None) -> bool
Mark device as online/operational.
mark_offline(device: WirelessChassis | WirelessUnit, reason: str = 'Not responding') -> bool
Mark device as offline.
map_api_state_to_status(api_state: str, current_status: str) -> str
Map manufacturer API state to HardwareStatus.
micboard.services.core.hardware_sync
Hardware sync operations for status, battery, and channel management.
Provides write operations that synchronize hardware state from external sources.
HardwareSyncService
sync_hardware_status(obj: WirelessChassis | WirelessUnit, online: bool) -> None
Update hardware online status.
Uses direct status update - lifecycle hooks handle timestamps, audit, broadcast.
ensure_channel_count(chassis: WirelessChassis, using: str = DEFAULT_DB_ALIAS) -> tuple[int, int]
Ensure RFChannel rows for a chassis match its model capacity.
Returns (created_count, deleted_count).
micboard.services.core.performer_assignment
Performer assignment service layer for binding devices to performers.
Handles assignment lifecycle, state transitions, and lookup helpers.
PerformerAssignmentService
Business logic for performer-to-device assignments.
get_visible_assignments(user: Any) -> QuerySet[PerformerAssignment]
Return user-scoped assignments with all row relations eager loaded.
get_visible_assignment_rows(user: Any, page: int | str | None = 1) -> QuerySet[PerformerAssignment]
Return one bounded live-refresh slice without a count query.
get_preferred_active_assignments_for_units(user: Any, unit_ids: Collection[int]) -> QuerySet[PerformerAssignment]
Return at most one deterministic active assignment for each requested unit.
get_preferred_active_assignments_for_serials(user: Any, serial_numbers: Collection[str]) -> QuerySet[PerformerAssignment]
Return at most one deterministic active assignment for each requested serial.
ensure_group_can_manage_unit(group: MonitoringGroup, unit: WirelessUnit) -> None
Require the selected group to cover the unit in tenant-aware deployments.
ensure_can_modify_unit(user: Any, unit: WirelessUnit) -> None
Require an MSP role that permits assignment changes for the unit.
create_assignment(command: CreatePerformerAssignment, user: Any) -> PerformerAssignment
Create an assignment after validating every object against user scope.
update_assignment(command: UpdatePerformerAssignment, user: Any) -> PerformerAssignment
Update fields on an existing assignment and return the instance.
Raises PerformerAssignment.DoesNotExist if the assignment is missing.
delete_assignment(assignment_id: int, user: Any) -> bool
Permanently delete an assignment. Returns True if deleted, False if not found.
deactivate_assignment(assignment_id: int, user: Any) -> bool
Deactivate an existing assignment.
micboard.services.core.performer_assignment_dtos
Validated commands for performer assignment writes.
CreatePerformerAssignment
Bases: PydanticBaseDTO
Data required to bind a performer, wireless unit, and monitoring group.
UpdatePerformerAssignment
Bases: PydanticBaseDTO
Partial assignment update with omission distinct from false or blank.
micboard.services.core.device_metadata
Device metadata accessor pattern for manufacturer-agnostic metadata handling.
Provides strategy pattern for accessing manufacturer-specific metadata from DiscoveredDevice models without hardcoding assumptions about metadata structure.
DeviceMetadataAccessor
Bases: ABC
Abstract base for accessing manufacturer-specific metadata.
get_compatibility_status() -> str | None
Get device compatibility status (e.g., COMPATIBLE, INCOMPATIBLE_TOO_OLD).
get_device_state() -> str | None
Get device state (e.g., DISCOVERED, ONLINE, OFFLINE).
get_incompatibility_reason() -> str | None
Get human-readable reason if device is incompatible.
get_for(manufacturer: Manufacturer | None, device_data: dict[str, Any] | None = None) -> DeviceMetadataAccessor
Factory method to get appropriate accessor for manufacturer.
Parameters:
manufacturer(Manufacturer | None) — Manufacturer instance (can be None).device_data(dict[str, Any] | None) — Metadata dict from DiscoveredDevice.
Returns:
DeviceMetadataAccessor— Appropriate DeviceMetadataAccessor subclass instance.
GenericMetadataAccessor
Bases: DeviceMetadataAccessor
Generic accessor for unknown manufacturers.
get_compatibility_status() -> str | None
get_device_state() -> str | None
get_incompatibility_reason() -> str | None
ShureMetadataAccessor
Bases: DeviceMetadataAccessor
Accessor for Shure-specific metadata structure.
get_compatibility_status() -> str | None
Get Shure compatibility status.
get_device_state() -> str | None
Get Shure device state.
get_incompatibility_reason() -> str | None
Get human-readable Shure incompatibility reason.
get_communication_protocol() -> str | None
Get Shure communication protocol name.
SennheiserMetadataAccessor
Bases: DeviceMetadataAccessor
Accessor for Sennheiser-specific metadata structure.
get_compatibility_status() -> str | None
Get Sennheiser compatibility status.
get_device_state() -> str | None
Get Sennheiser device state.
get_incompatibility_reason() -> str | None
Get human-readable Sennheiser incompatibility reason.
micboard.services.core.device_specs
Device specifications service for centralized spec lookups.
Provides a clean interface for accessing device specifications from the registry without embedding lookup logic in models.
DeviceSpec
Device specifications from registry.
DeviceSpecService
Service for looking up standardized device specifications.
get_specs(manufacturer: Manufacturer | None, model: str) -> DeviceSpec | None
Get device specifications for a manufacturer/model combination.
Parameters:
manufacturer(Manufacturer | None) — Manufacturer instance (can be None).model(str) — Device model name/code.
Returns:
DeviceSpec | None— DeviceSpec instance with specs, or None if not found.
apply_specs_to_chassis(chassis: object) -> None
Apply specifications to a WirelessChassis instance (for use in save()).
Parameters:
chassis(object) — WirelessChassis instance with manufacturer and model set.
micboard.services.core.user_profile
User profile preference operations.
UserProfileService
Persist validated user profile preferences.
set_display_width(user: Any, width_px: int) -> UserProfile
Set a user’s charger-dashboard width within supported browser limits.
micboard.services.settings.settings_service
Unified settings resolution service.
Composes Django settings, feature flags, package defaults, and the DB-backed SettingsRegistry into a single resolution chain.
SettingsService
Unified settings resolution with multi-source fallback.
Resolution order for get():
- Deployment controls mapped to immutable Django
MICBOARD_*settings - DB Setting with scope (org/site/manufacturer) via
SettingsRegistry settings.MICBOARD_CONFIGdict key- Package defaults (
POLL_INTERVAL, etc.) - Registered
SettingDefinitiondefault - Provided default
get(key: str, default: Any = None, organization: Any = None, site: Any = None, manufacturer: Any = None) -> Any
Resolve a setting value through the multi-source fallback chain.
Parameters:
key(str) — Canonical setting key.default(Any) — Fallback if not found in any source.organization(Any) — Scope hint for DB-backed setting.site(Any) — Scope hint for DB-backed setting.manufacturer(Any) — Scope hint for DB-backed setting.
Returns:
Any— Resolved value or default.
get_config_dict() -> dict[str, Any]
Return MICBOARD_CONFIG merged with package defaults.
invalidate_value_cache(key: str | None = None) -> None
Invalidate one resolved database value or every cached value.
invalidate_definition_cache(key: str | None = None) -> None
Invalidate definition metadata and every value derived from it.
micboard.services.settings.browser_refresh_service
One module that decides how often each live browser surface re-polls the server.
Micboard delivers every live browser update by short-polling over ordinary HTTP, so the poll interval multiplied by the number of open tabs is the entire request volume the deployment’s reverse proxy carries. Leaving each interval as a literal in its template put that number out of a deployer’s reach: slowing a busy page down meant forking presentation markup. This module owns the decision instead, resolving each surface through the same host-configuration seam the rest of Micboard uses and clamping the result to the bounds that already govern stored kiosk refresh rates.
bounded_refresh_interval(value: Any, default: int) -> int
Return one refresh interval clamped into the range a browser can be trusted with.
MICBOARD_CONFIG is host-supplied and unvalidated, so this has to survive anything a
deployment puts there. An interval that cannot be read as a whole number falls back to
default rather than raising, because a browser surface that will not render is worse
than one refreshing at the shipped rate.
Parameters:
value(Any) — Candidate interval from host configuration or a stored row.default(int) — Interval to use when value cannot be read as a whole number.
Returns:
int— A whole number of seconds within the shared refresh bounds.
BrowserRefreshCadence
Resolve the bounded poll interval for one named browser surface.
seconds_for(surface: str) -> int
Return how many seconds surface waits between refreshes.
Parameters:
surface(str) — A key of :data:BROWSER_REFRESH_SURFACES.
Returns:
int— The configured interval, clamped to the shared refresh bounds.
Raises:
ValueError— If surface is not a declared browser refresh surface.
milliseconds_for(surface: str) -> int
Return the same bounded interval as a JavaScript timer duration.
Parameters:
surface(str) — A key of :data:BROWSER_REFRESH_SURFACES.
Returns:
int— The configured interval in milliseconds.
micboard.services.settings.registry
Internal registry implementation for typed, explicitly scoped settings.
SettingsScopeReference
Bases: Protocol
Minimal model contract required to identify one settings scope.
SettingsRegistry
Centralized settings accessor honoring each definition’s declared scope.
get(key: str, default: Any = None, organization: SettingsScopeReference | None = None, site: SettingsScopeReference | None = None, manufacturer: SettingsScopeReference | None = None, required: bool = False, include_definition_default: bool = True) -> Any
Get a setting value at its definition’s declared scope.
Resolution order:
- Stored value at the definition’s exact scope
- SettingDefinition default, when requested
- User-provided default
- Raise if required
Parameters:
key(str) — Setting keydefault(Any) — Fallback default valueorganization(SettingsScopeReference | None) — Organization for scope (MSP mode)site(SettingsScopeReference | None) — Site for scope (multi-site mode)manufacturer(SettingsScopeReference | None) — Manufacturer for scoperequired(bool) — Raise error if not foundinclude_definition_default(bool) — Whether to use the registered definition default
Returns:
Any— Resolved setting value
Raises:
SettingNotFoundError— If required=True and not found
get_definition_default(key: str, default: Any = None) -> Any
Return the typed definition default without consulting stored values.
invalidate_cache(key: str | None = None) -> None
Invalidate settings cache.
Parameters:
key(str | None) — Specific key to invalidate, or None for all
invalidate_definition(key: str | None = None) -> None
Invalidate cached definition metadata and all values derived from it.
micboard.services.settings.dtos
Data-transfer objects for tenant-aware settings presentation.
SettingsVisibilityScope
Bases: PydanticBaseDTO
Identifiers whose stored overrides a user may inspect.
None means unrestricted access for that dimension. An empty set means
fail closed and expose no overrides for that dimension.
SettingWriteTarget
Bases: PydanticBaseDTO
One exact global, organization, site, or manufacturer scope.
validate_exact_scope() -> SettingWriteTarget
Reject identifiers that do not match the declared scope.
SettingWriteItem
Bases: PydanticBaseDTO
One validated form value awaiting definition serialization.
validate_identifier() -> SettingWriteItem
Require exactly one definition identifier.
SettingsWriteRequest
Bases: PydanticBaseDTO
Authorized batch of setting override writes.
SettingsWriteResult
Bases: PydanticBaseDTO
Best-effort persistence result safe to render to an operator.
micboard.services.shared.access_policy
Shared policy for tenant-wide read and mutation access decisions.
has_unrestricted_tenant_access(user: Any) -> bool
Return whether user may bypass organization membership boundaries.
visible_to(model: type[models.Model], user: Any, using: str | None = None) -> models.QuerySet[Any]
Return the rows of model that user may see.
Models with a tenant-aware manager narrow visibility themselves, sometimes with a model-specific rule on top of the shared cascade; models on Django’s default manager get the shared cascade directly. Callers ask the same question either way.
The database is bound before the tenant boundary is applied, not after. Answering this
in MSP mode takes two reads: for_user materialises the caller’s active memberships as
it builds the queryset, so retargeting only the finished queryset would leave that
boundary read on whichever database the manager defaulted to.
TenantRoleAccessService
Apply MSP membership roles without narrowing read-only visibility.
management_memberships(user: Any, using: str | None = None) -> list[tuple[int, int | None]]
Return active organization/campus scopes where user may administer.
is_platform_global_model(model: type[models.Model]) -> bool
Return whether model is a reviewed host-wide admin surface.
scope_manageable_queryset(queryset: models.QuerySet[Any], user: Any) -> models.QuerySet[Any]
Intersect queryset with scopes where user has an admin role.
can_add_model(user: Any, model: type[models.Model]) -> bool
Authorize adds only where a new row can carry exclusive tenant ownership.
can_manage_model(user: Any, model: type[models.Model]) -> bool
Authorize adding or bulk-mutating rows of one tenant-owned model.
can_manage_object(user: Any, obj: models.Model) -> bool
Authorize mutation only when the object’s exact tenant role permits it.
micboard.services.shared.base_dto
Base DTO class for all data transfer objects in the service layer.
All DTOs should inherit from this class to ensure consistent configuration.
PydanticBaseDTO
Bases: BaseModel
Base DTO with standard configuration for all service layer DTOs.
micboard.services.common.base.plugin
clear_plugin_cache() -> None
Forget resolved plugin classes, so a test starts from a cold cache.
build_manufacturer_plugin(manufacturer: Manufacturer) -> ManufacturerPlugin
Return a plugin bound to manufacturer.
This is the one way to obtain a plugin. It raises when a manufacturer has no shipped
integration, so every caller sees the same failure rather than a None some branch on
and others do not.
get_manufacturer_plugin(code: str) -> type[ManufacturerPlugin]
Return the plugin class for a manufacturer code, resolving it at most once.
Attempts to import micboard.integrations.<code>.plugin and
locate a concrete ManufacturerPlugin subclass. Prefers
<CodeTitle>Plugin, then falls back to another plugin subclass.
BasePlugin
Bases: ABC
Base interface for all manufacturer plugins.
get_devices() -> list[dict[str, Any]]
Retrieve a list of all devices associated with or discovered by this plugin.
ManufacturerPlugin
Bases: BasePlugin
Extended plugin interface specifically for manufacturer hardware integrations.
async subscribe_to_chassis(chassis: WirelessChassis, callback: Callable[[dict[str, Any]], Awaitable[None]]) -> None
Open this integration’s stream for one chassis and await its updates.
The integration owns connection setup, authentication, framing, and cleanup; the runner owns leasing, inventory selection, connection tracking, and persistence.
get_device_channels(device_id: str) -> list[dict[str, Any]]
Retrieve all channels associated with a specific device identifier.
get_client() -> BaseAPIClient
Get an instance of the configured API client for this manufacturer.
transform_device_data(api_data: dict[str, Any]) -> dict[str, Any] | None
Transform raw API device data into the standardized application format.
transform_transmitter_data(api_data: dict[str, Any], channel_number: int) -> dict[str, Any] | None
Normalize one raw wireless-unit payload for a channel.
DeviceUpdateService requires this of every plugin it persists through, so it is
part of the contract rather than an optional addition.
get_device(device_id: str) -> dict[str, Any] | None
Fetch details for a single device by its identifier.
is_healthy() -> bool
Check if the plugin and its underlying integrations are currently healthy.
check_health() -> dict[str, Any]
Perform a detailed health check and return the results as a dictionary.
add_discovery_ips(ips: list[str]) -> bool
Add a list of IP addresses to the plugin’s discovery targets.
get_discovery_ips() -> list[str]
Retrieve the list of currently configured discovery IP addresses.
remove_discovery_ips(ips: list[str]) -> bool
Remove a list of IP addresses from the plugin’s discovery targets.
micboard.services.common.base.client
standardize_health_response(status: str, details: dict[str, Any] | None = None, error: str | None = None) -> dict[str, Any]
Return one health result in the shape every consumer reads.
Consumers — the admin, the API-health context processor, and the manufacturer health task —
read status from a closed vocabulary plus a timestamp, so an unrecognized status becomes
unknown rather than propagating a vendor’s own word for it.
BaseAPIClient
Bases: ABC
Base API client interface.
is_healthy() -> bool
Check if the client is healthy.
check_health() -> dict[str, Any]
Perform a health check and return details.
BaseHTTPClient
Bases: BaseAPIClient
Base HTTP client with circuit breaker and retries.
get_exception_class() -> type[APIError]
Get the exception class for API errors.
get_rate_limit_exception_class() -> type[APIRateLimitError]
Get the exception class for rate limit errors.
is_healthy() -> bool
check_health() -> dict[str, Any]
close() -> None
Close the underlying HTTP connection pool.
micboard.services.common.base.resilience
create_resilient_session(max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, follow_redirects: bool = True) -> httpx.Client
Create an HTTPX client with bounded connection retries and pooling.
HTTP status retries belong in the calling service because only that layer knows whether a request is safe to replay and how to interpret Retry-After.
micboard.services.common.base.rate_limiter
rate_limit(calls_per_second: float = 10.0) -> Callable[[_CallableT], _CallableT]
Rate-limit calls to a decorated client method through the shared cache.