Coverage for janitoo_tutorial/tutorial3.py : 41%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
|
# -*- coding: utf-8 -*-
"""
This file is part of Janitoo.
Janitoo is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
Janitoo is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with Janitoo. If not, see <http://www.gnu.org/licenses/>.
"""
return AmbianceComponent(**kwargs)
return LedComponent(**kwargs)
return TemperatureComponent(**kwargs)
return CpuComponent(**kwargs)
"""A bus to manage Tutorial """
'booting', 'sleeping', 'reporting', 'ringing', ] """The tutorial states : """
{ 'trigger': 'wakeup', 'source': 'sleeping', 'dest': 'reporting', 'conditions': 'condition_values', }, { 'trigger': 'report', 'source': '*', 'dest': 'reporting', 'conditions': 'condition_values', }, { 'trigger': 'sleep', 'source': '*', 'dest': 'sleeping', }, { 'trigger': 'ring', 'source': 'reporting', 'dest': 'ringing', 'conditions': 'condition_values', }, ] """The transitions """
""" """ node_uuid=self.uuid, help='The delay between 2 checks', label='Timer.', default=30, )
node_uuid=self.uuid, help='The critical temperature. If 2 of the 3 temperature sensors are up to this value, a security notification is sent.', label='Critical', default=50, )
node_uuid=self.uuid, help='The average temperature of tutorial.', label='Temp', )
node_uuid=self.uuid, help='Temperature overheat.', label='Overheat', default=False, )
node_uuid=self.uuid, list_items=[ v['trigger'] for v in self.transitions ], fsm_bus=self, )
node_uuid=self.uuid, genre=0x01, help='The state of the fsm.', get_data_cb = self.get_state, label='State', )
"""Get the state of the fsm """ return self.state
def polled_sensors(self): """ """ return [ self.nodeman.find_value('temperature', 'temperature'), self.nodeman.find_value('ambiance', 'temperature'), self.nodeman.find_value('ambiance', 'humidity'), self.nodeman.find_value('cpu', 'temperature'), self.nodeman.find_value('cpu', 'voltage'), self.nodeman.find_value('cpu', 'frequency'), self.nodeman.find_value('led', 'switch'), self.nodeman.find_value('led', 'blink'), self.nodeman.find_bus_value('temperature'), self.nodeman.find_bus_value('transition'), self.nodeman.find_bus_value('overheat'), ]
""" """ logger.debug("[%s] - condition_values", self.__class__.__name__) return all(v is not None for v in self.polled_sensors)
""" """ logger.debug("[%s] - on_enter_reporting", self.__class__.__name__) self.bus_acquire() try: self.nodeman.find_value('led', 'blink').data = 'heartbeat' self.nodeman.add_polls(self.polled_sensors, slow_start=True, overwrite=False) #In sleeping mode, send the state of the fsm every 900 seconds #We update poll_delay directly to not update the value in configfile state = self.nodeman.find_bus_value('state') state.poll_delay = self.nodeman.find_bus_value('state_poll').data overheat = self.nodeman.find_bus_value('overheat') overheat.poll_delay = self.nodeman.find_bus_value('overheat_poll').data self.nodeman.publish_value(overheat) self.nodeman.add_polls([state, overheat], slow_start=True, overwrite=True) except Exception: logger.exception("[%s] - Error in on_enter_reporting", self.__class__.__name__) finally: self.bus_release()
""" """ logger.debug("[%s] - on_enter_sleeping", self.__class__.__name__) self.bus_acquire() try: self.stop_check() self.nodeman.remove_polls(self.polled_sensors) self.nodeman.find_value('led', 'blink').data = 'off' #In sleeping mode, send the state of the fsm every 900 seconds #We update poll_delay directly to nto update the value in config file self.nodeman.find_bus_value('state').poll_delay = 900 except Exception: logger.exception("[%s] - Error in on_enter_sleeping", self.__class__.__name__) finally: self.bus_release()
""" """ logger.debug("[%s] - on_exit_sleeping", self.__class__.__name__) self.on_check()
""" """ logger.debug("[%s] - on_enter_ringing", self.__class__.__name__) self.bus_acquire() try: self.nodeman.find_value('led', 'blink').data = 'warning' #In sleeping mode, send the state of the fsm every 900 seconds #We update poll_delay directly to not update the value in configfile state = self.nodeman.find_bus_value('state') state.poll_delay = 1.0 * self.nodeman.find_bus_value('state_poll').data / 3 overheat = self.nodeman.find_bus_value('overheat') overheat.poll_delay = 1.0 * self.nodeman.find_bus_value('overheat_poll').data / 3 self.nodeman.publish_value(overheat) self.nodeman.add_polls([state, overheat], slow_start=True, overwrite=True) except Exception: logger.exception("[%s] - Error in on_enter_ringing", self.__class__.__name__) finally: self.bus_release()
"""Get a lock on the bus""" if self._bus_lock.acquire(blocking): return True return False
"""Release a lock on the bus""" self._bus_lock.release()
"""Get status of the lock""" return self._bus_lock.locked()
"""Check that the component is 'available'
""" self.check_timer.cancel() self.check_timer = None
"""Make a check using a timer.
""" self.bus_acquire() try: self.stop_check() if self.check_timer is None and self.is_started: timer_delay = self.get_bus_value('timer_delay').data if self.state == 'ringing': timer_delay = 1.0 * timer_delay / 2 self.check_timer = threading.Timer(timer_delay, self.on_check) self.check_timer.start() finally: self.bus_release() try: state = True #Check the temperatures critical_temp = self.get_bus_value('temperature_critical').data criticals = 0 nums = 0 total = 0 mini = maxi = None for value in [('temperature', 'temperature'), ('ambiance', 'temperature'), ('cpu', 'temperature')]: data = self.nodeman.find_value(*value).data if data is None: #We should notify a sensor problem here. pass else: nums += 1 total += data if data > critical_temp: criticals += 1 if maxi is None or data > maxi: maxi = data if mini is None or data < mini: mini = data if criticals > 1: if self.state != 'ringing': #We should notify a security problem : fire ? self.nodeman.find_bus_value('overheat').data = True self.ring() elif self.state == 'ringing': #We should notify a security problem : fire ? self.nodeman.find_bus_value('overheat').data = False self.report() if nums != 0: self.get_bus_value('temperature').data = total / nums except Exception: logger.exception("[%s] - Error in on_check", self.__class__.__name__)
"""Start the bus """ for bus in self.buses: self.buses[bus].start(mqttc, trigger_thread_reload_cb=None) JNTBus.start(self, mqttc, trigger_thread_reload_cb) self._statemachine = self.create_fsm()
"""Create the fsm """ if hasattr(self, "get_graph"): delattr(self, "get_graph") return Machine(self, states=self.states, transitions=self.transitions, title='Bus', initial='booting')
"""Stop the bus """ for bus in self.buses: self.buses[bus].stop() JNTBus.stop(self)
"""Check that the component is 'available'
""" res = False #~ for bus in self.buses: #~ res = res and self.buses[bus].check_heartbeat() logger.debug("[%s] - sensors %s", self.__class__.__name__, self.polled_sensors) if self.state == 'booting' and all(v is not None for v in self.polled_sensors): #We try to enter in sleeping mode self.sleep() return self.state != 'booting'
"""Retrieve data Don't do long task in loop. Use a separated thread to not perturbate the nodeman
""" for bus in self.buses: self.buses[bus].loop(stopevent)
""" A component for ambiance """
""" """ oid = kwargs.pop('oid', '%s.ambiance'%OID) name = kwargs.pop('name', "Ambiance sensor") DHTComponent.__init__(self, oid=oid, bus=bus, addr=addr, name=name, **kwargs) logger.debug("[%s] - __init__ node uuid:%s", self.__class__.__name__, self.uuid)
""" A component for a Led (on/off) """
""" """ oid = kwargs.pop('oid', '%s.led'%OID) name = kwargs.pop('name', "Led") GPIOLed.__init__(self, oid=oid, bus=bus, addr=addr, name=name, **kwargs) logger.debug("[%s] - __init__ node uuid:%s", self.__class__.__name__, self.uuid)
""" A water temperature component """
""" """ oid = kwargs.pop('oid', '%s.temperature'%OID) name = kwargs.pop('name', "Temperature") DS18B20.__init__(self, oid=oid, bus=bus, addr=addr, name=name, **kwargs) logger.debug("[%s] - __init__ node uuid:%s", self.__class__.__name__, self.uuid)
""" A water temperature component """
""" """ oid = kwargs.pop('oid', '%s.cpu'%OID) name = kwargs.pop('name', "CPU") HardwareCpu.__init__(self, oid=oid, bus=bus, addr=addr, name=name, **kwargs) logger.debug("[%s] - __init__ node uuid:%s", self.__class__.__name__, self.uuid) |