new upstream 2.8.0
This commit is contained in:
lagertonne
parent
fc7f4b43c1
commit
b2b0c9995a
836 changed files with 137090 additions and 30018 deletions
scripts/python/module
|
|
@ -1,4 +1,4 @@
|
|||
#!/usr/bin/env python
|
||||
#!@PYTHON@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright (C) 2008 David Goncalves <david@lestat.st>
|
||||
|
|
@ -38,6 +38,11 @@
|
|||
# 2014-06-03 george2 - Version 1.3.0
|
||||
# Added custom exception class, fixed minor bug, added Python 3 support.
|
||||
#
|
||||
# 2021-09-27 Jim Klimov <jimklimov+nut@gmail.com> - Version 1.4.0
|
||||
# Revise strings used to be byte sequences as required by telnetlib
|
||||
# in Python 3.9, by spelling out b"STR" or str.encode('ascii');
|
||||
# the change was also tested to work with Python 2.7, 3.4, 3.5 and
|
||||
# 3.7 (to the extent of accompanying test_nutclient.py at least).
|
||||
|
||||
import telnetlib
|
||||
|
||||
|
|
@ -56,8 +61,8 @@ class PyNUTClient :
|
|||
__timeout = None
|
||||
__srv_handler = None
|
||||
|
||||
__version = "1.3.0"
|
||||
__release = "2014-06-03"
|
||||
__version = "1.4.0"
|
||||
__release = "2021-09-27"
|
||||
|
||||
|
||||
def __init__( self, host="127.0.0.1", port=3493, login=None, password=None, debug=False, timeout=5 ) :
|
||||
|
|
@ -89,7 +94,7 @@ timeout : Timeout used to wait for network response
|
|||
def __del__( self ) :
|
||||
""" Class destructor method """
|
||||
try :
|
||||
self.__srv_handler.write( "LOGOUT\n" )
|
||||
self.__srv_handler.write( b"LOGOUT\n" )
|
||||
except :
|
||||
pass
|
||||
|
||||
|
|
@ -105,16 +110,16 @@ if something goes wrong.
|
|||
self.__srv_handler = telnetlib.Telnet( self.__host, self.__port )
|
||||
|
||||
if self.__login != None :
|
||||
self.__srv_handler.write( "USERNAME %s\n" % self.__login )
|
||||
result = self.__srv_handler.read_until( "\n", self.__timeout )
|
||||
if result[:2] != "OK" :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( ("USERNAME %s\n" % self.__login).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n", self.__timeout )
|
||||
if result[:2] != b"OK" :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
if self.__password != None :
|
||||
self.__srv_handler.write( "PASSWORD %s\n" % self.__password )
|
||||
result = self.__srv_handler.read_until( "\n", self.__timeout )
|
||||
if result[:2] != "OK" :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( ("PASSWORD %s\n" % self.__password).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n", self.__timeout )
|
||||
if result[:2] != b"OK" :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
def GetUPSList( self ) :
|
||||
""" Returns the list of available UPS from the NUT server
|
||||
|
|
@ -124,18 +129,18 @@ The result is a dictionary containing 'key->val' pairs of 'UPSName' and 'UPS Des
|
|||
if self.__debug :
|
||||
print( "[DEBUG] GetUPSList from server" )
|
||||
|
||||
self.__srv_handler.write( "LIST UPS\n" )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if result != "BEGIN LIST UPS\n" :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( b"LIST UPS\n" )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if result != b"BEGIN LIST UPS\n" :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
result = self.__srv_handler.read_until( "END LIST UPS\n" )
|
||||
result = self.__srv_handler.read_until( b"END LIST UPS\n" )
|
||||
ups_list = {}
|
||||
|
||||
for line in result.split( "\n" ) :
|
||||
if line[:3] == "UPS" :
|
||||
ups, desc = line[4:-1].split( '"' )
|
||||
ups_list[ ups.replace( " ", "" ) ] = desc
|
||||
for line in result.split( b"\n" ) :
|
||||
if line[:3] == b"UPS" :
|
||||
ups, desc = line[4:-1].split( b'"' )
|
||||
ups_list[ ups.replace( b" ", b"" ) ] = desc
|
||||
|
||||
return( ups_list )
|
||||
|
||||
|
|
@ -148,19 +153,19 @@ available vars.
|
|||
if self.__debug :
|
||||
print( "[DEBUG] GetUPSVars called..." )
|
||||
|
||||
self.__srv_handler.write( "LIST VAR %s\n" % ups )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if result != "BEGIN LIST VAR %s\n" % ups :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( ("LIST VAR %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if result != ("BEGIN LIST VAR %s\n" % ups).encode('ascii') :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
ups_vars = {}
|
||||
result = self.__srv_handler.read_until( "END LIST VAR %s\n" % ups )
|
||||
offset = len( "VAR %s " % ups )
|
||||
end_offset = 0 - ( len( "END LIST VAR %s\n" % ups ) + 1 )
|
||||
result = self.__srv_handler.read_until( ("END LIST VAR %s\n" % ups).encode('ascii') )
|
||||
offset = len( ("VAR %s " % ups ).encode('ascii') )
|
||||
end_offset = 0 - ( len( ("END LIST VAR %s\n" % ups).encode('ascii') ) + 1 )
|
||||
|
||||
for current in result[:end_offset].split( "\n" ) :
|
||||
var = current[ offset: ].split( '"' )[0].replace( " ", "" )
|
||||
data = current[ offset: ].split( '"' )[1]
|
||||
for current in result[:end_offset].split( b"\n" ) :
|
||||
var = current[ offset: ].split( b'"' )[0].replace( b" ", b"" )
|
||||
data = current[ offset: ].split( b'"' )[1]
|
||||
ups_vars[ var ] = data
|
||||
|
||||
return( ups_vars )
|
||||
|
|
@ -174,28 +179,28 @@ of the command as value
|
|||
if self.__debug :
|
||||
print( "[DEBUG] GetUPSCommands called..." )
|
||||
|
||||
self.__srv_handler.write( "LIST CMD %s\n" % ups )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if result != "BEGIN LIST CMD %s\n" % ups :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( ("LIST CMD %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if result != ("BEGIN LIST CMD %s\n" % ups).encode('ascii') :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
ups_cmds = {}
|
||||
result = self.__srv_handler.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.__srv_handler.read_until( ("END LIST CMD %s\n" % ups).encode('ascii') )
|
||||
offset = len( ("CMD %s " % ups).encode('ascii') )
|
||||
end_offset = 0 - ( len( ("END LIST CMD %s\n" % ups).encode('ascii') ) + 1 )
|
||||
|
||||
for current in result[:end_offset].split( "\n" ) :
|
||||
var = current[ offset: ].split( '"' )[0].replace( " ", "" )
|
||||
for current in result[:end_offset].split( b"\n" ) :
|
||||
var = current[ offset: ].split( b'"' )[0].replace( b" ", b"" )
|
||||
|
||||
# For each var we try to get the available description
|
||||
try :
|
||||
self.__srv_handler.write( "GET CMDDESC %s %s\n" % ( ups, var ) )
|
||||
temp = self.__srv_handler.read_until( "\n" )
|
||||
if temp[:7] != "CMDDESC" :
|
||||
self.__srv_handler.write( ("GET CMDDESC %s %s\n" % ( ups, var )).encode('ascii') )
|
||||
temp = self.__srv_handler.read_until( b"\n" )
|
||||
if temp[:7] != b"CMDDESC" :
|
||||
raise PyNUTError
|
||||
else :
|
||||
off = len( "CMDDESC %s %s " % ( ups, var ) )
|
||||
desc = temp[off:-1].split('"')[1]
|
||||
off = len( ("CMDDESC %s %s " % ( ups, var )).encode('ascii') )
|
||||
desc = temp[off:-1].split(b'"')[1]
|
||||
except :
|
||||
desc = var
|
||||
|
||||
|
|
@ -211,20 +216,20 @@ The result is presented as a dictionary containing 'key->val' pairs
|
|||
if self.__debug :
|
||||
print( "[DEBUG] GetUPSVars from '%s'..." % ups )
|
||||
|
||||
self.__srv_handler.write( "LIST RW %s\n" % ups )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if ( result != "BEGIN LIST RW %s\n" % ups ) :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( ("LIST RW %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result != ("BEGIN LIST RW %s\n" % ups).encode('ascii') ) :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
result = self.__srv_handler.read_until( "END LIST RW %s\n" % ups )
|
||||
offset = len( "VAR %s" % ups )
|
||||
end_offset = 0 - ( len( "END LIST RW %s\n" % ups ) + 1 )
|
||||
result = self.__srv_handler.read_until( ("END LIST RW %s\n" % ups).encode('ascii') )
|
||||
offset = len( ("VAR %s" % ups).encode('ascii') )
|
||||
end_offset = 0 - ( len( ("END LIST RW %s\n" % ups).encode('ascii') ) + 1 )
|
||||
rw_vars = {}
|
||||
|
||||
try :
|
||||
for current in result[:end_offset].split( "\n" ) :
|
||||
var = current[ offset: ].split( '"' )[0].replace( " ", "" )
|
||||
data = current[ offset: ].split( '"' )[1]
|
||||
for current in result[:end_offset].split( b"\n" ) :
|
||||
var = current[ offset: ].split( b'"' )[0].replace( b" ", b"" )
|
||||
data = current[ offset: ].split( b'"' )[1]
|
||||
rw_vars[ var ] = data
|
||||
|
||||
except :
|
||||
|
|
@ -239,9 +244,9 @@ The variable must be a writable value (cf GetRWVars) and you must have the prope
|
|||
rights to set it (maybe login/password).
|
||||
"""
|
||||
|
||||
self.__srv_handler.write( "SET VAR %s %s %s\n" % ( ups, var, value ) )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if ( result == "OK\n" ) :
|
||||
self.__srv_handler.write( ("SET VAR %s %s %s\n" % ( ups, var, value )).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result == b"OK\n" ) :
|
||||
return( "OK" )
|
||||
else :
|
||||
raise PyNUTError( result )
|
||||
|
|
@ -255,35 +260,43 @@ Returns OK on success or raises an error
|
|||
if self.__debug :
|
||||
print( "[DEBUG] RunUPSCommand called..." )
|
||||
|
||||
self.__srv_handler.write( "INSTCMD %s %s\n" % ( ups, command ) )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if ( result == "OK\n" ) :
|
||||
self.__srv_handler.write( ("INSTCMD %s %s\n" % ( ups, command )).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result == b"OK\n" ) :
|
||||
return( "OK" )
|
||||
else :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
def FSD( self, ups="") :
|
||||
""" Send FSD command
|
||||
|
||||
Returns OK on success or raises an error
|
||||
|
||||
NOTE: API changed since NUT 2.8.0 to replace MASTER with PRIMARY
|
||||
(and backwards-compatible alias handling)
|
||||
"""
|
||||
|
||||
if self.__debug :
|
||||
print( "[DEBUG] MASTER called..." )
|
||||
print( "[DEBUG] PRIMARY called..." )
|
||||
|
||||
self.__srv_handler.write( "MASTER %s\n" % ups )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if ( result != "OK MASTER-GRANTED\n" ) :
|
||||
raise PyNUTError( ( "Master level function are not available", "" ) )
|
||||
self.__srv_handler.write( ("PRIMARY %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result != b"OK PRIMARY-GRANTED\n" ) :
|
||||
if self.__debug :
|
||||
print( "[DEBUG] Retrying: MASTER called..." )
|
||||
self.__srv_handler.write( ("MASTER %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result != b"OK MASTER-GRANTED\n" ) :
|
||||
raise PyNUTError( ( "Primary level functions are not available", "" ) )
|
||||
|
||||
if self.__debug :
|
||||
print( "[DEBUG] FSD called..." )
|
||||
self.__srv_handler.write( "FSD %s\n" % ups )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if ( result == "OK FSD-SET\n" ) :
|
||||
self.__srv_handler.write( ("FSD %s\n" % ups).encode('ascii') )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if ( result == b"OK FSD-SET\n" ) :
|
||||
return( "OK" )
|
||||
else :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
def help(self) :
|
||||
""" Send HELP command
|
||||
|
|
@ -292,8 +305,8 @@ Returns OK on success or raises an error
|
|||
if self.__debug :
|
||||
print( "[DEBUG] HELP called..." )
|
||||
|
||||
self.__srv_handler.write( "HELP\n")
|
||||
return self.__srv_handler.read_until( "\n" )
|
||||
self.__srv_handler.write( b"HELP\n" )
|
||||
return self.__srv_handler.read_until( b"\n" )
|
||||
|
||||
def ver(self) :
|
||||
""" Send VER command
|
||||
|
|
@ -302,8 +315,8 @@ Returns OK on success or raises an error
|
|||
if self.__debug :
|
||||
print( "[DEBUG] VER called..." )
|
||||
|
||||
self.__srv_handler.write( "VER\n")
|
||||
return self.__srv_handler.read_until( "\n" )
|
||||
self.__srv_handler.write( b"VER\n" )
|
||||
return self.__srv_handler.read_until( b"\n" )
|
||||
|
||||
def ListClients( self, ups = None ) :
|
||||
""" Returns the list of connected clients from the NUT server
|
||||
|
|
@ -317,20 +330,20 @@ The result is a dictionary containing 'key->val' pairs of 'UPSName' and a list o
|
|||
raise PyNUTError( "%s is not a valid UPS" % ups )
|
||||
|
||||
if ups:
|
||||
self.__srv_handler.write( "LIST CLIENTS %s\n" % ups)
|
||||
self.__srv_handler.write( ("LIST CLIENTS %s\n" % ups).encode('ascii') )
|
||||
else:
|
||||
self.__srv_handler.write( "LIST CLIENTS\n" )
|
||||
result = self.__srv_handler.read_until( "\n" )
|
||||
if result != "BEGIN LIST CLIENTS\n" :
|
||||
raise PyNUTError( result.replace( "\n", "" ) )
|
||||
self.__srv_handler.write( b"LIST CLIENTS\n" )
|
||||
result = self.__srv_handler.read_until( b"\n" )
|
||||
if result != b"BEGIN LIST CLIENTS\n" :
|
||||
raise PyNUTError( result.replace( b"\n", b"" ) )
|
||||
|
||||
result = self.__srv_handler.read_until( "END LIST CLIENTS\n" )
|
||||
result = self.__srv_handler.read_until( b"END LIST CLIENTS\n" )
|
||||
ups_list = {}
|
||||
|
||||
for line in result.split( "\n" ):
|
||||
if line[:6] == "CLIENT" :
|
||||
host, ups = line[7:].split(' ')
|
||||
ups.replace(' ', '')
|
||||
for line in result.split( b"\n" ):
|
||||
if line[:6] == b"CLIENT" :
|
||||
host, ups = line[7:].split(b' ')
|
||||
ups.replace(b' ', b'')
|
||||
if not ups in ups_list:
|
||||
ups_list[ups] = []
|
||||
ups_list[ups].append(host)
|
||||
|
|
@ -1,22 +1,31 @@
|
|||
#!/usr/bin/env python
|
||||
#!@PYTHON@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# This source code is provided for testing/debuging purpose ;)
|
||||
|
||||
import PyNUT
|
||||
import sys
|
||||
import os
|
||||
|
||||
if __name__ == "__main__" :
|
||||
NUT_PORT = int(os.getenv('NUT_PORT', '3493'))
|
||||
NUT_USER = os.getenv('NUT_USER', None)
|
||||
NUT_PASS = os.getenv('NUT_PASS', None)
|
||||
|
||||
print( "PyNUTClient test..." )
|
||||
nut = PyNUT.PyNUTClient( debug=True )
|
||||
#nut = PyNUT.PyNUTClient( login="upsadmin", password="upsadmin", debug=True )
|
||||
#nut = PyNUT.PyNUTClient( debug=True, port=NUT_PORT )
|
||||
nut = PyNUT.PyNUTClient( login=NUT_USER, password=NUT_PASS, debug=True, port=NUT_PORT )
|
||||
#nut = PyNUT.PyNUTClient( login="upsadmin", password="upsadmin", debug=True, port=NUT_PORT )
|
||||
|
||||
print( 80*"-" + "\nTesting 'GetUPSList' :")
|
||||
result = nut.GetUPSList( )
|
||||
print( "\033[01;33m%s\033[0m\n" % result )
|
||||
|
||||
print( 80*"-" + "\nTesting 'GetUPSVars' :")
|
||||
# [dummy]
|
||||
# driver = dummy-ups
|
||||
# desc = "Test device"
|
||||
# port = /src/nut/data/evolution500.seq
|
||||
print( 80*"-" + "\nTesting 'GetUPSVars' for 'dummy' (should be registered in upsd.conf) :")
|
||||
result = nut.GetUPSVars( "dummy" )
|
||||
print( "\033[01;33m%s\033[0m\n" % result )
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue