Initial commit

library/grow/__init__.py

@@ -0,0 +1 @@
+__version__ = '0.0.3'

library/grow/gas.py

@@ -0,0 +1,140 @@
+"""Read the MICS6814 via an ads1015 ADC"""
+
+import time
+import atexit
+import ads1015
+import RPi.GPIO as GPIO
+
+MICS6814_HEATER_PIN = 24
+MICS6814_GAIN = 6.144
+
+ads1015.I2C_ADDRESS_DEFAULT = ads1015.I2C_ADDRESS_ALTERNATE
+_is_setup = False
+_adc_enabled = False
+_adc_gain = 6.148
+
+
+class Mics6814Reading(object):
+    __slots__ = 'oxidising', 'reducing', 'nh3', 'adc'
+
+    def __init__(self, ox, red, nh3, adc=None):
+        self.oxidising = ox
+        self.reducing = red
+        self.nh3 = nh3
+        self.adc = adc
+
+    def __repr__(self):
+        fmt = """Oxidising: {ox:05.02f} Ohms
+Reducing: {red:05.02f} Ohms
+NH3: {nh3:05.02f} Ohms"""
+        if self.adc is not None:
+            fmt += """
+ADC: {adc:05.02f} Volts
+"""
+        return fmt.format(
+            ox=self.oxidising,
+            red=self.reducing,
+            nh3=self.nh3,
+            adc=self.adc)
+
+    __str__ = __repr__
+
+
+def setup():
+    global adc, _is_setup
+    if _is_setup:
+        return
+    _is_setup = True
+
+    adc = ads1015.ADS1015(i2c_addr=0x49)
+    adc.set_mode('single')
+    adc.set_programmable_gain(MICS6814_GAIN)
+    adc.set_sample_rate(1600)
+
+    GPIO.setwarnings(False)
+    GPIO.setmode(GPIO.BCM)
+    GPIO.setup(MICS6814_HEATER_PIN, GPIO.OUT)
+    GPIO.output(MICS6814_HEATER_PIN, 1)
+    atexit.register(cleanup)
+
+
+def enable_adc(value=True):
+    """Enable reading from the additional ADC pin."""
+    global _adc_enabled
+    _adc_enabled = value
+
+
+def set_adc_gain(value):
+    """Set gain value for the additional ADC pin."""
+    global _adc_gain
+    _adc_gain = value
+
+
+def cleanup():
+    GPIO.output(MICS6814_HEATER_PIN, 0)
+
+
+def read_all():
+    """Return gas resistence for oxidising, reducing and NH3"""
+    setup()
+    ox = adc.get_voltage('in0/gnd')
+    red = adc.get_voltage('in1/gnd')
+    nh3 = adc.get_voltage('in2/gnd')
+
+    try:
+        ox = (ox * 56000) / (3.3 - ox)
+    except ZeroDivisionError:
+        ox = 0
+
+    try:
+        red = (red * 56000) / (3.3 - red)
+    except ZeroDivisionError:
+        red = 0
+
+    try:
+        nh3 = (nh3 * 56000) / (3.3 - nh3)
+    except ZeroDivisionError:
+        nh3 = 0
+
+    analog = None
+
+    if _adc_enabled:
+        if _adc_gain == MICS6814_GAIN:
+            analog = adc.get_voltage('ref/gnd')
+        else:
+            adc.set_programmable_gain(_adc_gain)
+            time.sleep(0.05)
+            analog = adc.get_voltage('ref/gnd')
+            adc.set_programmable_gain(MICS6814_GAIN)
+
+    return Mics6814Reading(ox, red, nh3, analog)
+
+
+def read_oxidising():
+    """Return gas resistance for oxidising gases.
+
+    Eg chlorine, nitrous oxide
+    """
+    setup()
+    return read_all().oxidising
+
+
+def read_reducing():
+    """Return gas resistance for reducing gases.
+
+    Eg hydrogen, carbon monoxide
+    """
+    setup()
+    return read_all().reducing
+
+
+def read_nh3():
+    """Return gas resistance for nh3/ammonia"""
+    setup()
+    return read_all().nh3
+
+
+def read_adc():
+    """Return spare ADC channel value"""
+    setup()
+    return read_all().adc

library/grow/noise.py

@@ -0,0 +1,90 @@
+import sounddevice
+import numpy
+
+
+class Noise():
+    def __init__(self,
+                 sample_rate=16000,
+                 duration=0.5):
+        """Noise measurement.
+
+        :param sample_rate: Sample rate in Hz
+        :param duraton: Duration, in seconds, of noise sample capture
+
+        """
+
+        self.duration = duration
+        self.sample_rate = sample_rate
+
+    def get_amplitudes_at_frequency_ranges(self, ranges):
+        """Return the mean amplitude of frequencies in the given ranges.
+
+        :param ranges: List of ranges including a start and end range
+
+        """
+        recording = self._record()
+        magnitude = numpy.abs(numpy.fft.rfft(recording[:, 0], n=self.sample_rate))
+        result = []
+        for r in ranges:
+            start, end = r
+            result.append(numpy.mean(magnitude[start:end]))
+        return result
+
+    def get_amplitude_at_frequency_range(self, start, end):
+        """Return the mean amplitude of frequencies in the specified range.
+
+        :param start: Start frequency (in Hz)
+        :param end: End frequency (in Hz)
+
+        """
+        n = self.sample_rate // 2
+        if start > n or end > n:
+            raise ValueError("Maxmimum frequency is {}".format(n))
+
+        recording = self._record()
+        magnitude = numpy.abs(numpy.fft.rfft(recording[:, 0], n=self.sample_rate))
+        return numpy.mean(magnitude[start:end])
+
+    def get_noise_profile(self,
+                          noise_floor=100,
+                          low=0.12,
+                          mid=0.36,
+                          high=None):
+        """Returns a noise charateristic profile.
+
+        Bins all frequencies into 3 weighted groups expressed as a percentage of the total frequency range.
+
+        :param noise_floor: "High-pass" frequency, exclude frequencies below this value
+        :param low: Percentage of frequency ranges to count in the low bin (as a float, 0.5 = 50%)
+        :param mid: Percentage of frequency ranges to count in the mid bin (as a float, 0.5 = 50%)
+        :param high: Optional percentage for high bin, effectively creates a "Low-pass" if total percentage is less than 100%
+
+        """
+
+        if high is None:
+            high = 1.0 - low - mid
+
+        recording = self._record()
+        magnitude = numpy.abs(numpy.fft.rfft(recording[:, 0], n=self.sample_rate))
+
+        sample_count = (self.sample_rate // 2) - noise_floor
+
+        mid_start = noise_floor + int(sample_count * low)
+        high_start = mid_start + int(sample_count * mid)
+        noise_ceiling = high_start + int(sample_count * high)
+
+        amp_low = numpy.mean(magnitude[noise_floor:mid_start])
+        amp_mid = numpy.mean(magnitude[mid_start:high_start])
+        amp_high = numpy.mean(magnitude[high_start:noise_ceiling])
+        amp_total = (amp_low + amp_mid + amp_high) / 3.0
+
+        return amp_low, amp_mid, amp_high, amp_total
+
+    def _record(self):
+        return sounddevice.rec(
+            int(self.duration * self.sample_rate),
+            samplerate=self.sample_rate,
+            blocking=True,
+            channels=1,
+            dtype='float64'
+        )