Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -101,8 +101,10 @@ poetry run neongetter
- `authorizations_or`: List of authorizations, any one sufficient to operate
- `unauthorized_warn_only`: (optional) Allow operation but log warning for unauthorized users
- `always_enabled`: (optional) Machine always enabled without RFID authentication, displays "Always On"
- `alias`: (optional) Human-friendly name used in Slack messages and logs instead of machine name
- Users: `users.json` (schema in `models/users.py::CONFIG_SCHEMA`)
- Machine names must match ESPHome configs and can only contain `[a-z0-9_-]`
- Machines can be looked up by either name or alias in Slack commands

**State Persistence**:
- Machine state is persisted to disk on every update using pickle
Expand Down
2 changes: 2 additions & 0 deletions docs/source/configuration.rst
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ machines.json

Machines are configured via a ``machines.json`` file in the current directory, or another file name/path specified in the ``MACHINES_CONFIG`` environment variable. This file lists all of the supported/configured machines and which authorization(s) are required to use them. Note that the names in this file must match the names configured in your ESPHome :ref:`hardware`. Machine names must be unqiue and can only contain alphanumeric characters, underscores, and dashes. No spaces, no dots.

Each machine configuration supports an optional ``alias`` field, which provides a human-friendly name for the machine. When present, the alias will be used in Slack messages and log output instead of the machine name. Both the machine name and alias can be used in incoming Slack commands.

The schema of this file is as follows:

.. jsonschema:: dm_mac.models.machine.CONFIG_SCHEMA
Expand Down
8 changes: 5 additions & 3 deletions docs/source/slack.rst
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,10 @@ The slack bot is controlled by mentioning its name (``@your-bot-name``) along wi
Using an example bot name of ``@machine-access-control``, the supported commands are:

* ``@machine-access-control status`` - List all machines and their current status. This command is the only one that is usable from channels other than the control channel.
* ``@machine-access-control oops <machine-name>`` - Set Oops'ed status on the machine with name ``machine-name``. This takes effect immediately, even if the machine is currently in use.
* ``@machine-access-control lock <machine-name>`` - Set maintenance lock-out status on the machine with name ``machine-name``. This takes effect immediately, even if the machine is currently in use.
* ``@machine-access-control clear <machine-name>`` - Clear all Oops and/or maintenance lock-out states on the machine with name ``machine-name``.
* ``@machine-access-control oops <machine-name>`` - Set Oops'ed status on the machine with name ``machine-name``. This takes effect immediately, even if the machine is currently in use. You can use either the machine name or its alias (if configured).
* ``@machine-access-control lock <machine-name>`` - Set maintenance lock-out status on the machine with name ``machine-name``. This takes effect immediately, even if the machine is currently in use. You can use either the machine name or its alias (if configured).
* ``@machine-access-control clear <machine-name>`` - Clear all Oops and/or maintenance lock-out states on the machine with name ``machine-name``. You can use either the machine name or its alias (if configured).

**Note:** If a machine has an ``alias`` configured in ``machines.json``, the bot's responses will use the alias instead of the machine name for better readability.

In addition, changes to all machines' Oops and maintenance lock-out states will be posted as messages in the ``SLACK_OOPS_CHANNEL_ID`` channel.
79 changes: 53 additions & 26 deletions src/dm_mac/models/machine.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,11 @@
"Displays 'Always On' and relay is always "
"on unless Oopsed or Locked.",
},
"alias": {
"type": "string",
"description": "Optional human-friendly alias for the machine. "
"Used in Slack messages and logs instead of the machine name.",
},
},
"additionalProperties": False,
"description": "Unique machine name, alphanumeric _ and - only.",
Expand All @@ -77,6 +82,7 @@ def __init__(
authorizations_or: List[str],
unauthorized_warn_only: bool = False,
always_enabled: bool = False,
alias: Optional[str] = None,
):
"""Initialize a new MachineState instance."""
#: The name of the machine
Expand All @@ -88,6 +94,8 @@ def __init__(
self.unauthorized_warn_only: bool = unauthorized_warn_only
#: Whether machine is always enabled without RFID authentication
self.always_enabled: bool = always_enabled
#: Optional human-friendly alias for the machine
self.alias: Optional[str] = alias
#: state of the machine
self.state: "MachineState" = MachineState(self)

Expand Down Expand Up @@ -145,6 +153,11 @@ async def unoops(self, slack: Optional["SlackHandler"] = None) -> None:
return
await slack.log_unoops(self, source)

@property
def display_name(self) -> str:
"""Return the display name for this machine (alias if present, else name)."""
return self.alias if self.alias else self.name

@property
def as_dict(self) -> Dict[str, Any]:
"""Return a dict representation of this machine."""
Expand All @@ -153,6 +166,7 @@ def as_dict(self) -> Dict[str, Any]:
"authorizations_or": self.authorizations_or,
"unauthorized_warn_only": self.unauthorized_warn_only,
"always_enabled": self.always_enabled,
"alias": self.alias,
}


Expand All @@ -163,15 +177,24 @@ def __init__(self) -> None:
"""Initialize MachinesConfig."""
logger.debug("Initializing MachinesConfig")
self.machines_by_name: Dict[str, Machine] = {}
self.machines_by_alias: Dict[str, Machine] = {}
self.machines: List[Machine] = []
mdict: Dict[str, Any]
mname: str
for mname, mdict in self._load_and_validate_config().items():
mach: Machine = Machine(name=mname, **mdict)
self.machines.append(mach)
self.machines_by_name[mach.name] = mach
if mach.alias:
self.machines_by_alias[mach.alias] = mach
self.load_time: float = time()

def get_machine(self, name_or_alias: str) -> Optional[Machine]:
"""Get a machine by name or alias."""
return self.machines_by_name.get(name_or_alias) or self.machines_by_alias.get(
name_or_alias
)

def _load_and_validate_config(self) -> Dict[str, Dict[str, Any]]:
"""Load and validate the config file."""
config: Dict[str, Dict[str, Any]] = cast(
Expand Down Expand Up @@ -307,7 +330,8 @@ async def _handle_reboot(self) -> None:
For always-enabled machines, restores the always-on state.
"""
logging.getLogger("AUTH").warning(
"Machine %s rebooted; resetting relay and RFID state", self.machine.name
"Machine %s rebooted; resetting relay and RFID state",
self.machine.display_name,
)
# locking handled in update()
self.current_user = None
Expand All @@ -327,12 +351,12 @@ async def _handle_reboot(self) -> None:
if not slack:
# Slack integration is not enabled
return
await slack.admin_log(f"Machine {self.machine.name} has rebooted.")
await slack.admin_log(f"Machine {self.machine.display_name} has rebooted.")

def lockout(self) -> None:
"""Lock-out the machine."""
logging.getLogger("OOPS").warning(
"Machine %s was locked out for maintenance.", self.machine.name
"Machine %s was locked out for maintenance.", self.machine.display_name
)
with self._lock:
self.is_locked_out = True
Expand All @@ -345,7 +369,8 @@ def lockout(self) -> None:
def unlock(self) -> None:
"""Un-lock-out the machine."""
logging.getLogger("OOPS").warning(
"Machine %s was removed from maintenance lock-out.", self.machine.name
"Machine %s was removed from maintenance lock-out.",
self.machine.display_name,
)
with self._lock:
self.is_locked_out = False
Expand All @@ -364,7 +389,9 @@ def unlock(self) -> None:

def oops(self, do_locking: bool = True) -> None:
"""Oops the machine."""
logging.getLogger("OOPS").warning("Machine %s was Oopsed.", self.machine.name)
logging.getLogger("OOPS").warning(
"Machine %s was Oopsed.", self.machine.display_name
)
locker = self._lock if do_locking else nullcontext()
with locker:
self.is_oopsed = True
Expand All @@ -377,7 +404,7 @@ def oops(self, do_locking: bool = True) -> None:
def unoops(self, do_locking: bool = True) -> None:
"""Un-oops the machine."""
logging.getLogger("OOPS").warning(
"Machine %s was un-Oopsed.", self.machine.name
"Machine %s was un-Oopsed.", self.machine.display_name
)
locker = self._lock if do_locking else nullcontext()
with locker:
Expand Down Expand Up @@ -465,7 +492,7 @@ async def _handle_oops(self, users: UsersConfig) -> None:
ustr = f" Current user is: {user.full_name}."
uname = user.full_name
logging.getLogger("OOPS").warning(
"Machine %s was Oopsed.%s", self.machine.name, ustr
"Machine %s was Oopsed.%s", self.machine.display_name, ustr
)
# locking handled in update()
self.oops(do_locking=False)
Expand All @@ -485,12 +512,12 @@ async def _handle_rfid_remove(self) -> None:
"""Handle RFID card removed."""
logging.getLogger("AUTH").info(
"RFID logout on %s by %s; session duration %d seconds",
self.machine.name,
self.machine.display_name,
self.current_user.full_name if self.current_user else self.rfid_value,
time() - cast(float, self.rfid_present_since),
)
log_str: str = (
f"RFID logout on {self.machine.name} by "
f"RFID logout on {self.machine.display_name} by "
+ (self.current_user.full_name if self.current_user else "unknown")
+ "; session duration "
+ naturaldelta(time() - cast(float, self.rfid_present_since))
Expand Down Expand Up @@ -521,13 +548,13 @@ async def _handle_rfid_insert(self, users: UsersConfig, rfid_value: str) -> None
if not user:
logging.getLogger("AUTH").warning(
"RFID login attempt on %s by unknown fob %s",
self.machine.name,
self.machine.display_name,
rfid_value,
)
if self.is_oopsed or self.is_locked_out:
if slack:
await slack.admin_log(
f"RFID login attempt on {self.machine.name} "
f"RFID login attempt on {self.machine.display_name} "
"by unknown fob when oopsed or locked out."
)
return
Expand All @@ -536,34 +563,34 @@ async def _handle_rfid_insert(self, users: UsersConfig, rfid_value: str) -> None
self.status_led_brightness = self.STATUS_LED_BRIGHTNESS
if slack:
await slack.admin_log(
f"RFID login attempt on {self.machine.name} by unknown fob"
f"RFID login attempt on {self.machine.display_name} by unknown fob"
)
return
# ok, we have a known user
logname = f"{user.full_name} ({rfid_value})"
if self.is_oopsed:
logging.getLogger("AUTH").warning(
"RFID login attempt while oopsed on %s by %s",
self.machine.name,
self.machine.display_name,
logname,
)
# don't change anything
if slack:
await slack.admin_log(
f"RFID login attempt on {self.machine.name} by "
f"RFID login attempt on {self.machine.display_name} by "
f"{user.full_name} when oopsed."
)
return
if self.is_locked_out:
logging.getLogger("AUTH").warning(
"RFID login attempt while locked out on %s by %s",
self.machine.name,
self.machine.display_name,
logname,
)
# don't change anything
if slack:
await slack.admin_log(
f"RFID login attempt on {self.machine.name} by "
f"RFID login attempt on {self.machine.display_name} by "
f"{user.full_name} when machine locked-out."
)
return
Expand All @@ -572,7 +599,7 @@ async def _handle_rfid_insert(self, users: UsersConfig, rfid_value: str) -> None
"User %s (%s) authorized for %s; session start",
user.full_name,
user.account_id,
self.machine.name,
self.machine.display_name,
)
self.current_user = user
self.relay_desired_state = True
Expand All @@ -581,23 +608,23 @@ async def _handle_rfid_insert(self, users: UsersConfig, rfid_value: str) -> None
self.status_led_brightness = self.STATUS_LED_BRIGHTNESS
if slack:
await slack.admin_log(
f"RFID login on {self.machine.name} by authorized user "
f"RFID login on {self.machine.display_name} by authorized user "
f"{user.full_name}"
)
else:
logging.getLogger("AUTH").info(
"User %s (%s) UNAUTHORIZED for %s",
user.full_name,
user.account_id,
self.machine.name,
self.machine.display_name,
)
self.relay_desired_state = False
self.display_text = "Unauthorized"
self.status_led_rgb = (1.0, 0.5, 0.0) # orange
self.status_led_brightness = self.STATUS_LED_BRIGHTNESS
if slack:
await slack.admin_log(
f"rejected RFID login on {self.machine.name} by "
f"rejected RFID login on {self.machine.display_name} by "
f"UNAUTHORIZED user {user.full_name}"
)

Expand All @@ -614,7 +641,7 @@ async def _handle_rfid_tracking_always_enabled(
# RFID removed
logging.getLogger("AUTH").info(
"RFID removed on always-enabled machine %s (was %s); session duration %d seconds",
self.machine.name,
self.machine.display_name,
self.current_user.full_name if self.current_user else self.rfid_value,
(
time() - cast(float, self.rfid_present_since)
Expand All @@ -635,14 +662,14 @@ async def _handle_rfid_tracking_always_enabled(
self.current_user = user
logging.getLogger("AUTH").info(
"RFID inserted on always-enabled machine %s by %s (%s)",
self.machine.name,
self.machine.display_name,
user.full_name,
rfid_value,
)
else:
logging.getLogger("AUTH").warning(
"RFID inserted on always-enabled machine %s by unknown fob %s",
self.machine.name,
self.machine.display_name,
rfid_value,
)
# State remains always-on (relay/display/LED not changed)
Expand All @@ -657,7 +684,7 @@ async def _user_is_authorized(
"User %s (%s) authorized for %s based on %s",
user.full_name,
user.account_id,
self.machine.name,
self.machine.display_name,
auth,
)
return True
Expand All @@ -667,12 +694,12 @@ async def _user_is_authorized(
"unauthorized_warn_only==True",
user.full_name,
user.account_id,
self.machine.name,
self.machine.display_name,
)
if slack:
await slack.admin_log(
f"WARNING - Authorizing user {user.full_name} for "
f"{self.machine.name} based on unauthorized_warn_only "
f"{self.machine.display_name} based on unauthorized_warn_only "
"setting for machine. User is NOT authorized for this "
"machine."
)
Expand Down
Loading
Loading