diff --git a/pynut2/nut2.py b/pynut2/nut2.py
index ee4164c..636b5a7 100644
--- a/pynut2/nut2.py
+++ b/pynut2/nut2.py
@@ -3,8 +3,7 @@
"""A Python module for dealing with NUT (Network UPS Tools) servers.
* PyNUTError: Base class for custom exceptions.
-* PyNUTClient: Allows connecting to and communicating with PyNUT
- servers.
+* PyNUTClient: Allows connecting to and communicating with PyNUT servers.
Copyright (C) 2013 george2
@@ -21,14 +20,16 @@
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
-along with this program. If not, see .
+along with this program. If not, see .
"""
-import telnetlib
import logging
+import socket
+from telnetlib import Telnet
+from typing import Any, Dict, List, Optional
-__version__ = '2.1.2'
+__version__ = '2.1.3'
__all__ = ['PyNUTError', 'PyNUTClient']
_LOGGER = logging.getLogger(__name__)
@@ -42,8 +43,9 @@ class PyNUTError(Exception):
class PyNUTClient(object):
"""Access NUT (Network UPS Tools) servers."""
- def __init__(self, host="127.0.0.1", port=3493, login=None, password=None,
- timeout=5, persistent=True):
+ def __init__(self, host: str = '127.0.0.1', port: int = 3493,
+ login: Optional[str] = None, password: Optional[str] = None,
+ timeout: float = 5, persistent: bool = True) -> None:
"""Class initialization method.
host : Host to connect (defaults to 127.0.0.1).
@@ -58,89 +60,92 @@ def __init__(self, host="127.0.0.1", port=3493, login=None, password=None,
when calling each method
"""
- _LOGGER.debug("NUT Class initialization, Host/Port: %s:%s,"
- " Login: %s/%s", host, port, login, password)
+ _LOGGER.debug('NUT Class initialization, Host/Port: %s:%s, Login: %s', host, port, login)
- self._host = host
- self._port = port
- self._login = login
- self._password = password
- self._timeout = timeout
- self._persistent = persistent
- self._srv_handler = None
+ self._host: str = host
+ self._port: int = port
+ self._login: Optional[str] = login
+ self._password: Optional[str] = password
+ self._timeout: float = timeout
+ self._persistent: bool = persistent
+ self._srv_handler: Optional[Telnet] = None
if self._persistent:
self._connect()
- def __del__(self):
+ def __del__(self) -> None:
# Try to disconnect cleanly when class is deleted.
- _LOGGER.debug("NUT Class deleted, trying to disconnect.")
+ _LOGGER.debug('NUT Class deleted, trying to disconnect.')
self._disconnect()
- def __enter__(self):
+ def __enter__(self) -> 'PyNUTClient':
return self
- def __exit__(self, exc_t, exc_v, trace):
+ def __exit__(self, exc_t: type, exc_v: IndexError, trace: Any) -> None:
self.__del__()
- def _disconnect(self):
+ def _disconnect(self) -> None:
""" Disconnects from the defined server."""
if self._srv_handler:
try:
- self._write("LOGOUT\n")
+ self._write('LOGOUT\n')
self._srv_handler.close()
- except (telnetlib.socket.error, AttributeError):
+ except (socket.error, AttributeError):
# The socket is already disconnected.
pass
- def _connect(self):
+ def _connect(self) -> None:
"""Connects to the defined server.
If login/pass was specified, the class tries to authenticate.
An error is raised if something goes wrong.
"""
try:
- self._srv_handler = telnetlib.Telnet(self._host, self._port,
- timeout=self._timeout)
-
+ self._srv_handler = Telnet(self._host, self._port, timeout=self._timeout)
+
+ result: str
if self._login is not None:
- self._write("USERNAME %s\n" % self._login)
- result = self._read_until("\n")
- if not result == "OK\n":
- raise PyNUTError(result.replace("\n", ""))
+ self._write(f'USERNAME {self._login}\n')
+ result = self._read_until('\n')
+ if result != 'OK\n':
+ raise PyNUTError(result.replace('\n', ''))
if self._password is not None:
- self._write("PASSWORD %s\n" % self._password)
- result = self._read_until("\n")
- if not result == "OK\n":
- raise PyNUTError(result.replace("\n", ""))
- except telnetlib.socket.error:
- raise PyNUTError("Socket error.")
-
- def _read_until(self, string):
+ self._write(f'PASSWORD {self._password}\n')
+ result = self._read_until('\n')
+ if result != 'OK\n':
+ raise PyNUTError(result.replace('\n', ''))
+ except socket.error:
+ raise PyNUTError('Socket error.')
+
+ def _read_until(self, string: str) -> str:
""" Wrapper for _srv_handler read_until method."""
try:
- return self._srv_handler.read_until(string.encode('ascii'),
- self._timeout).decode()
+ if self._srv_handler is None:
+ raise RuntimeError('NUT2 connection has not been opened.')
+ return self._srv_handler.read_until(string.encode('ascii'), self._timeout).decode()
except (EOFError, BrokenPipeError):
- _LOGGER.error("NUT2 problem reading from server.")
+ _LOGGER.error('NUT2 problem reading from server.')
+ return ''
- def _write(self, string):
+ def _write(self, string: str) -> None:
""" Wrapper for _srv_handler write method."""
try:
+ if self._srv_handler is None:
+ raise RuntimeError('NUT2 connection has not been opened.')
return self._srv_handler.write(string.encode('ascii'))
except (EOFError, BrokenPipeError):
- _LOGGER.error("NUT2 problem writing to server.")
+ _LOGGER.error('NUT2 problem writing to server.')
- def description(self, ups):
+ def description(self, ups: str) -> str:
"""Returns the description for a given UPS."""
- _LOGGER.debug("NUT2 requesting description from server %s", self._host)
+ _LOGGER.debug('NUT2 requesting description from server %s', self._host)
if not self._persistent:
self._connect()
- self._write("GET UPSDESC %s\n" % ups)
- result = self._read_until("\n")
+ self._write(f'GET UPSDESC {ups}\n')
+ result: str = self._read_until('\n')
if not self._persistent:
self._disconnect()
@@ -148,30 +153,36 @@ def description(self, ups):
try:
return result.split('"')[1].strip()
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def list_ups(self):
+ def list_ups(self) -> Dict[str, str]:
"""Returns the list of available UPS from the NUT server.
The result is a dictionary containing 'key->val' pairs of
'UPSName' and 'UPS Description'.
"""
- _LOGGER.debug("NUT2 requesting list_ups from server %s", self._host)
+ _LOGGER.debug('NUT2 requesting list_ups from server %s', self._host)
if not self._persistent:
self._connect()
- self._write("LIST UPS\n")
- result = self._read_until("\n")
- if result != "BEGIN LIST UPS\n":
- raise PyNUTError(result.replace("\n", ""))
-
- result = self._read_until("END LIST UPS\n")
-
- ups_dict = {}
- for line in result.split("\n"):
- if line.startswith("UPS"):
- ups, desc = line[len("UPS "):-len('"')].split('"')[:2]
+ self._write('LIST UPS\n')
+ result: str = self._read_until('\n')
+ if result != 'BEGIN LIST UPS\n':
+ raise PyNUTError(result.replace('\n', ''))
+
+ result = self._read_until('END LIST UPS\n')
+
+ ups_dict: Dict[str, str] = {}
+ line: str
+ for line in result.split('\n'):
+ if line.startswith('UPS'):
+ line = line[len('UPS '):-len('"')]
+ if '"' not in line:
+ continue
+ ups: str
+ desc: str
+ ups, desc = line.split('"')[:2]
ups_dict[ups.strip()] = desc.strip()
if not self._persistent:
@@ -179,7 +190,7 @@ def list_ups(self):
return ups_dict
- def list_vars(self, ups):
+ def list_vars(self, ups: str) -> Dict[str, str]:
"""Get all available vars from the specified UPS.
The result is a dictionary containing 'key->val' pairs of all
@@ -190,18 +201,24 @@ def list_vars(self, ups):
if not self._persistent:
self._connect()
- self._write("LIST VAR %s\n" % ups)
- result = self._read_until("\n")
- if result != "BEGIN LIST VAR %s\n" % ups:
- raise PyNUTError(result.replace("\n", ""))
-
- result = self._read_until("END LIST VAR %s\n" % ups)
- offset = len("VAR %s " % ups)
- end_offset = 0 - (len("END LIST VAR %s\n" % ups) + 1)
-
- ups_vars = {}
- for current in result[:end_offset].split("\n"):
- var, data = current[offset:].split('"')[:2]
+ self._write(f'LIST VAR {ups}\n')
+ result: str = self._read_until('\n')
+ if result != f'BEGIN LIST VAR {ups}\n':
+ raise PyNUTError(result.replace('\n', ''))
+
+ result = self._read_until(f'END LIST VAR {ups}\n')
+ offset: int = len(f'VAR {ups} ')
+ end_offset: int = 0 - (len(f'END LIST VAR {ups}\n') + 1)
+
+ ups_vars: Dict[str, str] = {}
+ current: str
+ for current in result[:end_offset].split('\n'):
+ current = current[offset:]
+ if '"' not in current:
+ continue
+ var: str
+ data: str
+ var, data = current.split('"')[:2]
ups_vars[var.strip()] = data
if not self._persistent:
@@ -209,7 +226,7 @@ def list_vars(self, ups):
return ups_vars
- def list_commands(self, ups):
+ def list_commands(self, ups: str) -> Dict[str, str]:
"""Get all available commands for the specified UPS.
The result is a dict object with command name as key and a description
@@ -221,26 +238,30 @@ def list_commands(self, ups):
if not self._persistent:
self._connect()
- self._write("LIST CMD %s\n" % ups)
- result = self._read_until("\n")
- if result != "BEGIN LIST CMD %s\n" % ups:
- raise PyNUTError(result.replace("\n", ""))
+ self._write(f'LIST CMD {ups}\n')
+ result: str = self._read_until('\n')
+ if result != f'BEGIN LIST CMD {ups}\n':
+ raise PyNUTError(result.replace('\n', ''))
- result = self._read_until("END LIST CMD %s\n" % ups)
- offset = len("CMD %s " % ups)
- end_offset = 0 - (len("END LIST CMD %s\n" % ups) + 1)
+ result = self._read_until(f'END LIST CMD {ups}\n')
+ offset: int = len(f'CMD {ups} ')
+ end_offset: int = 0 - (len(f'END LIST CMD {ups}\n') + 1)
- commands = {}
- for current in result[:end_offset].split("\n"):
- command = current[offset:].split('"')[0].strip()
+ commands: Dict[str, str] = {}
+ current: str
+ for current in result[:end_offset].split('\n'):
+ command: str = current[offset:].split('"')[0].strip()
# For each var we try to get the available description
try:
- self._write("GET CMDDESC %s %s\n" % (ups, command))
- temp = self._read_until("\n")
- if temp.startswith("CMDDESC"):
- desc_offset = len("CMDDESC %s %s " % (ups, command))
- commands[command] = temp[desc_offset:-1].split('"')[1]
+ self._write(f'GET CMDDESC {ups} {command}\n')
+ temp: str = self._read_until('\n')
+ if temp.startswith('CMDDESC'):
+ desc_offset = len(f'CMDDESC {ups} {command} ')
+ temp = temp[desc_offset:-1]
+ if '"' not in temp:
+ continue
+ commands[command] = temp.split('"')[1]
else:
commands[command] = command
except IndexError:
@@ -251,7 +272,7 @@ def list_commands(self, ups):
return commands
- def list_clients(self, ups=None):
+ def list_clients(self, ups: str = '') -> Dict[str, List[str]]:
"""Returns the list of connected clients from the NUT server.
The result is a dictionary containing 'key->val' pairs of
@@ -264,22 +285,27 @@ def list_clients(self, ups=None):
self._connect()
if ups and (ups not in self.list_ups()):
- raise PyNUTError("%s is not a valid UPS" % ups)
+ raise PyNUTError(f'{ups} is not a valid UPS')
if ups:
- self._write("LIST CLIENTS %s\n" % ups)
+ self._write(f'LIST CLIENTS {ups}\n')
else:
- self._write("LIST CLIENTS\n")
- result = self._read_until("\n")
- if result != "BEGIN LIST CLIENTS\n":
- raise PyNUTError(result.replace("\n", ""))
-
- result = self._read_until("END LIST CLIENTS\n")
-
- clients = {}
- for line in result.split("\n"):
- if line.startswith("CLIENT"):
- host, ups = line[len("CLIENT "):].split(' ')[:2]
+ self._write('LIST CLIENTS\n')
+ result = self._read_until('\n')
+ if result != 'BEGIN LIST CLIENTS\n':
+ raise PyNUTError(result.replace('\n', ''))
+
+ result = self._read_until('END LIST CLIENTS\n')
+
+ clients: Dict[str, List[str]] = {}
+ line: str
+ for line in result.split('\n'):
+ if line.startswith('CLIENT') and ' ' in line[len('CLIENT '):]:
+ line = line[len('CLIENT '):]
+ if ' ' not in line:
+ continue
+ host: str
+ host, ups = line.split(' ')[:2]
if ups not in clients:
clients[ups] = []
clients[ups].append(host)
@@ -289,30 +315,34 @@ def list_clients(self, ups=None):
return clients
- def list_rw_vars(self, ups):
+ def list_rw_vars(self, ups: str) -> Dict[str, str]:
"""Get a list of all writable vars from the selected UPS.
The result is presented as a dictionary containing 'key->val'
pairs.
"""
- _LOGGER.debug("NUT2 requesting list_rw_vars from server %s",
- self._host)
+ _LOGGER.debug("NUT2 requesting list_rw_vars from server %s", self._host)
if not self._persistent:
self._connect()
- self._write("LIST RW %s\n" % ups)
- result = self._read_until("\n")
- if result != "BEGIN LIST RW %s\n" % ups:
- raise PyNUTError(result.replace("\n", ""))
-
- result = self._read_until("END LIST RW %s\n" % ups)
- offset = len("VAR %s" % ups)
- end_offset = 0 - (len("END LIST RW %s\n" % ups) + 1)
-
- rw_vars = {}
- for current in result[:end_offset].split("\n"):
- var, data = current[offset:].split('"')[:2]
+ self._write(f'LIST RW {ups}\n')
+ result: str = self._read_until('\n')
+ if result != f'BEGIN LIST RW {ups}\n':
+ raise PyNUTError(result.replace('\n', ''))
+
+ result = self._read_until(f'END LIST RW {ups}\n')
+ offset: int = len(f'VAR {ups}')
+ end_offset: int = 0 - (len(f'END LIST RW {ups}\n') + 1)
+
+ rw_vars: Dict[str, str] = {}
+ for current in result[:end_offset].split('\n'):
+ current = current[offset:]
+ if '"' not in current:
+ continue
+ var: str
+ data: str
+ var, data = current.split('"')[:2]
rw_vars[var.strip()] = data
if not self._persistent:
@@ -320,213 +350,203 @@ def list_rw_vars(self, ups):
return rw_vars
- def list_enum(self, ups, var):
+ def list_enum(self, ups: str, var: str) -> List[str]:
"""Get a list of valid values for an enum variable.
The result is presented as a list.
"""
- _LOGGER.debug("NUT2 requesting list_enum from server %s",
- self._host)
+ _LOGGER.debug("NUT2 requesting list_enum from server %s", self._host)
if not self._persistent:
self._connect()
- self._write("LIST ENUM %s %s\n" % (ups, var))
- result = self._read_until("\n")
- if result != "BEGIN LIST ENUM %s %s\n" % (ups, var):
- raise PyNUTError(result.replace("\n", ""))
+ self._write(f'LIST ENUM {ups} {var}\n')
+ result: str = self._read_until('\n')
+ if result != f'BEGIN LIST ENUM {ups} {var}\n':
+ raise PyNUTError(result.replace('\n', ''))
- result = self._read_until("END LIST ENUM %s %s\n" % (ups, var))
- offset = len("ENUM %s %s" % (ups, var))
- end_offset = 0 - (len("END LIST ENUM %s %s\n" % (ups, var)) + 1)
+ result = self._read_until(f'END LIST ENUM {ups} {var}\n')
+ offset: int = len(f'ENUM {ups} {var}')
+ end_offset: int = 0 - (len(f'END LIST ENUM {ups} {var}\n') + 1)
if not self._persistent:
self._disconnect()
try:
return [c[offset:].split('"')[1].strip()
- for c in result[:end_offset].split("\n")]
+ for c in result[:end_offset].split('\n')
+ if '"' in c[offset:]]
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def list_range(self, ups, var):
+ def list_range(self, ups: str, var: str) -> List[str]:
"""Get a list of valid values for an range variable.
The result is presented as a list.
"""
- _LOGGER.debug("NUT2 requesting list_range from server %s",
- self._host)
+ _LOGGER.debug("NUT2 requesting list_range from server %s", self._host)
if not self._persistent:
self._connect()
- self._write("LIST RANGE %s %s\n" % (ups, var))
- result = self._read_until("\n")
- if result != "BEGIN LIST RANGE %s %s\n" % (ups, var):
- raise PyNUTError(result.replace("\n", ""))
+ self._write(f'LIST RANGE {ups} {var}\n')
+ result: str = self._read_until('\n')
+ if result != f'BEGIN LIST RANGE {ups} {var}\n':
+ raise PyNUTError(result.replace('\n', ''))
- result = self._read_until("END LIST RANGE %s %s\n" % (ups, var))
- offset = len("RANGE %s %s" % (ups, var))
- end_offset = 0 - (len("END LIST RANGE %s %s\n" % (ups, var)) + 1)
+ result = self._read_until(f'END LIST RANGE {ups} {var}\n')
+ offset: int = len(f'RANGE {ups} {var}')
+ end_offset: int = 0 - (len(f'END LIST RANGE {ups} {var}\n') + 1)
if not self._persistent:
self._disconnect()
try:
return [c[offset:].split('"')[1].strip()
- for c in result[:end_offset].split("\n")]
+ for c in result[:end_offset].split('\n')
+ if '"' in c[offset:]]
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def set_var(self, ups, var, value):
+ def set_var(self, ups: str, var: str, value: str) -> None:
"""Set a variable to the specified value on selected UPS.
The variable must be a writable value (cf list_rw_vars) and you
must have the proper rights to set it (maybe login/password).
"""
- _LOGGER.debug("NUT2 setting set_var '%s' on '%s' to '%s'",
- var, self._host, value)
+ _LOGGER.debug("NUT2 setting set_var '%s' on '%s' to '%s'", var, self._host, value)
if not self._persistent:
self._connect()
- self._write("SET VAR %s %s %s\n" % (ups, var, value))
- result = self._read_until("\n")
+ self._write(f'SET VAR {ups} {var} {value}\n')
+ result = self._read_until('\n')
- if result != "OK\n":
- raise PyNUTError(result.replace("\n", ""))
+ if result != 'OK\n':
+ raise PyNUTError(result.replace('\n', ''))
if not self._persistent:
self._disconnect()
- def get_var(self, ups, var):
+ def get_var(self, ups: str, var: str) -> str:
"""Get the value of a variable."""
- _LOGGER.debug("NUT2 requesting get_var '%s' on '%s'.",
- var, self._host)
+ _LOGGER.debug("NUT2 requesting get_var '%s' on '%s'.", var, self._host)
if not self._persistent:
self._connect()
- self._write("GET VAR %s %s\n" % (ups, var))
- result = self._read_until("\n")
+ self._write(f'GET VAR {ups} {var}\n')
+ result = self._read_until('\n')
if not self._persistent:
self._disconnect()
try:
- # result = 'VAR %s %s "%s"\n' % (ups, var, value)
return result.split('"')[1].strip()
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
# Alias for convenience
- def get(self, ups, var):
+ def get(self, ups: str, var: str) -> str:
"""Get the value of a variable (alias for get_var)."""
return self.get_var(ups, var)
- def var_description(self, ups, var):
+ def var_description(self, ups: str, var: str) -> str:
"""Get a variable's description."""
- _LOGGER.debug("NUT2 requesting var_description '%s' on '%s'.",
- var, self._host)
+ _LOGGER.debug("NUT2 requesting var_description '%s' on '%s'.", var, self._host)
if not self._persistent:
self._connect()
- self._write("GET DESC %s %s\n" % (ups, var))
- result = self._read_until("\n")
+ self._write(f'GET DESC {ups} {var}\n')
+ result = self._read_until('\n')
if not self._persistent:
self._disconnect()
try:
- # result = 'DESC %s %s "%s"\n' % (ups, var, description)
return result.split('"')[1].strip()
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def var_type(self, ups, var):
+ def var_type(self, ups: str, var: str) -> str:
"""Get a variable's type."""
- _LOGGER.debug("NUT2 requesting var_type '%s' on '%s'.",
- var, self._host)
+ _LOGGER.debug("NUT2 requesting var_type '%s' on '%s'.", var, self._host)
if not self._persistent:
self._connect()
- self._write("GET TYPE %s %s\n" % (ups, var))
- result = self._read_until("\n")
+ self._write(f'GET TYPE {ups} {var}\n')
+ result: str = self._read_until('\n')
if not self._persistent:
self._disconnect()
try:
- # result = 'TYPE %s %s %s\n' % (ups, var, type)
- type_ = ' '.join(result.split(' ')[3:]).strip()
+ type_: str = ' '.join(result.split(' ')[3:]).strip()
# Ensure the response was valid.
assert len(type_) > 0
- assert result.startswith("TYPE")
+ assert result.startswith('TYPE')
return type_
except AssertionError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def command_description(self, ups, command):
+ def command_description(self, ups: str, command: str) -> str:
"""Get a command's description."""
- _LOGGER.debug("NUT2 requesting command_description '%s' on '%s'.",
- command, self._host)
+ _LOGGER.debug("NUT2 requesting command_description '%s' on '%s'.", command, self._host)
if not self._persistent:
self._connect()
- self._write("GET CMDDESC %s %s\n" % (ups, command))
- result = self._read_until("\n")
+ self._write(f'GET CMDDESC {ups} {command}\n')
+ result: str = self._read_until('\n')
if not self._persistent:
self._disconnect()
try:
- # result = 'CMDDESC %s %s "%s"' % (ups, command, description)
return result.split('"')[1].strip()
except IndexError:
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def run_command(self, ups, command):
+ def run_command(self, ups: str, command: str) -> None:
"""Send a command to the specified UPS."""
- _LOGGER.debug("NUT2 run_command called '%s' on '%s'.",
- command, self._host)
+ _LOGGER.debug("NUT2 run_command called '%s' on '%s'.", command, self._host)
if not self._persistent:
self._connect()
- self._write("INSTCMD %s %s\n" % (ups, command))
- result = self._read_until("\n")
+ self._write(f'INSTCMD {ups} {command}\n')
+ result: str = self._read_until('\n')
- if result != "OK\n":
- raise PyNUTError(result.replace("\n", ""))
+ if result != 'OK\n':
+ raise PyNUTError(result.replace('\n', ''))
if not self._persistent:
self._disconnect()
- def fsd(self, ups):
+ def fsd(self, ups: str) -> None:
"""Send MASTER and FSD commands."""
_LOGGER.debug("NUT2 MASTER called on '%s'.", self._host)
if not self._persistent:
self._connect()
- self._write("MASTER %s\n" % ups)
- result = self._read_until("\n")
- if result != "OK MASTER-GRANTED\n":
- raise PyNUTError(("Master level function are not available", ""))
+ self._write(f'MASTER {ups}\n')
+ result: str = self._read_until('\n')
+ if result != 'OK MASTER-GRANTED\n':
+ raise PyNUTError(('Master level function are not available', ''))
- _LOGGER.debug("FSD called...")
- self._write("FSD %s\n" % ups)
- result = self._read_until("\n")
- if result != "OK FSD-SET\n":
- raise PyNUTError(result.replace("\n", ""))
+ _LOGGER.debug('FSD called...')
+ self._write(f'FSD {ups}\n')
+ result = self._read_until('\n')
+ if result != 'OK FSD-SET\n':
+ raise PyNUTError(result.replace('\n', ''))
if not self._persistent:
self._disconnect()
- def num_logins(self, ups):
+ def num_logins(self, ups: str) -> int:
"""Send GET NUMLOGINS command to get the number of users logged
into a given UPS.
"""
@@ -535,42 +555,41 @@ def num_logins(self, ups):
if not self._persistent:
self._connect()
- self._write("GET NUMLOGINS %s\n" % ups)
- result = self._read_until("\n")
+ self._write(f'GET NUMLOGINS {ups}\n')
+ result: str = self._read_until('\n')
if not self._persistent:
self._disconnect()
try:
- # result = "NUMLOGINS %s %s\n" % (ups, int(numlogins))
return int(result.split(' ')[2].strip())
except (ValueError, IndexError):
- raise PyNUTError(result.replace("\n", ""))
+ raise PyNUTError(result.replace('\n', ''))
- def help(self):
+ def help(self) -> str:
"""Send HELP command."""
_LOGGER.debug("NUT2 HELP called on '%s'", self._host)
if not self._persistent:
self._connect()
- self._write("HELP\n")
+ self._write('HELP\n')
if not self._persistent:
self._disconnect()
- return self._read_until("\n")
+ return self._read_until('\n')
- def ver(self):
+ def ver(self) -> str:
"""Send VER command."""
_LOGGER.debug("NUT2 VER called on '%s'", self._host)
if not self._persistent:
self._connect()
- self._write("VER\n")
+ self._write('VER\n')
if not self._persistent:
self._disconnect()
- return self._read_until("\n")
+ return self._read_until('\n')