<reponame>l-hahn/green-environment
#!/usr/bin/env python3
"""
This module provides a class to interact with the DS18B20 temperature
sensor and is considered mostly to be used on a Raspberry Pi. With
aprorpiate naming this class might be used for other SoC like Arduino,
Genuino etc.

This module can also be used as a standalone script to retrieve values from.
attached sensors.

Classes: DS18B20
Functions: main
"""
import os
import argparse
import time

os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')

class DS18B20:
    """
    This class provides a set of functions and methods that can be used to
    interact with the DS18B20 sensor. As values can be set/adjusted during
    runtime, there is no need to instantiate a new object for every config
    change. Multiple sensors can be handled at once by passing a list of
    devices or using None (take all available devices).

    Methods:
        default_functor(values, **kwargs)
        __init__(device, scale, idle_time)
        set_devices()
        devices()
        set_device_path()
        device_path()
        set_scale()
        scale()
        set_idle_time()
        idle_time()
        __to_celsius
        __to_kelvin
        __to_fahrenheit
        get(iteration)
        run(iteration, functor, **kwargs)
        run_endless(iteration, functor, **kwargs)
    """
    def default_functor(values: "list of ints" or int, **kwargs):
        """
        An example of how functions can be passed to the run function,
        to make use of a value handling (e.g. printing values).

        Keyword arguments:
            values (list of ints) -- the measured values for one run.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        if not isinstance(values, list):
            print(values, **kwargs)
        else:
            print("\t".join(str(value) for value in values), **kwargs)


    def __init__(self, device, scale: str = "C", idle_time: int = 2):
        """
        Constructor to instantiate an object, that is able to handle multiple
        devices at once.

        Keyword arguments:
            device (str, list of str) -- The 1-wire devices that should be used.
            scale (str, optional) -- The temperature scalce (K,F,C) to be used
            idle_time (int, optional) -- The idle time between two measurements.
        """
        self.set_scale(scale)
        self._idle_time = idle_time
        self._device_path = "/sys/bus/w1/devices"
        dev = [device] if not isinstance(device,list) else device
        self.set_devices(dev)
        


    def set_devices(self, devices: list=None):
        """
        This functions loads the required 1-wire devices.
        If no device list is provided, all available devices will be used.

        Keyword arguments:
            devices (list of str) -- the devices to be used; leave empty
                to use all devices

        """
        device_list = [devices] if not isinstance(devices, list) else devices
        # filter available 1-wire devices #
        one_wire_devices = [
            dev.name for dev in os.scandir(self._device_path)
            if dev.is_dir() and dev.name != "w1_bus_master1"
        ]

        # filter for active 1-wire devices #
        self._devices = [
            dev for dev in one_wire_devices
            if "w1_slave" in (
                item.name for item in os.scandir("{}/{}".format(self._device_path, dev))
            )
        ]

        # reduce to chosen, active devices #
        if device_list != [None]:
            self._devices = [
                device
                for device in self._devices
                if device in device_list
            ]
            if len(self._devices) != len(device_list):
                missing_devices = [
                    dev for dev in devices if dev not in self._devices
                ]
                raise FileNotFoundError(
                    "Provided devices [{}] cannot be found! Aborting.".format(
                        ", ".join(missing_devices)
                    )
                )
        if not self._devices:
            raise FileNotFoundError("No 1-wire device found!")


    def devices(self):
        """
        The function to get (by return) the current device list in use.

        Returns:
            (list) -- The device list.
        """
        return list(self._devices)


    def set_device_path(self, device_path: str):
        """
        Sets the device path, were 1-wire devices can be found.
        Usually it is '/sys/bus/w1/devices'

        Keyword arguments:
            device_path (str) -- The device path, e.g. /sys/bus/w1/devices
        """
        self._device_path = device_path


    def device_path(self):
        """
        The function to get (by return) the current device_path, where
        1-wire devices should be found.

        Returns:
            (str) -- The device_path as a string.
        """
        return self._device_path


    def set_scale(self, scale: str):
        """
        Sets the temperature scale, that should be used; it is either
        'K', 'F' or 'C'; relates to Kevlin, Fahrenheit or Celsius.
        Returned values are to be read with that scale; default is C.

        Keyword arguments:
            scale (str) -- The temperature scale: 'K', 'F' or 'C'.
        """
        if scale not in ("K", "F", "C"):
            raise ValueError(
                "Unknown scale type '{}'! Needs to be K, F or C!".format(scale)
            )
        self._scale = scale
        if self._scale == "C":
            self._translator = self.__to_celsius
        elif self._scale == "F":
            # Fahrenheit temperature scale #
            self._translator = self.__to_fahrenheit
        else:
            # Kelvin temperature scale #
            self._translator = self.__to_kelvin


    def scale(self):
        """
        The function to get (by return) the current used temperature scale.
        This is either 'K', 'F' or 'C'.

        Returns:
            (str) -- The uses temperature scale as a string.
        """
        return self._scale


    def set_idle_time(self, idle_time: int):
        """
        Sets the idle time that is used in between two
        measurements when using run(...) or run_endless(...)

        Keyword arguments:
            idle_time (int) -- The idle time.
        """
        self._idle_time = idle_time


    def idle_time(self):
        """
        The function to get (by return) the current used idle time.

        Returns:
            (int) -- The currently used idle time.
        """
        return self._idle_time


    def __to_celsius(self, value: int):
        """
        Transforms measured value into Celcius temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return value / 1000.0


    def __to_kelvin(self, value: int):
        """
        Transforms measured value into Kelvin temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return (value/1000) + 273.15


    def __to_fahrenheit(self, value: int):
        """
        Transforms measured value into Fahrenheit temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return (value/1000)*(9/5) + 32


    def get(self, iteration: int = 1):
        """
        Function to get a certain amount of measured values; there is no
        on-line handling. Values will be measured and returned.

        Keyword arguments:
            iteration (int, optional) -- Measured values amount. Defaults to 1.

        Returns:
            (list of list of ints or list of ints) -- The measured values
        """
        iteration_values = []
        for _ in range(iteration):
            current_temp = []
            for device in self._devices:
                one_wire_file = "{}/{}/w1_slave".format(self._device_path, device)
                with open(one_wire_file, "r") as bytestream:
                    data = [
                        item.rstrip().split(" ")[-1] for item in bytestream.readlines()
                    ]
                    if data[0] == "YES":
                        current_temp.append(self._translator(int(data[1][2:])))
                    else:
                        current_temp.append(None)
            iteration_values.append(current_temp)
            time.sleep(self._idle_time)
        if iteration == 1:
            return iteration_values[0]
        return iteration_values


    def run(self, iteration: int = 1, functor=default_functor, **kwargs):
        """
        Function to measure a certain amount of values and evaluate them directly.
        Evaluation can be a print function or a self-defined function.
        Options can be passed with **kwargs.

        Keyword arguments:
            iteration (int, optional) -- Number of measurements to be done.
                Defaults to 1.
            functor (function_ptr, optional) -- An evaluationfunction.
                Defaults to default_functor.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        while iteration != 0:
            if iteration > 0:
                iteration -= 1
            current_temp = []
            for device in self._devices:
                one_wire_file = "{}/{}/w1_slave".format(self._device_path, device)
                with open(one_wire_file, "r") as bytestream:
                    data = [
                        item.rstrip().split(" ")[-1] for item in bytestream.readlines()
                    ]
                    if data[0] == "YES":
                        current_temp.append(self._translator(int(data[1][2:])))
                    else:
                        current_temp.append(None)
            functor(current_temp, **kwargs)
            time.sleep(self._idle_time)


    def run_endless(self, functor=default_functor, **kwargs):
        """
        Function to permanently measure values and evaluate them directly.
        Evaluation can be a print function or a self-defined function.
        Options can be passed with **kwargs.

        Keyword arguments:
            functor (function_ptr, optional) -- An evaluationfunction.
                Defaults to default_functor.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        self.run(iteration=-1, functor=functor, **kwargs)


def main():
    """
    A main function that is used, when this module is used as a stand-alone script.
    Arguments can be passed and it will simply print results to std-out.
    """
    parser = argparse.ArgumentParser(
        description="A short programm to print values from DS18B20 sensor."
    )
    parser.add_argument(
        "-d", metavar="D", nargs="+", type=str, required=True,
        help="The devices 1-wire, that should be used to measure temperature."
    )
    parser.add_argument(
        "-t", metavar="T", default=2, type=int, required=False,
        help="Set idle time (break between measurements); default t = 2."
    )
    parser.add_argument(
        "-c", metavar="C", default="C", type=str, required=False,
        help="The tempearute scale to use (K,F,C); default c = C."
    )
    parser.add_argument(
        "-i", metavar="I", default=10, type=int, required=False,
        help="Number of iterations to get a value; use -1 for infinity."
    )

    args = parser.parse_args()
    devices = args.d
    iterations = -1 if args.i < 0 else args.i
    idle_time = args.t
    scale = args.c

    connector = DS18B20(
        device=devices,
        scale=scale,
        idle_time=idle_time
    )
    connector.run(iterations)


if __name__ == "__main__":
    # execute only if run as a script
    main()
#!/usr/bin/env python3
"""
This module provides a class to interact with the DS18B20 temperature
sensor and is considered mostly to be used on a Raspberry Pi. With
aprorpiate naming this class might be used for other SoC like Arduino,
Genuino etc.

This module can also be used as a standalone script to retrieve values from.
attached sensors.

Classes: DS18B20
Functions: main
"""
import os
import argparse
import time

os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')

class DS18B20:
    """
    This class provides a set of functions and methods that can be used to
    interact with the DS18B20 sensor. As values can be set/adjusted during
    runtime, there is no need to instantiate a new object for every config
    change. Multiple sensors can be handled at once by passing a list of
    devices or using None (take all available devices).

    Methods:
        default_functor(values, **kwargs)
        __init__(device, scale, idle_time)
        set_devices()
        devices()
        set_device_path()
        device_path()
        set_scale()
        scale()
        set_idle_time()
        idle_time()
        __to_celsius
        __to_kelvin
        __to_fahrenheit
        get(iteration)
        run(iteration, functor, **kwargs)
        run_endless(iteration, functor, **kwargs)
    """
    def default_functor(values: "list of ints" or int, **kwargs):
        """
        An example of how functions can be passed to the run function,
        to make use of a value handling (e.g. printing values).

        Keyword arguments:
            values (list of ints) -- the measured values for one run.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        if not isinstance(values, list):
            print(values, **kwargs)
        else:
            print("\t".join(str(value) for value in values), **kwargs)


    def __init__(self, device, scale: str = "C", idle_time: int = 2):
        """
        Constructor to instantiate an object, that is able to handle multiple
        devices at once.

        Keyword arguments:
            device (str, list of str) -- The 1-wire devices that should be used.
            scale (str, optional) -- The temperature scalce (K,F,C) to be used
            idle_time (int, optional) -- The idle time between two measurements.
        """
        self.set_scale(scale)
        self._idle_time = idle_time
        self._device_path = "/sys/bus/w1/devices"
        dev = [device] if not isinstance(device,list) else device
        self.set_devices(dev)
        


    def set_devices(self, devices: list=None):
        """
        This functions loads the required 1-wire devices.
        If no device list is provided, all available devices will be used.

        Keyword arguments:
            devices (list of str) -- the devices to be used; leave empty
                to use all devices

        """
        device_list = [devices] if not isinstance(devices, list) else devices
        # filter available 1-wire devices #
        one_wire_devices = [
            dev.name for dev in os.scandir(self._device_path)
            if dev.is_dir() and dev.name != "w1_bus_master1"
        ]

        # filter for active 1-wire devices #
        self._devices = [
            dev for dev in one_wire_devices
            if "w1_slave" in (
                item.name for item in os.scandir("{}/{}".format(self._device_path, dev))
            )
        ]

        # reduce to chosen, active devices #
        if device_list != [None]:
            self._devices = [
                device
                for device in self._devices
                if device in device_list
            ]
            if len(self._devices) != len(device_list):
                missing_devices = [
                    dev for dev in devices if dev not in self._devices
                ]
                raise FileNotFoundError(
                    "Provided devices [{}] cannot be found! Aborting.".format(
                        ", ".join(missing_devices)
                    )
                )
        if not self._devices:
            raise FileNotFoundError("No 1-wire device found!")


    def devices(self):
        """
        The function to get (by return) the current device list in use.

        Returns:
            (list) -- The device list.
        """
        return list(self._devices)


    def set_device_path(self, device_path: str):
        """
        Sets the device path, were 1-wire devices can be found.
        Usually it is '/sys/bus/w1/devices'

        Keyword arguments:
            device_path (str) -- The device path, e.g. /sys/bus/w1/devices
        """
        self._device_path = device_path


    def device_path(self):
        """
        The function to get (by return) the current device_path, where
        1-wire devices should be found.

        Returns:
            (str) -- The device_path as a string.
        """
        return self._device_path


    def set_scale(self, scale: str):
        """
        Sets the temperature scale, that should be used; it is either
        'K', 'F' or 'C'; relates to Kevlin, Fahrenheit or Celsius.
        Returned values are to be read with that scale; default is C.

        Keyword arguments:
            scale (str) -- The temperature scale: 'K', 'F' or 'C'.
        """
        if scale not in ("K", "F", "C"):
            raise ValueError(
                "Unknown scale type '{}'! Needs to be K, F or C!".format(scale)
            )
        self._scale = scale
        if self._scale == "C":
            self._translator = self.__to_celsius
        elif self._scale == "F":
            # Fahrenheit temperature scale #
            self._translator = self.__to_fahrenheit
        else:
            # Kelvin temperature scale #
            self._translator = self.__to_kelvin


    def scale(self):
        """
        The function to get (by return) the current used temperature scale.
        This is either 'K', 'F' or 'C'.

        Returns:
            (str) -- The uses temperature scale as a string.
        """
        return self._scale


    def set_idle_time(self, idle_time: int):
        """
        Sets the idle time that is used in between two
        measurements when using run(...) or run_endless(...)

        Keyword arguments:
            idle_time (int) -- The idle time.
        """
        self._idle_time = idle_time


    def idle_time(self):
        """
        The function to get (by return) the current used idle time.

        Returns:
            (int) -- The currently used idle time.
        """
        return self._idle_time


    def __to_celsius(self, value: int):
        """
        Transforms measured value into Celcius temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return value / 1000.0


    def __to_kelvin(self, value: int):
        """
        Transforms measured value into Kelvin temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return (value/1000) + 273.15


    def __to_fahrenheit(self, value: int):
        """
        Transforms measured value into Fahrenheit temperature scale.

        Keyword arguments:
            value (int) -- A measured value that should be transformed.

        Returns:
            (int) -- The transformed input value.
        """
        return (value/1000)*(9/5) + 32


    def get(self, iteration: int = 1):
        """
        Function to get a certain amount of measured values; there is no
        on-line handling. Values will be measured and returned.

        Keyword arguments:
            iteration (int, optional) -- Measured values amount. Defaults to 1.

        Returns:
            (list of list of ints or list of ints) -- The measured values
        """
        iteration_values = []
        for _ in range(iteration):
            current_temp = []
            for device in self._devices:
                one_wire_file = "{}/{}/w1_slave".format(self._device_path, device)
                with open(one_wire_file, "r") as bytestream:
                    data = [
                        item.rstrip().split(" ")[-1] for item in bytestream.readlines()
                    ]
                    if data[0] == "YES":
                        current_temp.append(self._translator(int(data[1][2:])))
                    else:
                        current_temp.append(None)
            iteration_values.append(current_temp)
            time.sleep(self._idle_time)
        if iteration == 1:
            return iteration_values[0]
        return iteration_values


    def run(self, iteration: int = 1, functor=default_functor, **kwargs):
        """
        Function to measure a certain amount of values and evaluate them directly.
        Evaluation can be a print function or a self-defined function.
        Options can be passed with **kwargs.

        Keyword arguments:
            iteration (int, optional) -- Number of measurements to be done.
                Defaults to 1.
            functor (function_ptr, optional) -- An evaluationfunction.
                Defaults to default_functor.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        while iteration != 0:
            if iteration > 0:
                iteration -= 1
            current_temp = []
            for device in self._devices:
                one_wire_file = "{}/{}/w1_slave".format(self._device_path, device)
                with open(one_wire_file, "r") as bytestream:
                    data = [
                        item.rstrip().split(" ")[-1] for item in bytestream.readlines()
                    ]
                    if data[0] == "YES":
                        current_temp.append(self._translator(int(data[1][2:])))
                    else:
                        current_temp.append(None)
            functor(current_temp, **kwargs)
            time.sleep(self._idle_time)


    def run_endless(self, functor=default_functor, **kwargs):
        """
        Function to permanently measure values and evaluate them directly.
        Evaluation can be a print function or a self-defined function.
        Options can be passed with **kwargs.

        Keyword arguments:
            functor (function_ptr, optional) -- An evaluationfunction.
                Defaults to default_functor.
        Args:
            **kwargs -- arguments that can be evaluated in another function.
        """
        self.run(iteration=-1, functor=functor, **kwargs)


def main():
    """
    A main function that is used, when this module is used as a stand-alone script.
    Arguments can be passed and it will simply print results to std-out.
    """
    parser = argparse.ArgumentParser(
        description="A short programm to print values from DS18B20 sensor."
    )
    parser.add_argument(
        "-d", metavar="D", nargs="+", type=str, required=True,
        help="The devices 1-wire, that should be used to measure temperature."
    )
    parser.add_argument(
        "-t", metavar="T", default=2, type=int, required=False,
        help="Set idle time (break between measurements); default t = 2."
    )
    parser.add_argument(
        "-c", metavar="C", default="C", type=str, required=False,
        help="The tempearute scale to use (K,F,C); default c = C."
    )
    parser.add_argument(
        "-i", metavar="I", default=10, type=int, required=False,
        help="Number of iterations to get a value; use -1 for infinity."
    )

    args = parser.parse_args()
    devices = args.d
    iterations = -1 if args.i < 0 else args.i
    idle_time = args.t
    scale = args.c

    connector = DS18B20(
        device=devices,
        scale=scale,
        idle_time=idle_time
    )
    connector.run(iterations)


if __name__ == "__main__":
    # execute only if run as a script
    main()
#!/usr/bin/env python3 This module provides a class to interact with the DS18B20 temperature sensor and is considered mostly to be used on a Raspberry Pi. With aprorpiate naming this class might be used for other SoC like Arduino, Genuino etc.  This module can also be used as a standalone script to retrieve values from. attached sensors.  Classes: DS18B20 Functions: main This class provides a set of functions and methods that can be used to     interact with the DS18B20 sensor. As values can be set/adjusted during     runtime, there is no need to instantiate a new object for every config     change. Multiple sensors can be handled at once by passing a list of     devices or using None (take all available devices).      Methods:         default_functor(values, **kwargs)         __init__(device, scale, idle_time)         set_devices()         devices()         set_device_path()         device_path()         set_scale()         scale()         set_idle_time()         idle_time()         __to_celsius         __to_kelvin         __to_fahrenheit         get(iteration)         run(iteration, functor, **kwargs)         run_endless(iteration, functor, **kwargs) An example of how functions can be passed to the run function,         to make use of a value handling (e.g. printing values).          Keyword arguments:             values (list of ints) -- the measured values for one run.         Args:             **kwargs -- arguments that can be evaluated in another function. Constructor to instantiate an object, that is able to handle multiple         devices at once.          Keyword arguments:             device (str, list of str) -- The 1-wire devices that should be used.             scale (str, optional) -- The temperature scalce (K,F,C) to be used             idle_time (int, optional) -- The idle time between two measurements. This functions loads the required 1-wire devices.         If no device list is provided, all available devices will be used.          Keyword arguments:             devices (list of str) -- the devices to be used; leave empty                 to use all devices # filter available 1-wire devices # # filter for active 1-wire devices # # reduce to chosen, active devices # The function to get (by return) the current device list in use.          Returns:             (list) -- The device list. Sets the device path, were 1-wire devices can be found.         Usually it is '/sys/bus/w1/devices'          Keyword arguments:             device_path (str) -- The device path, e.g. /sys/bus/w1/devices The function to get (by return) the current device_path, where         1-wire devices should be found.          Returns:             (str) -- The device_path as a string. Sets the temperature scale, that should be used; it is either         'K', 'F' or 'C'; relates to Kevlin, Fahrenheit or Celsius.         Returned values are to be read with that scale; default is C.          Keyword arguments:             scale (str) -- The temperature scale: 'K', 'F' or 'C'. # Fahrenheit temperature scale # # Kelvin temperature scale # The function to get (by return) the current used temperature scale.         This is either 'K', 'F' or 'C'.          Returns:             (str) -- The uses temperature scale as a string. Sets the idle time that is used in between two         measurements when using run(...) or run_endless(...)          Keyword arguments:             idle_time (int) -- The idle time. The function to get (by return) the current used idle time.          Returns:             (int) -- The currently used idle time. Transforms measured value into Celcius temperature scale.          Keyword arguments:             value (int) -- A measured value that should be transformed.          Returns:             (int) -- The transformed input value. Transforms measured value into Kelvin temperature scale.          Keyword arguments:             value (int) -- A measured value that should be transformed.          Returns:             (int) -- The transformed input value. Transforms measured value into Fahrenheit temperature scale.          Keyword arguments:             value (int) -- A measured value that should be transformed.          Returns:             (int) -- The transformed input value. Function to get a certain amount of measured values; there is no         on-line handling. Values will be measured and returned.          Keyword arguments:             iteration (int, optional) -- Measured values amount. Defaults to 1.          Returns:             (list of list of ints or list of ints) -- The measured values Function to measure a certain amount of values and evaluate them directly.         Evaluation can be a print function or a self-defined function.         Options can be passed with **kwargs.          Keyword arguments:             iteration (int, optional) -- Number of measurements to be done.                 Defaults to 1.             functor (function_ptr, optional) -- An evaluationfunction.                 Defaults to default_functor.         Args:             **kwargs -- arguments that can be evaluated in another function. Function to permanently measure values and evaluate them directly.         Evaluation can be a print function or a self-defined function.         Options can be passed with **kwargs.          Keyword arguments:             functor (function_ptr, optional) -- An evaluationfunction.                 Defaults to default_functor.         Args:             **kwargs -- arguments that can be evaluated in another function. A main function that is used, when this module is used as a stand-alone script.     Arguments can be passed and it will simply print results to std-out. # execute only if run as a script