Config Reference

Source code reference for the collector configuration code.

collector.src.config.datastructures.DeviceConfig

Configuration object which describes a type of device that the collector node runs on.

Attributes:

Name	Type	Description
model	str	A string representing the type of device, for example "Raspberry Pi 3."
requires_micropython	bool	If true, the program will not enable a start unless it is being run in a micropython environment.

Source code in collector/src/config/datastructures.py

class DeviceConfig:
    """
    Configuration object which describes a type of device
    that the collector node runs on.

    Attributes:
        model: A string representing the type of device,
            for example "Raspberry Pi 3."
        requires_micropython: If true, the program will
            not enable a start unless it is being run in a
            micropython environment.
    """

    model: str
    requires_micropython: bool

    def __init__(self, config: dict[str, Any], micropython: bool):
        """
        Initializes the config and ensures the required attributes are present.

        Args:
            config: The device config object, ie. the
                dictionary contained within the "device" key in the config file.
            micropython: Whether the current environment is micropython
        """
        self.model = get_attribute(config, "model", "device")
        self.requires_micropython = get_attribute(
            config, "requires_micropython", "device"
        )

        if self.requires_micropython and not micropython:
            raise ValueError(
                f"The current device {self.model} requires micropython but the environment is not micropython."
            )

__init__(config, micropython)

Initializes the config and ensures the required attributes are present.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	The device config object, ie. the dictionary contained within the "device" key in the config file.	required
micropython	bool	Whether the current environment is micropython	required

Source code in collector/src/config/datastructures.py

def __init__(self, config: dict[str, Any], micropython: bool):
    """
    Initializes the config and ensures the required attributes are present.

    Args:
        config: The device config object, ie. the
            dictionary contained within the "device" key in the config file.
        micropython: Whether the current environment is micropython
    """
    self.model = get_attribute(config, "model", "device")
    self.requires_micropython = get_attribute(
        config, "requires_micropython", "device"
    )

    if self.requires_micropython and not micropython:
        raise ValueError(
            f"The current device {self.model} requires micropython but the environment is not micropython."
        )

collector.src.config.datastructures.UploadConfig

Configuration object which describes a QuestDB target.

Attributes:

Name	Type	Description
host	str	The database URL.
port	str	The database port.
user	str	The database username.
password	str	The database password.

Source code in collector/src/config/datastructures.py

class UploadConfig:
    """
    Configuration object which describes a QuestDB target.

    Attributes:
        host: The database URL.
        port: The database port.
        user: The database username.
        password: The database password.
    """

    host: str
    port: str
    user: str
    password: str

    def __init__(self, config: dict[str, Any]):
        """
        Initializes the config and ensures the required attributes are present.

        Args:
            config: The upload config object, ie. the
                dictionary contained within the "upload" key in the config file.
        """
        self.host = get_attribute(config, "host", "upload")
        self.port = get_attribute(config, "port", "upload")
        self.user = get_attribute(config, "user", "upload")
        self.password = get_attribute(config, "password", "upload")

__init__(config)

Initializes the config and ensures the required attributes are present.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	The upload config object, ie. the dictionary contained within the "upload" key in the config file.	required

Source code in collector/src/config/datastructures.py

def __init__(self, config: dict[str, Any]):
    """
    Initializes the config and ensures the required attributes are present.

    Args:
        config: The upload config object, ie. the
            dictionary contained within the "upload" key in the config file.
    """
    self.host = get_attribute(config, "host", "upload")
    self.port = get_attribute(config, "port", "upload")
    self.user = get_attribute(config, "user", "upload")
    self.password = get_attribute(config, "password", "upload")

collector.src.config.datastructures.SensorConfig

Configuration object which describes a sensor.

Attributes:

Name	Type	Description
sensor_id	str	An ID associated with the sensor.
sensor_code	SensorCodeEnum	Associated with a specific implementation of AbstractSensorDriver.
name	str	A descriptive name for the sensor.
gpio	dict[str, int]	A dictionary where the values are GPIO pins which the sensor requires use of, and the keys are string identifiers. Used to validate that the configured GPIO pins match the allowed_gpio attribute of the device config.
attributes	dict[str, Any] | None	Any other attributes which are required to use the sensor.

Source code in collector/src/config/datastructures.py

class SensorConfig:
    """
    Configuration object which describes a sensor.

    Attributes:
        sensor_id: An ID associated with the sensor.
        sensor_code: Associated with
            a specific implementation of AbstractSensorDriver.
        name: A descriptive name for the sensor.
        gpio: A dictionary where the values are
            GPIO pins which the sensor requires use of, and
            the keys are string identifiers. Used to validate
            that the configured GPIO pins match the allowed_gpio
            attribute of the device config.
        attributes: Any other attributes
            which are required to use the sensor.
    """

    sensor_id: str
    sensor_code: SensorCodeEnum
    name: str
    gpio: dict[str, int]
    attributes: dict[str, Any] | None

    def __init__(
        self,
        config: dict[str, Any],
        index: int,
    ):
        """
        Initializes the config and ensures the required attributes are present.

        Args:
            config: The sensor config object, ie. a
                dictionary contained within the "sensors" array in the config file.
            index: The index of the sensor config within the "sensors" array.
            allowed_gpio: The allowed GPIO pins as configured in the device config.
        """
        self.sensor_id = get_attribute(config, "sensor_id", f"sensor[{index}]")
        self.sensor_code = SensorCodeEnum[
            get_attribute(config, "sensor_code", f"sensor[{index}]")
        ]
        self.name = get_attribute(config, "name", f"sensor[{index}]")
        self.gpio = get_attribute(config, "gpio", f"sensor[{index}]")
        self.attributes = get_attribute(
            config, "attributes", f"sensor[{index}]", required=False
        )

__init__(config, index)

Initializes the config and ensures the required attributes are present.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	The sensor config object, ie. a dictionary contained within the "sensors" array in the config file.	required
index	int	The index of the sensor config within the "sensors" array.	required
allowed_gpio		The allowed GPIO pins as configured in the device config.	required

Source code in collector/src/config/datastructures.py

def __init__(
    self,
    config: dict[str, Any],
    index: int,
):
    """
    Initializes the config and ensures the required attributes are present.

    Args:
        config: The sensor config object, ie. a
            dictionary contained within the "sensors" array in the config file.
        index: The index of the sensor config within the "sensors" array.
        allowed_gpio: The allowed GPIO pins as configured in the device config.
    """
    self.sensor_id = get_attribute(config, "sensor_id", f"sensor[{index}]")
    self.sensor_code = SensorCodeEnum[
        get_attribute(config, "sensor_code", f"sensor[{index}]")
    ]
    self.name = get_attribute(config, "name", f"sensor[{index}]")
    self.gpio = get_attribute(config, "gpio", f"sensor[{index}]")
    self.attributes = get_attribute(
        config, "attributes", f"sensor[{index}]", required=False
    )

collector.src.config.datastructures.CollectorConfig

Configuration object which describes the collector node device.

Attributes:

Name	Type	Description
collector_id	str	An ID associated with the node.
node_name	str	A name associated with the node.
ssid_name	str | None	If defined, the device will attempt to connect to this SSID.
ssid_password	str | None	If defined, the device will attempt to connect to this SSID.
polling_interval	int	The number of miliseconds to wait between sensor polls. Must not be negative.
device	DeviceConfig	The device config.
upload	UploadConfig	The configured upload target.
sensors	list[SensorConfig]	All configured sensors.

Source code in collector/src/config/datastructures.py

class CollectorConfig:
    """
    Configuration object which describes the collector node device.

    Attributes:
        collector_id: An ID associated with the node.
        node_name: A name associated with the node.
        ssid_name: If defined, the device will attempt to connect
            to this SSID.
        ssid_password: If defined, the device will attempt to connect
            to this SSID.
        polling_interval: The number of miliseconds to wait
            between sensor polls. Must not be negative.
        device: The device config.
        upload: The configured upload target.
        sensors: All configured sensors.
    """

    collector_id: str
    node_name: str
    polling_interval: int
    ssid_name: str | None
    ssid_password: str | None
    device: DeviceConfig
    upload: UploadConfig
    sensors: list[SensorConfig]

    def __init__(
        self,
        config: dict[str, Any],
        device: DeviceConfig,
        upload: UploadConfig,
        sensors: list[SensorConfig],
    ):
        """
        Initializes the config and ensures the required attributes are present.

        Args:
            config: The config object, ie. the
                dictionary pulled from the config file.
            device: The parsed device config.
            upload: The parsed upload config.
            sensors: The parsed sensor configs.

        Raises:
            ValueError: Raised if the polling interval is a negative number.
        """
        self.collector_id = get_attribute(config, "collector_id", "config")
        self.node_name = get_attribute(config, "node_name", "config")
        self.polling_interval = get_attribute(config, "polling_interval", "config")
        self.ssid_name = get_attribute(config, "ssid_name", "config", required=False)
        self.ssid_password = get_attribute(
            config, "ssid_password", "config", required=False
        )
        self.device = device
        self.upload = upload
        self.sensors = sensors

        # Validate that polling_interval is not negative.
        if self.polling_interval < 0:
            raise ValueError("Cannot set a negative polling interval.")

__init__(config, device, upload, sensors)

Initializes the config and ensures the required attributes are present.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	The config object, ie. the dictionary pulled from the config file.	required
device	DeviceConfig	The parsed device config.	required
upload	UploadConfig	The parsed upload config.	required
sensors	list[SensorConfig]	The parsed sensor configs.	required

Raises:

Type	Description
ValueError	Raised if the polling interval is a negative number.

Source code in collector/src/config/datastructures.py

def __init__(
    self,
    config: dict[str, Any],
    device: DeviceConfig,
    upload: UploadConfig,
    sensors: list[SensorConfig],
):
    """
    Initializes the config and ensures the required attributes are present.

    Args:
        config: The config object, ie. the
            dictionary pulled from the config file.
        device: The parsed device config.
        upload: The parsed upload config.
        sensors: The parsed sensor configs.

    Raises:
        ValueError: Raised if the polling interval is a negative number.
    """
    self.collector_id = get_attribute(config, "collector_id", "config")
    self.node_name = get_attribute(config, "node_name", "config")
    self.polling_interval = get_attribute(config, "polling_interval", "config")
    self.ssid_name = get_attribute(config, "ssid_name", "config", required=False)
    self.ssid_password = get_attribute(
        config, "ssid_password", "config", required=False
    )
    self.device = device
    self.upload = upload
    self.sensors = sensors

    # Validate that polling_interval is not negative.
    if self.polling_interval < 0:
        raise ValueError("Cannot set a negative polling interval.")

collector.src.config.manager.ConfigManager

Parses and stores all configured paramteres for a collector node.

Attributes:

Name	Type	Description
micropython	bool	If true, the program is being run within a micropython environment.
_config	CollectorConfig | None	The config, or None if it has not been successfully parsed.

Source code in collector/src/config/manager.py

class ConfigManager:
    """
    Parses and stores all configured paramteres for a collector node.

    Attributes:
        micropython: If true, the program is being run within
            a micropython environment.
        _config: The config, or None if it
            has not been successfully parsed.
    """

    micropython: bool = True
    _config: CollectorConfig | None = None

    def __init__(self, micropython: bool):
        """
        Initialize the config manager.

        Args:
            micropython: If true, the program is being run within
                a micropython environment.
        """
        self.micropython = micropython
        self.initialize_config()

    def initialize_config(self):
        """
        Initializes config objects for all
        configuration parameters contained within the
        configuration file and sets the config on the class.

        Raises:
            ValueError: Raised if any config is missing.
        """
        # Load config into a dictionary.
        with open("config.json") as f:
            config: dict[str, Any] = json.load(f)

        # Initialize all configs

        # Device config
        if "device" not in config:
            raise ValueError("Config missing device config.")
        device_config = DeviceConfig(config["device"], self.micropython)

        # Upload config
        if "upload" not in config:
            raise ValueError("Config missing upload config.")
        upload_config = UploadConfig(config["upload"])

        # Sensor configs
        if "sensors" not in config:
            raise ValueError("Config missing sensors config.")

        sensors: list[SensorConfig] = []
        for config, index in config["sensors"]:
            sensor = SensorConfig(config, index)
            sensors.append(sensor)

        # Collector config
        collector_config = CollectorConfig(
            config, device_config, upload_config, sensors
        )

        self._config = collector_config

    @property
    def config(self) -> CollectorConfig:
        """
        Retrieves the device config.

        Returns:
            The current device config.
        """
        if self._config is None:
            self.initialize_config()
            if self._config is None:
                raise Exception("Failed to load device config.")

        return self._config

    @property
    def metadata(self) -> tuple[CollectorMetadata, list[SensorMetadata]]:
        """
        Packages the configuration into the metadata objects.

        Returns:
            The metadata.
        """
        collector_metadata = CollectorMetadata(
            collector_id=self.config.collector_id,
            collector_name=self.config.node_name,
            device_model=self.config.device.model,
            polling_interval=self.config.polling_interval,
        )
        sensor_metadata = []
        for sensor in self.config.sensors:
            sensor_metadata.append(
                SensorMetadata(
                    collector_id=self.config.collector_id,
                    sensor_id=sensor.sensor_id,
                    sensor_code=sensor.sensor_code,
                    sensor_name=sensor.name,
                )
            )
        return collector_metadata, sensor_metadata

config property

Retrieves the device config.

Returns:

Type	Description
CollectorConfig	The current device config.

metadata property

Packages the configuration into the metadata objects.

Returns:

Type	Description
tuple[CollectorMetadata, list[SensorMetadata]]	The metadata.

__init__(micropython)

Initialize the config manager.

Parameters:

Name	Type	Description	Default
micropython	bool	If true, the program is being run within a micropython environment.	required

Source code in collector/src/config/manager.py

def __init__(self, micropython: bool):
    """
    Initialize the config manager.

    Args:
        micropython: If true, the program is being run within
            a micropython environment.
    """
    self.micropython = micropython
    self.initialize_config()

initialize_config()

Initializes config objects for all configuration parameters contained within the configuration file and sets the config on the class.

Raises:

Type	Description
ValueError	Raised if any config is missing.

Source code in collector/src/config/manager.py

def initialize_config(self):
    """
    Initializes config objects for all
    configuration parameters contained within the
    configuration file and sets the config on the class.

    Raises:
        ValueError: Raised if any config is missing.
    """
    # Load config into a dictionary.
    with open("config.json") as f:
        config: dict[str, Any] = json.load(f)

    # Initialize all configs

    # Device config
    if "device" not in config:
        raise ValueError("Config missing device config.")
    device_config = DeviceConfig(config["device"], self.micropython)

    # Upload config
    if "upload" not in config:
        raise ValueError("Config missing upload config.")
    upload_config = UploadConfig(config["upload"])

    # Sensor configs
    if "sensors" not in config:
        raise ValueError("Config missing sensors config.")

    sensors: list[SensorConfig] = []
    for config, index in config["sensors"]:
        sensor = SensorConfig(config, index)
        sensors.append(sensor)

    # Collector config
    collector_config = CollectorConfig(
        config, device_config, upload_config, sensors
    )

    self._config = collector_config

collector.src.config.utils.get_attribute(config, attribute, config_name, required=True)

Retrieves the attribute from the config, or raises an error if it does not exist and the attribute is required.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	The config to retrieve from.	required
attribute	str	The name of the attribute.	required
config_name	str	The name of the config.	required
required	bool	If true, an exception will be raised if the attribute doesn't exist.	True

Raises:

Type	Description
ValueError	Raised if the attribute does not exist within the config and is required.

Returns:

Type	Description
Any	The config attribute, or None if it does not exist and is not required.

Source code in collector/src/config/utils.py

def get_attribute(
    config: dict[str, Any], attribute: str, config_name: str, required: bool = True
) -> Any:
    """
    Retrieves the attribute from the config, or
    raises an error if it does not exist and the attribute is required.

    Args:
        config: The config to retrieve from.
        attribute: The name of the attribute.
        config_name: The name of the config.
        required: If true, an exception will be raised
            if the attribute doesn't exist.

    Raises:
        ValueError: Raised if the attribute does not
            exist within the config and is required.

    Returns:
        The config attribute, or None if it does not exist and is not required.
    """
    try:
        value = config["attribute"]
    except KeyError:
        if required:
            raise ValueError(
                f"Config '{config_name}' missing required parameter '{attribute}'."
            )
        else:
            return None

    if value is None and required:
        raise ValueError(
            f"Config '{config_name}' missing required parameter '{attribute}'."
        )

    return value