15356518725

Committed 30 May 2025 09:59PM UTC coverage: 38.156% (-0.01%) from 38.167%

Build # 15356518725

Build Type

Pull #889

github

Committed by

web-flow

Commit Message

Merge 7e6ad62b8 into 99d37db7a

Pull Request Pull Request #889: hwp-gripper: fix alarm message handling

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

2 existing lines in 1 file now uncovered.

7165 of 18778 relevant lines covered (38.16%)

0.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.79

/socs/agents/ups/agent.py

import argparse
import os
import signal
import time

import txaio
from autobahn.twisted.util import sleep as dsleep
from ocs import ocs_agent, site_config
from ocs.ocs_twisted import TimeoutLock
from twisted.internet.defer import inlineCallbacks

from socs.snmp import SNMPTwister

# For logging
txaio.use_twisted()


def _extract_oid_field_and_value(get_result):
    """Extract field names and OID values from SNMP GET results.

    The ObjectType objects returned from pysnmp interactions contain the
    info we want to use for field names, specifically the OID and associated
    integer for uniquely identifying duplicate OIDs, as well as the value of the
    OID, which we want to save.

    Here we use the prettyPrint() method to get the OID name, requiring
    some string manipulation. We also just grab the hidden ._value
    directly, as this seemed the cleanest way to get an actual value of a
    normal type. Without doing this we get a pysnmp defined Integer32 or
    DisplayString, which were awkward to handle, particularly the
    DisplayString.

    Parameters
    ----------
    get_result : pysnmp.smi.rfc1902.ObjectType
        Result from a pysnmp GET command.

    Returns
    -------
    field_name : str
        Field name for an OID, i.e. 'upsIdentModel'
    oid_value : int or str
        Associated value for the OID. Returns None if not an int or str
    oid_description : str
        String description of the OID value.
    """
    # OID from SNMP GET
    oid = get_result[0].prettyPrint()
    # Makes something like 'UPS-MIB::upsIdentModel.0'
    # look like 'upsIdentModel_0'
    field_name = oid.split("::")[1].replace('.', '_')

    # Grab OID value, mostly these are integers
    oid_value = get_result[1]._value
    oid_description = get_result[1].prettyPrint()

    # Decode string values
    if isinstance(oid_value, bytes):
        oid_value = oid_value.decode("utf-8")

    # I don't expect any other types at the moment, but just in case.
    if not isinstance(oid_value, (int, bytes, str)):
        oid_value = None

    return field_name, oid_value, oid_description


def _build_message(get_result, time, blockname):
    """Build the message for publication on an OCS Feed.

    Parameters
    ----------
    get_result : pysnmp.smi.rfc1902.ObjectType
        Result from a pysnmp GET command.
    time : float
        Timestamp for when the SNMP GET was issued.

    Returns
    -------
    message : dict
        OCS Feed formatted message for publishing
    """
    message = {
        'block_name': blockname,
        'timestamp': time,
        'data': {}
    }

    for item in get_result:
        field_name, oid_value, oid_description = _extract_oid_field_and_value(item)

        if oid_value is None:
            continue

        message['data'][field_name] = oid_value
        message['data'][field_name + "_description"] = oid_description

    return message


def update_cache(get_result, timestamp):
    """Update the OID Value Cache.

    The OID Value Cache is used to store each unique OID and will be passed to
    session.data

    The cache consists of a dictionary, with the unique OIDs as keys, and
    another dictionary as the value. Each of these nested dictionaries contains the
    OID values, name, and description (decoded string). An example for a single OID::

        {"upsBatteryStatus":
            {"status": 2,
                "description": "batteryNormal"}}

    Additionally there is connection status and timestamp information under::

        {"ups_connection":
            {"last_attempt": 1598543359.6326838,
            "connected": True}
        {"timestamp": 1656085022.680916}

    Parameters
    ----------
    get_result : pysnmp.smi.rfc1902.ObjectType
        Result from a pysnmp GET command.
    timestamp : float
        Timestamp for when the SNMP GET was issued.
    """
    oid_cache = {}
    # Return disconnected if SNMP response is empty
    if get_result is None:
        oid_cache['ups_connection'] = {'last_attempt': time.time(),
                                       'connected': False}
        return oid_cache

    for item in get_result:
        field_name, oid_value, oid_description = _extract_oid_field_and_value(item)
        if oid_value is None:
            continue

        # Update OID Cache for session.data
        oid_cache[field_name] = {"status": oid_value}
        oid_cache[field_name]["description"] = oid_description
        oid_cache['ups_connection'] = {'last_attempt': time.time(),
                                       'connected': True}
        oid_cache['timestamp'] = timestamp

    return oid_cache


class UPSAgent:
    """Monitor the UPS system via SNMP.

    Parameters
    ----------
    agent : OCSAgent
        OCSAgent object which forms this Agent
    address : str
        Address of the UPS.
    port : int
        SNMP port to issue GETs to, default to 161.
    version : int
        SNMP version for communication (1, 2, or 3), defaults to 3.

    Attributes
    ----------
    agent : OCSAgent
        OCSAgent object which forms this Agent
    is_streaming : bool
        Tracks whether or not the agent is actively issuing SNMP GET commands
        to the UPS. Setting to false stops sending commands.
    log : txaio.tx.Logger
        txaio logger object, created by the OCSAgent
    """

    def __init__(self, agent, address, port=161, version=1, restart_time=0):
        self.agent = agent
        self.is_streaming = False
        self.log = self.agent.log
        self.lock = TimeoutLock()

        self.log.info(f'Using SNMP version {version}.')
        self.version = version
        self.snmp = SNMPTwister(address, port)
        self.connected = True
        self.restart = restart_time

        self.lastGet = 0

        agg_params = {
            'frame_length': 10 * 60  # [sec]
        }
        self.agent.register_feed('ups',
                                 record=True,
                                 agg_params=agg_params,
                                 buffer_time=0)

    @ocs_agent.param('test_mode', default=False, type=bool)
    @inlineCallbacks
    def acq(self, session, params=None):
        """acq()

        **Process** - Fetch values from the UPS via SNMP.

        Parameters
        ----------
        test_mode : bool, optional
            Run the Process loop only once. Meant only for testing.
            Default is False.

        Notes
        -----
        The most recent data collected is stored in session.data in the
        structure::

            >>> response.session['data']
            {'upsIdentManufacturer':
                {'description': 'Falcon'},
            'upsIdentModel':
                {'description': 'SSG3KRM-2'},
            'upsBatteryStatus':
                {'status': 2,
                 'description': 'batteryNormal'},
            'upsSecondsOnBattery':
                {'status': 0,
                 'description': 0},
            'upsEstimatedMinutesRemaining':
                {'status': 60,
                 'description': 60},
            'upsEstimatedChargeRemaining':
                {'status': 100,
                 'description': 100},
            'upsBatteryVoltage':
                {'status': 120,
                 'description': 120},
            'upsBatteryCurrent':
                {'status': 10,
                 'description': 10},
            'upsBatteryTemperature':
                {'status': 30,
                 'description': 30},
            'upsOutputSource':
                {'status': 3,
                 'description': normal},
            'upsOutputVoltage':
                {'status': 120,
                 'description': 120},
            'upsOutputCurrent':
                {'status': 10,
                 'description': 10},
            'upsOutputPower':
                {'status': 120,
                 'description': 120},
            'upsOutputPercentLoad':
                {'status': 25,
                 'description': 25}
            'upsInputVoltage':
                {'status': 120,
                 'description': 120},
            'upsInputCurrent':
                {'status': 10,
                 'description': 10},
            'upsInputTruePower':
                {'status': 120,
                 'description': 120},
            'ups_connection':
                {'last_attempt': 1656085022.680916,
                 'connected': True},
            'timestamp': 1656085022.680916}

        Some relevant options and units for the above OIDs::

            upsBatteryStatus::
                Options:: unknown(1),
                          batteryNormal(2),
                          batteryLow(3),
                          batteryDepleted(4)
            upsSecondsOnBattery::
                Note:: Zero shall be returned if the unit is not on
                       battery power.
            upsEstimatedChargeRemaining::
                Units:: percentage
            upsBatteryVoltage::
                Units:: 0.1 Volt DC
            upsBatteryCurrent::
                Units:: 0.1 Amp DC
            upsBatteryTemperature::
                Units:: degrees Centigrade
            upsOutputSource::
                Options:: other(1),
                          none(2),
                          normal(3),
                          bypass(4),
                          battery(5),
                          booster(6),
                          reducer(7)
            upsOutputVoltage::
                Units:: RMS Volts
            upsOutputCurrent::
                Units:: 0.1 RMS Amp
            upsOutputPower::
                Units:: Watts
            upsInputVoltage::
                Units:: RMS Volts
            upsInputCurrent::
                Units:: 0.1 RMS Amp
            upsInputTruePower::
                Units:: Watts
        """

        self.is_streaming = True
        timeout = time.time() + 60 * self.restart  # exit loop after self.restart minutes
        while self.is_streaming:
            if ((self.restart != 0) and (time.time() > timeout)):
                break
            yield dsleep(1)
            if not self.connected:
                self.log.error('No SNMP response. Check your connection.')
                self.log.info('Trying to reconnect.')

            read_time = time.time()

            # Check if 60 seconds has passed before getting status
            if (read_time - self.lastGet) < 60:
                continue

            main_get_list = []
            get_list = []

            # Create the list of OIDs to send get commands
            oids = ['upsIdentManufacturer',
                    'upsIdentModel',
                    'upsBatteryStatus',
                    'upsSecondsOnBattery',
                    'upsEstimatedMinutesRemaining',
                    'upsEstimatedChargeRemaining',
                    'upsBatteryVoltage',
                    'upsBatteryCurrent',
                    'upsBatteryTemperature',
                    'upsOutputSource']

            for oid in oids:
                main_get_list.append(('UPS-MIB', oid, 0))
                get_list.append(('UPS-MIB', oid, 0))

            ups_get_result = yield self.snmp.get(get_list, self.version)
            if ups_get_result is None:
                self.connected = False
                continue
            self.connected = True

            # Append input OIDs to GET list
            input_oids = ['upsInputVoltage',
                          'upsInputCurrent',
                          'upsInputTruePower']

            # Use number of input lines used to append correct number of input OIDs
            input_num_lines = [('UPS-MIB', 'upsInputNumLines', 0)]
            num_res = yield self.snmp.get(input_num_lines, self.version)
            if num_res is None:
                self.connected = False
                continue
            self.connected = True
            input_get_results = []
            inputs = num_res[0][1]._value
            for i in range(inputs):
                get_list = []
                for oid in input_oids:
                    main_get_list.append(('UPS-MIB', oid, i + 1))
                    get_list.append(('UPS-MIB', oid, i + 1))
                input_get_result = yield self.snmp.get(get_list, self.version)
                input_get_results.append(input_get_result)

            # Append output OIDs to GET list
            output_oids = ['upsOutputVoltage',
                           'upsOutputCurrent',
                           'upsOutputPower',
                           'upsOutputPercentLoad']

            # Use number of output lines used to append correct number of output OIDs
            output_num_lines = [('UPS-MIB', 'upsOutputNumLines', 0)]
            num_res = yield self.snmp.get(output_num_lines, self.version)
            if num_res is None:
                self.connected = False
                continue
            self.connected = True
            output_get_results = []
            outputs = num_res[0][1]._value
            for i in range(outputs):
                get_list = []
                for oid in output_oids:
                    main_get_list.append(('UPS-MIB', oid, i + 1))
                    get_list.append(('UPS-MIB', oid, i + 1))
                output_get_result = yield self.snmp.get(get_list, self.version)
                output_get_results.append(output_get_result)

            # Issue SNMP GET command
            get_result = yield self.snmp.get(main_get_list, self.version)
            if get_result is None:
                self.connected = False
                continue
            self.connected = True

            # Do not publish if UPS connection has dropped
            try:
                # Update session.data
                session.data = update_cache(get_result, read_time)
                self.log.debug("{data}", data=session.data)

                if not self.connected:
                    raise ConnectionError('No SNMP response. Check your connection.')

                self.lastGet = time.time()
                # Publish to feed
                if ups_get_result is not None:
                    message = _build_message(ups_get_result, read_time, 'ups')
                    self.log.debug("{msg}", msg=message)
                    session.app.publish_to_feed('ups', message)
                for i, result in enumerate(input_get_results):
                    if result is not None:
                        blockname = f'input_{i}'
                        message = _build_message(result, read_time, blockname)
                        self.log.debug("{msg}", msg=message)
                        session.app.publish_to_feed('ups', message)
                for i, result in enumerate(output_get_results):
                    if result is not None:
                        blockname = f'output_{i}'
                        message = _build_message(result, read_time, blockname)
                        self.log.debug("{msg}", msg=message)
                        session.app.publish_to_feed('ups', message)
            except ConnectionError as e:
                self.log.error(f'{e}')
                yield dsleep(1)
                self.log.info('Trying to reconnect.')

            if params['test_mode']:
                break

        # Exit agent to release memory
        # Add "restart: unless-stopped" to docker compose to automatically restart container
        if ((not params['test_mode']) and (timeout != 0) and (self.is_streaming)):
            self.log.info(f"{self.restart} minutes have elasped. Exiting agent.")
            os.kill(os.getppid(), signal.SIGTERM)

        return True, "Finished Recording"

    def _stop_acq(self, session, params=None):
        """_stop_acq()
        **Task** - Stop task associated with acq process.
        """
        if self.is_streaming:
            session.set_status('stopping')
            self.is_streaming = False
            return True, "Stopping Recording"
        else:
            return False, "Acq is not currently running"


def add_agent_args(parser=None):
    """
    Build the argument parser for the Agent. Allows sphinx to automatically
    build documentation based on this function.
    """
    if parser is None:
        parser = argparse.ArgumentParser()

    pgroup = parser.add_argument_group("Agent Options")
    pgroup.add_argument("--address", help="Address to listen to.")
    pgroup.add_argument("--port", default=161,
                        help="Port to listen on.")
    pgroup.add_argument("--snmp-version", default='1', choices=['1', '2', '3'],
                        help="SNMP version for communication. Must match "
                             + "configuration on the UPS.")
    pgroup.add_argument("--restart-time", default=0,
                        help="Number of minutes before restarting agent.")
    pgroup.add_argument("--mode", choices=['acq', 'test'])

    return parser


def main(args=None):
    # Start logging
    txaio.start_logging(level=os.environ.get("LOGLEVEL", "info"))

    parser = add_agent_args()
    args = site_config.parse_args(agent_class='UPSAgent',
                                  parser=parser,
                                  args=args)

    if args.mode == 'acq':
        init_params = True
    elif args.mode == 'test':
        init_params = False

    agent, runner = ocs_agent.init_site_agent(args)
    p = UPSAgent(agent,
                 address=args.address,
                 port=int(args.port),
                 version=int(args.snmp_version),
                 restart_time=int(args.restart_time))

    agent.register_process("acq",
                           p.acq,
                           p._stop_acq,
                           startup=init_params, blocking=False)

    runner.run(agent, auto_reconnect=True)


if __name__ == "__main__":
    main()

1	import argparse	1✔
2	import os	1✔
3	import signal	1✔
4	import time	1✔
5
6	import txaio	1✔
7	from autobahn.twisted.util import sleep as dsleep	1✔
8	from ocs import ocs_agent, site_config	1✔
9	from ocs.ocs_twisted import TimeoutLock	1✔
10	from twisted.internet.defer import inlineCallbacks	1✔
11
12	from socs.snmp import SNMPTwister	1✔
13
14	# For logging
15	txaio.use_twisted()	1✔
16
17
18	def _extract_oid_field_and_value(get_result):	1✔
19	"""Extract field names and OID values from SNMP GET results.
20
21	The ObjectType objects returned from pysnmp interactions contain the
22	info we want to use for field names, specifically the OID and associated
23	integer for uniquely identifying duplicate OIDs, as well as the value of the
24	OID, which we want to save.
25
26	Here we use the prettyPrint() method to get the OID name, requiring
27	some string manipulation. We also just grab the hidden ._value
28	directly, as this seemed the cleanest way to get an actual value of a
29	normal type. Without doing this we get a pysnmp defined Integer32 or
30	DisplayString, which were awkward to handle, particularly the
31	DisplayString.
32
33	Parameters
34	----------
35	get_result : pysnmp.smi.rfc1902.ObjectType
36	Result from a pysnmp GET command.
37
38	Returns
39	-------
40	field_name : str
41	Field name for an OID, i.e. 'upsIdentModel'
42	oid_value : int or str
43	Associated value for the OID. Returns None if not an int or str
44	oid_description : str
45	String description of the OID value.
46	"""
47	# OID from SNMP GET
48	oid = get_result[0].prettyPrint()	1✔
49	# Makes something like 'UPS-MIB::upsIdentModel.0'
50	# look like 'upsIdentModel_0'
51	field_name = oid.split("::")[1].replace('.', '_')	1✔
52
53	# Grab OID value, mostly these are integers
54	oid_value = get_result[1]._value	1✔
55	oid_description = get_result[1].prettyPrint()	1✔
56
57	# Decode string values
58	if isinstance(oid_value, bytes):	1✔
59	oid_value = oid_value.decode("utf-8")	1✔
60
61	# I don't expect any other types at the moment, but just in case.
62	if not isinstance(oid_value, (int, bytes, str)):	1✔
63	oid_value = None	×
64
65	return field_name, oid_value, oid_description	1✔
66
67
68	def _build_message(get_result, time, blockname):	1✔
69	"""Build the message for publication on an OCS Feed.
70
71	Parameters
72	----------
73	get_result : pysnmp.smi.rfc1902.ObjectType
74	Result from a pysnmp GET command.
75	time : float
76	Timestamp for when the SNMP GET was issued.
77
78	Returns
79	-------
80	message : dict
81	OCS Feed formatted message for publishing
82	"""
83	message = {	1✔
84	'block_name': blockname,
85	'timestamp': time,
86	'data': {}
87	}
88
89	for item in get_result:	1✔
90	field_name, oid_value, oid_description = _extract_oid_field_and_value(item)	1✔
91
92	if oid_value is None:	1✔
93	continue	×
94
95	message['data'][field_name] = oid_value	1✔
96	message['data'][field_name + "_description"] = oid_description	1✔
97
98	return message	1✔
99
100
101	def update_cache(get_result, timestamp):	1✔
102	"""Update the OID Value Cache.
103
104	The OID Value Cache is used to store each unique OID and will be passed to
105	session.data
106
107	The cache consists of a dictionary, with the unique OIDs as keys, and
108	another dictionary as the value. Each of these nested dictionaries contains the
109	OID values, name, and description (decoded string). An example for a single OID::
110
111	{"upsBatteryStatus":
112	{"status": 2,
113	"description": "batteryNormal"}}
114
115	Additionally there is connection status and timestamp information under::
116
117	{"ups_connection":
118	{"last_attempt": 1598543359.6326838,
119	"connected": True}
120	{"timestamp": 1656085022.680916}
121
122	Parameters
123	----------
124	get_result : pysnmp.smi.rfc1902.ObjectType
125	Result from a pysnmp GET command.
126	timestamp : float
127	Timestamp for when the SNMP GET was issued.
128	"""
129	oid_cache = {}	1✔
130	# Return disconnected if SNMP response is empty
131	if get_result is None:	1✔
132	oid_cache['ups_connection'] = {'last_attempt': time.time(),	×
133	'connected': False}
134	return oid_cache	×
135
136	for item in get_result:	1✔
137	field_name, oid_value, oid_description = _extract_oid_field_and_value(item)	1✔
138	if oid_value is None:	1✔
139	continue	×
140
141	# Update OID Cache for session.data
142	oid_cache[field_name] = {"status": oid_value}	1✔
143	oid_cache[field_name]["description"] = oid_description	1✔
144	oid_cache['ups_connection'] = {'last_attempt': time.time(),	1✔
145	'connected': True}
146	oid_cache['timestamp'] = timestamp	1✔
147
148	return oid_cache	1✔
149
150
151	class UPSAgent:	1✔
152	"""Monitor the UPS system via SNMP.
153
154	Parameters
155	----------
156	agent : OCSAgent
157	OCSAgent object which forms this Agent
158	address : str
159	Address of the UPS.
160	port : int
161	SNMP port to issue GETs to, default to 161.
162	version : int
163	SNMP version for communication (1, 2, or 3), defaults to 3.
164
165	Attributes
166	----------
167	agent : OCSAgent
168	OCSAgent object which forms this Agent
169	is_streaming : bool
170	Tracks whether or not the agent is actively issuing SNMP GET commands
171	to the UPS. Setting to false stops sending commands.
172	log : txaio.tx.Logger
173	txaio logger object, created by the OCSAgent
174	"""
175
176	def __init__(self, agent, address, port=161, version=1, restart_time=0):	1✔
177	self.agent = agent	1✔
178	self.is_streaming = False	1✔
179	self.log = self.agent.log	1✔
180	self.lock = TimeoutLock()	1✔
181
182	self.log.info(f'Using SNMP version {version}.')	1✔
183	self.version = version	1✔
184	self.snmp = SNMPTwister(address, port)	1✔
185	self.connected = True	1✔
186	self.restart = restart_time	1✔
187
188	self.lastGet = 0	1✔
189
190	agg_params = {	1✔
191	'frame_length': 10 * 60 # [sec]
192	}
193	self.agent.register_feed('ups',	1✔
194	record=True,
195	agg_params=agg_params,
196	buffer_time=0)
197
198	@ocs_agent.param('test_mode', default=False, type=bool)	1✔
199	@inlineCallbacks	1✔
200	def acq(self, session, params=None):	1✔
201	"""acq()
202
203	Process - Fetch values from the UPS via SNMP.
204
205	Parameters
206	----------
207	test_mode : bool, optional
208	Run the Process loop only once. Meant only for testing.
209	Default is False.
210
211	Notes
212	-----
213	The most recent data collected is stored in session.data in the
214	structure::
215
216	>>> response.session['data']
217	{'upsIdentManufacturer':
218	{'description': 'Falcon'},
219	'upsIdentModel':
220	{'description': 'SSG3KRM-2'},
221	'upsBatteryStatus':
222	{'status': 2,
223	'description': 'batteryNormal'},
224	'upsSecondsOnBattery':
225	{'status': 0,
226	'description': 0},
227	'upsEstimatedMinutesRemaining':
228	{'status': 60,
229	'description': 60},
230	'upsEstimatedChargeRemaining':
231	{'status': 100,
232	'description': 100},
233	'upsBatteryVoltage':
234	{'status': 120,
235	'description': 120},
236	'upsBatteryCurrent':
237	{'status': 10,
238	'description': 10},
239	'upsBatteryTemperature':
240	{'status': 30,
241	'description': 30},
242	'upsOutputSource':
243	{'status': 3,
244	'description': normal},
245	'upsOutputVoltage':
246	{'status': 120,
247	'description': 120},
248	'upsOutputCurrent':
249	{'status': 10,
250	'description': 10},
251	'upsOutputPower':
252	{'status': 120,
253	'description': 120},
254	'upsOutputPercentLoad':
255	{'status': 25,
256	'description': 25}
257	'upsInputVoltage':
258	{'status': 120,
259	'description': 120},
260	'upsInputCurrent':
261	{'status': 10,
262	'description': 10},
263	'upsInputTruePower':
264	{'status': 120,
265	'description': 120},
266	'ups_connection':
267	{'last_attempt': 1656085022.680916,
268	'connected': True},
269	'timestamp': 1656085022.680916}
270
271	Some relevant options and units for the above OIDs::
272
273	upsBatteryStatus::
274	Options:: unknown(1),
275	batteryNormal(2),
276	batteryLow(3),
277	batteryDepleted(4)
278	upsSecondsOnBattery::
279	Note:: Zero shall be returned if the unit is not on
280	battery power.
281	upsEstimatedChargeRemaining::
282	Units:: percentage
283	upsBatteryVoltage::
284	Units:: 0.1 Volt DC
285	upsBatteryCurrent::
286	Units:: 0.1 Amp DC
287	upsBatteryTemperature::
288	Units:: degrees Centigrade
289	upsOutputSource::
290	Options:: other(1),
291	none(2),
292	normal(3),
293	bypass(4),
294	battery(5),
295	booster(6),
296	reducer(7)
297	upsOutputVoltage::
298	Units:: RMS Volts
299	upsOutputCurrent::
300	Units:: 0.1 RMS Amp
301	upsOutputPower::
302	Units:: Watts
303	upsInputVoltage::
304	Units:: RMS Volts
305	upsInputCurrent::
306	Units:: 0.1 RMS Amp
307	upsInputTruePower::
308	Units:: Watts
309	"""
310
311	self.is_streaming = True	1✔
312	timeout = time.time() + 60 * self.restart # exit loop after self.restart minutes	1✔
313	while self.is_streaming:	1✔
314	if ((self.restart != 0) and (time.time() > timeout)):	1✔
315	break	×
316	yield dsleep(1)	1✔
317	if not self.connected:	1✔
318	self.log.error('No SNMP response. Check your connection.')	1✔
319	self.log.info('Trying to reconnect.')	1✔
320
321	read_time = time.time()	1✔
322
323	# Check if 60 seconds has passed before getting status
324	if (read_time - self.lastGet) < 60:	1✔
325	continue	×
326
327	main_get_list = []	1✔
328	get_list = []	1✔
329
330	# Create the list of OIDs to send get commands
331	oids = ['upsIdentManufacturer',	1✔
332	'upsIdentModel',
333	'upsBatteryStatus',
334	'upsSecondsOnBattery',
335	'upsEstimatedMinutesRemaining',
336	'upsEstimatedChargeRemaining',
337	'upsBatteryVoltage',
338	'upsBatteryCurrent',
339	'upsBatteryTemperature',
340	'upsOutputSource']
341
342	for oid in oids:	1✔
343	main_get_list.append(('UPS-MIB', oid, 0))	1✔
344	get_list.append(('UPS-MIB', oid, 0))	1✔
345
346	ups_get_result = yield self.snmp.get(get_list, self.version)	1✔
347	if ups_get_result is None:	1✔
348	self.connected = False	1✔
349	continue	1✔
350	self.connected = True	1✔
351
352	# Append input OIDs to GET list
353	input_oids = ['upsInputVoltage',	1✔
354	'upsInputCurrent',
355	'upsInputTruePower']
356
357	# Use number of input lines used to append correct number of input OIDs
358	input_num_lines = [('UPS-MIB', 'upsInputNumLines', 0)]	1✔
359	num_res = yield self.snmp.get(input_num_lines, self.version)	1✔
360	if num_res is None:	1✔
361	self.connected = False	×
362	continue	×
363	self.connected = True	1✔
364	input_get_results = []	1✔
365	inputs = num_res[0][1]._value	1✔
366	for i in range(inputs):	1✔
367	get_list = []	1✔
368	for oid in input_oids:	1✔
369	main_get_list.append(('UPS-MIB', oid, i + 1))	1✔
370	get_list.append(('UPS-MIB', oid, i + 1))	1✔
371	input_get_result = yield self.snmp.get(get_list, self.version)	1✔
372	input_get_results.append(input_get_result)	1✔
373
374	# Append output OIDs to GET list
375	output_oids = ['upsOutputVoltage',	1✔
376	'upsOutputCurrent',
377	'upsOutputPower',
378	'upsOutputPercentLoad']
379
380	# Use number of output lines used to append correct number of output OIDs
381	output_num_lines = [('UPS-MIB', 'upsOutputNumLines', 0)]	1✔
382	num_res = yield self.snmp.get(output_num_lines, self.version)	1✔
383	if num_res is None:	1✔
384	self.connected = False	×
385	continue	×
386	self.connected = True	1✔
387	output_get_results = []	1✔
388	outputs = num_res[0][1]._value	1✔
389	for i in range(outputs):	1✔
390	get_list = []	1✔
391	for oid in output_oids:	1✔
392	main_get_list.append(('UPS-MIB', oid, i + 1))	1✔
393	get_list.append(('UPS-MIB', oid, i + 1))	1✔
394	output_get_result = yield self.snmp.get(get_list, self.version)	1✔
395	output_get_results.append(output_get_result)	1✔
396
397	# Issue SNMP GET command
398	get_result = yield self.snmp.get(main_get_list, self.version)	1✔
399	if get_result is None:	1✔
UNCOV 400	self.connected = False	×
UNCOV 401	continue	×
402	self.connected = True	1✔
403
404	# Do not publish if UPS connection has dropped
405	try:	1✔
406	# Update session.data
407	session.data = update_cache(get_result, read_time)	1✔
408	self.log.debug("{data}", data=session.data)	1✔
409
410	if not self.connected:	1✔
411	raise ConnectionError('No SNMP response. Check your connection.')	×
412
413	self.lastGet = time.time()	1✔
414	# Publish to feed
415	if ups_get_result is not None:	1✔
416	message = _build_message(ups_get_result, read_time, 'ups')	1✔
417	self.log.debug("{msg}", msg=message)	1✔
418	session.app.publish_to_feed('ups', message)	1✔
419	for i, result in enumerate(input_get_results):	1✔
420	if result is not None:	1✔
421	blockname = f'input_{i}'	1✔
422	message = _build_message(result, read_time, blockname)	1✔
423	self.log.debug("{msg}", msg=message)	1✔
424	session.app.publish_to_feed('ups', message)	1✔
425	for i, result in enumerate(output_get_results):	1✔
426	if result is not None:	1✔
427	blockname = f'output_{i}'	1✔
428	message = _build_message(result, read_time, blockname)	1✔
429	self.log.debug("{msg}", msg=message)	1✔
430	session.app.publish_to_feed('ups', message)	1✔
431	except ConnectionError as e:	×
432	self.log.error(f'{e}')	×
433	yield dsleep(1)	×
434	self.log.info('Trying to reconnect.')	×
435
436	if params['test_mode']:	1✔
437	break	1✔
438
439	# Exit agent to release memory
440	# Add "restart: unless-stopped" to docker compose to automatically restart container
441	if ((not params['test_mode']) and (timeout != 0) and (self.is_streaming)):	1✔
442	self.log.info(f"{self.restart} minutes have elasped. Exiting agent.")	×
443	os.kill(os.getppid(), signal.SIGTERM)	×
444
445	return True, "Finished Recording"	1✔
446
447	def _stop_acq(self, session, params=None):	1✔
448	"""_stop_acq()
449	Task - Stop task associated with acq process.
450	"""
451	if self.is_streaming:	×
452	session.set_status('stopping')	×
453	self.is_streaming = False	×
454	return True, "Stopping Recording"	×
455	else:
456	return False, "Acq is not currently running"	×
457
458
459	def add_agent_args(parser=None):	1✔
460	"""
461	Build the argument parser for the Agent. Allows sphinx to automatically
462	build documentation based on this function.
463	"""
464	if parser is None:	1✔
465	parser = argparse.ArgumentParser()	1✔
466
467	pgroup = parser.add_argument_group("Agent Options")	1✔
468	pgroup.add_argument("--address", help="Address to listen to.")	1✔
469	pgroup.add_argument("--port", default=161,	1✔
470	help="Port to listen on.")
471	pgroup.add_argument("--snmp-version", default='1', choices=['1', '2', '3'],	1✔
472	help="SNMP version for communication. Must match "
473	+ "configuration on the UPS.")
474	pgroup.add_argument("--restart-time", default=0,	1✔
475	help="Number of minutes before restarting agent.")
476	pgroup.add_argument("--mode", choices=['acq', 'test'])	1✔
477
478	return parser	1✔
479
480
481	def main(args=None):	1✔
482	# Start logging
483	txaio.start_logging(level=os.environ.get("LOGLEVEL", "info"))	1✔
484
485	parser = add_agent_args()	1✔
486	args = site_config.parse_args(agent_class='UPSAgent',	1✔
487	parser=parser,
488	args=args)
489
490	if args.mode == 'acq':	1✔
491	init_params = True	×
492	elif args.mode == 'test':	1✔
493	init_params = False	1✔
494
495	agent, runner = ocs_agent.init_site_agent(args)	1✔
496	p = UPSAgent(agent,	1✔
497	address=args.address,
498	port=int(args.port),
499	version=int(args.snmp_version),
500	restart_time=int(args.restart_time))
501
502	agent.register_process("acq",	1✔
503	p.acq,
504	p._stop_acq,
505	startup=init_params, blocking=False)
506
507	runner.run(agent, auto_reconnect=True)	1✔
508
509
510	if __name__ == "__main__":	1✔
511	main()	1✔

simonsobs / socs / 15356518725

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous