Shared Module

allsky_shared

allsky_shared.py

Part of allsky postprocess.py modules. https://github.com/AllskyTeam/allsky

This module is a common dumping ground for shared variables and functions used by various Allsky components.

It provides helpers for:

• Reading environment variables and Allsky configuration
• Managing small on-disk "databases" used as debug stores
• Filesystem utilities (paths, permissions, extra-data files)
• Database connection helpers and automatic purge routines
• Overlay "extra data" JSON formatting and persistence

convert_lat_lon(input)

Helper for converting latitude/longitude strings.

This wrapper calls the legacy :func:convertLatLon. New code should use this snake_case name; the camelCase function is kept so existing callers continue to work.

See :func:convertLatLon for details.

Source code in scripts/modules/allsky_shared.py

def convert_lat_lon(input):
    """
    Helper for converting latitude/longitude strings.

    This wrapper calls the legacy :func:`convertLatLon`. New code should
    use this snake_case name; the camelCase function is kept so existing
    callers continue to work.

    See :func:`convertLatLon` for details.
    """
    return convertLatLon(input)

count_starts_in_image(image, mask_file_name=None)

Detect stars in an image using Photutils' DAOStarFinder.

The image is converted to grayscale if needed, optionally masked with :func:mask_image, and then processed with sigma-clipped statistics and DAOStarFinder to locate star centroids.

Parameters:

Name	Type	Description	Default
image	ndarray	Input image (grayscale or BGR).	required
mask_file_name	str | None	Optional mask file name to apply before detection.	None

Returns:

Type	Description
	tuple[list[tuple[float, float]], numpy.ndarray]: A tuple containing: A list of (x, y) star coordinates. The (possibly masked) image used for detection.

Source code in scripts/modules/allsky_shared.py

def count_starts_in_image(image, mask_file_name=None):
    """
    Detect stars in an image using Photutils' DAOStarFinder.

    The image is converted to grayscale if needed, optionally masked with
    :func:`mask_image`, and then processed with sigma-clipped statistics
    and DAOStarFinder to locate star centroids.

    Args:
        image (numpy.ndarray):
            Input image (grayscale or BGR).
        mask_file_name (str | None, optional):
            Optional mask file name to apply before detection.

    Returns:
        tuple[list[tuple[float, float]], numpy.ndarray]:
            A tuple containing:

              * A list of ``(x, y)`` star coordinates.
              * The (possibly masked) image used for detection.
    """
    from photutils.detection import DAOStarFinder
    from astropy.stats import sigma_clipped_stats
    import cv2

    # Convert to grayscale if it's RGB
    if image.ndim == 3:
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    else:
        gray = image

    if mask_file_name is not None and mask_file_name != '':
        gray = mask_image(gray, mask_file_name)

    # Convert to float for processing
    image_data = gray.astype(float)

    # Estimate background stats
    mean, median, std = sigma_clipped_stats(image_data, sigma=3.0)

    # Detect stars
    daofind = DAOStarFinder(fwhm=3.0, threshold=5.0 * std)
    sources = daofind(image_data - median)

    # Convert to list of (x, y) tuples if sources were found
    coords = []
    if sources is not None and len(sources) > 0:
        x = sources['xcentroid'].tolist()
        y = sources['ycentroid'].tolist()
        coords = list(zip(x, y))

    return coords, image

create_cardinal(degrees)

Convert a wind direction in degrees into a cardinal point.

Parameters:

Name	Type	Description	Default
degrees		Direction in degrees (0–360). North is 0°/360°, east is 90°, and so on.	required

Returns:

Type	Description
	A string containing one of the 16-point compass directions
	(e.g., 'N', 'NE', 'SW'), or 'N/A' if the input
	cannot be interpreted.

Source code in scripts/modules/allsky_shared.py

def create_cardinal(degrees):
    """
    Convert a wind direction in degrees into a cardinal point.

    Args:
        degrees: Direction in degrees (0–360). North is 0°/360°, east is
            90°, and so on.

    Returns:
        A string containing one of the 16-point compass directions
        (e.g., ``'N'``, ``'NE'``, ``'SW'``), or ``'N/A'`` if the input
        cannot be interpreted.
    """
    try:
        cardinals = ['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW','W', 'WNW', 'NW', 'NNW', 'N']
        cardinal = cardinals[round(degrees / 22.5)]
    except Exception:
        cardinal = 'N/A'

    return cardinal

create_device(import_name, class_name, bus_number, i2c_address='')

Instantiate an I2C device class on a given bus.

The device class is imported dynamically and initialised with an Adafruit Blinka busio.I2C object constructed from a known set of SCL/SDA pins for the requested bus number.

Parameters:

Name	Type	Description	Default
import_name	str	Module path to import (e.g. "adafruit_bme280").	required
class_name	str	Name of the device class in that module.	required
bus_number	int	I2C bus number (e.g. 1, 3, 4, 5, 6).	required
i2c_address	str	Optional I2C address string (e.g. "0x76"). If omitted, the device class is constructed without an explicit address.	''

Returns:

Name	Type	Description
Any		An instance of the requested device class.

Raises:

Type	Description
ImportError	If the module or class cannot be imported.
ValueError	If no pin mapping exists for the given bus number.

Source code in scripts/modules/allsky_shared.py

def create_device(import_name: str, class_name: str, bus_number: int, i2c_address: str = ""):
    """
    Instantiate an I2C device class on a given bus.

    The device class is imported dynamically and initialised with an
    Adafruit Blinka ``busio.I2C`` object constructed from a known set of
    SCL/SDA pins for the requested bus number.

    Args:
        import_name (str):
            Module path to import (e.g. ``"adafruit_bme280"``).
        class_name (str):
            Name of the device class in that module.
        bus_number (int):
            I2C bus number (e.g. 1, 3, 4, 5, 6).
        i2c_address (str, optional):
            Optional I2C address string (e.g. ``"0x76"``). If omitted, the
            device class is constructed without an explicit address.

    Returns:
        Any:
            An instance of the requested device class.

    Raises:
        ImportError:
            If the module or class cannot be imported.
        ValueError:
            If no pin mapping exists for the given bus number.
    """
    bus_number = int(bus_number)

    # Define SCL/SDA pins for each bus
    I2C_BUS_PINS = {
        1: (board.SCL, board.SDA),
        3: (board.D5, board.D4),
        4: (board.D9, board.D8),
        5: (board.D13, board.D12),
        6: (board.D23, board.D22)
    }

    # Dynamically import the module and get the class
    try:
        module = importlib.import_module(import_name)
        cls = getattr(module, class_name)
    except (ImportError, AttributeError) as e:
        raise ImportError(f"Could not import '{class_name}' from '{import_name}': {e}")

    try:
        scl, sda = I2C_BUS_PINS[bus_number]
    except KeyError:
        raise ValueError(f"No pin mapping defined for I2C bus {bus_number}")

    i2c = busio.I2C(scl, sda)

    # Instantiate device
    if i2c_address:
        return cls(i2c, int(i2c_address, 0))
    else:
        return cls(i2c)

delete_extra_data(fileName)

Preferred wrapper for removing extra data files.

This is the underscore version and should be used in new code. It simply delegates to :func:deleteExtraData, which contains the legacy implementation.

Parameters:

Name	Type	Description	Default
fileName	str	File name to remove from all configured extra data directories.	required

Source code in scripts/modules/allsky_shared.py

def delete_extra_data(fileName):
    """
    Preferred wrapper for removing extra data files.

    This is the underscore version and should be used in new code. It simply
    delegates to :func:`deleteExtraData`, which contains the legacy
    implementation.

    Args:
        fileName (str):
            File name to remove from all configured extra data directories.
    """
    deleteExtraData(fileName)

get_all_allsky_variables(show_empty=True, module='', indexed=False, raw=False)

Retrieve all known Allsky variables via the ALLSKYVARIABLES helper.

Parameters:

Name	Type	Description	Default
show_empty	bool	Whether to include variables that have no value. Defaults to True.	True
module	str	Optional module filter, depending on the ALLSKYVARIABLES implementation.	''
indexed	bool	If True, variables may be returned in an indexed form.	False
raw	bool	If True, return raw data structures from ALLSKYVARIABLES.	False

Returns:

Name	Type	Description
Any		Whatever is returned by ALLSKYVARIABLES().get_variables(...).

Source code in scripts/modules/allsky_shared.py

def get_all_allsky_variables(show_empty=True, module='', indexed=False, raw=False):
    """
    Retrieve all known Allsky variables via the ALLSKYVARIABLES helper.

    Args:
        show_empty (bool, optional):
            Whether to include variables that have no value. Defaults to True.
        module (str, optional):
            Optional module filter, depending on the ALLSKYVARIABLES
            implementation.
        indexed (bool, optional):
            If True, variables may be returned in an indexed form.
        raw (bool, optional):
            If True, return raw data structures from ALLSKYVARIABLES.

    Returns:
        Any:
            Whatever is returned by ``ALLSKYVARIABLES().get_variables(...)``.
    """
    allskyvariables = ALLSKYVARIABLES()
    return allskyvariables.get_variables(show_empty, module, indexed, raw)

get_allsky_variable(variable)

Look up a single Allsky variable from environment or extra-data files.

The lookup order is:

1. Environment via :func:getEnvironmentVariable.
2. JSON extra-data files in all extra-data directories.
3. Text extra-data files in all extra-data directories.

Parameters:

Name	Type	Description	Default
variable	str	The variable name to retrieve.	required

Returns:

Name	Type	Description
Any		The variable value if found, otherwise None.

Source code in scripts/modules/allsky_shared.py

def get_allsky_variable(variable):
    """
    Look up a single Allsky variable from environment or extra-data files.

    The lookup order is:

      1. Environment via :func:`getEnvironmentVariable`.
      2. JSON extra-data files in all extra-data directories.
      3. Text extra-data files in all extra-data directories.

    Args:
        variable (str):
            The variable name to retrieve.

    Returns:
        Any:
            The variable value if found, otherwise None.
    """
    result = getEnvironmentVariable(variable)
    if result is None:

        extra_data_paths = get_extra_dir()
        for extra_data_path in extra_data_paths:
            directory = Path(extra_data_path)

            for file_path in directory.iterdir():
                if file_path.is_file() and isFileReadable(file_path):

                    file_extension = Path(file_path).suffix

                    if file_extension == '.json':
                        result = _get_value_from_json_file(file_path, variable)

                    if file_extension == '.txt':
                        result = _get_value_from_text_file(file_path, variable)

                if result is not None:
                    break

    return result

get_allsky_version()

Convenience helper to retrieve and parse the Allsky version.

The version file path is taken from the ALLSKY_VERSION_FILE environment variable and passed to :func:parse_version.

Returns:

Name	Type	Description
dict		Parsed version info as returned by :func:parse_version.

Source code in scripts/modules/allsky_shared.py

def get_allsky_version():
    """
    Convenience helper to retrieve and parse the Allsky version.

    The version file path is taken from the ``ALLSKY_VERSION_FILE``
    environment variable and passed to :func:`parse_version`.

    Returns:
        dict:
            Parsed version info as returned by :func:`parse_version`.
    """
    version_file = os.environ['ALLSKY_VERSION_FILE']
    version_info = parse_version(version_file)

    return version_info

get_api_url()

Resolve the Allsky API base URL from the environment.

If ALLSKY_API_URL is not present in the current environment, this helper calls :func:setupForCommandLine to load variables from the usual variables.json file, and then re-reads the environment.

Returns:

Name	Type	Description
str		The API base URL from ALLSKY_API_URL.

Source code in scripts/modules/allsky_shared.py

def get_api_url():
    """
    Resolve the Allsky API base URL from the environment.

    If ``ALLSKY_API_URL`` is not present in the current environment, this
    helper calls :func:`setupForCommandLine` to load variables from the
    usual ``variables.json`` file, and then re-reads the environment.

    Returns:
        str:
            The API base URL from ``ALLSKY_API_URL``.
    """
    try:
        api_url = os.environ['ALLSKY_API_URL']
    except KeyError:
        setupForCommandLine()
        api_url = os.environ['ALLSKY_API_URL']

    return api_url

get_camera_gain()

Get the current camera gain.

For Raspberry Pi cameras, this reads the AnalogueGain value from the Pi metadata. For other camera types it uses the AS_GAIN environment variable.

If no gain can be determined, 0.0 is returned.

Returns:

Name	Type	Description
float		Camera gain value.

Source code in scripts/modules/allsky_shared.py

def get_camera_gain():
    """
    Get the current camera gain.

    For Raspberry Pi cameras, this reads the ``AnalogueGain`` value from the
    Pi metadata. For other camera types it uses the ``AS_GAIN`` environment
    variable.

    If no gain can be determined, 0.0 is returned.

    Returns:
        float:
            Camera gain value.
    """
    gain = 0
    camera_type = get_camera_type()

    if camera_type == 'rpi':
        gain = get_rpi_meta_value('AnalogueGain')
    else:
        gain = get_environment_variable('AS_GAIN')

    if gain == None:
        gain = 0

    return float(gain)

get_camera_type()

Get the configured camera type from the environment.

Returns:

Name	Type	Description
str		Lowercase camera type string (e.g. "rpi").

Source code in scripts/modules/allsky_shared.py

def get_camera_type():
    """
    Get the configured camera type from the environment.

    Returns:
        str:
            Lowercase camera type string (e.g. ``"rpi"``).
    """
    camera_type = get_environment_variable('CAMERA_TYPE')
    return camera_type.lower()

get_ecowitt_data(api_key, app_key, mac_address, temp_unitid=1, pressure_unitid=3)

Fetch live weather data from the remote Ecowitt cloud API.

If all of the required credentials are non-empty, the function builds an API URL and attempts to parse a range of fields such as outdoor and indoor temperatures, humidity, rainfall, wind, pressure, and lightning.

All fields are returned in a nested dict with sensible defaults of None.

Parameters:

Name	Type	Description	Default
api_key	str	Ecowitt API key.	required
app_key	str	Ecowitt application key.	required
mac_address	str	Device MAC address registered with Ecowitt.	required
temp_unitid	int	Temperature unit ID expected by the API. Defaults to 1.	1
pressure_unitid	int	Pressure unit ID expected by the API. Defaults to 3.	3

Returns:

Type	Description
	dict | str: On success, a nested dictionary of parsed values. On HTTP error, a descriptive error string may be returned instead.

Source code in scripts/modules/allsky_shared.py

def get_ecowitt_data(api_key, app_key, mac_address, temp_unitid=1, pressure_unitid=3):
    """
    Fetch live weather data from the remote Ecowitt cloud API.

    If all of the required credentials are non-empty, the function builds
    an API URL and attempts to parse a range of fields such as outdoor and
    indoor temperatures, humidity, rainfall, wind, pressure, and lightning.

    All fields are returned in a nested dict with sensible defaults of None.

    Args:
        api_key (str):
            Ecowitt API key.
        app_key (str):
            Ecowitt application key.
        mac_address (str):
            Device MAC address registered with Ecowitt.
        temp_unitid (int, optional):
            Temperature unit ID expected by the API. Defaults to 1.
        pressure_unitid (int, optional):
            Pressure unit ID expected by the API. Defaults to 3.

    Returns:
        dict | str:
            On success, a nested dictionary of parsed values. On HTTP error,
            a descriptive error string may be returned instead.
    """
    result = {
        'outdoor': {
            'temperature': None,
            'feels_like': None,
            'humidity': None,
            'app_temp': None,
            'dew_point': None
        },
        'indoor': {
            'temperature': None,
            'humidity': None
        },
        'solar_and_uvi': {
            'solar': None,
            'uvi': None
        },
        'rainfall': {
            'rain_rate': None,
            'daily': None,
            'event': None,
            'hourly': None,
            'weekly': None,
            'monthly': None,
            'yearly': None
        },
        'wind': {
            'wind_speed': None,
            'wind_gust': None,
            'wind_direction': None
        },
        'pressure': {
            'relative': None,
            'absolute': None
        },
        'lightning': {
            'distance': None,
            'count': None
        }
    }
    if all(var.strip() for var in (app_key, api_key, mac_address)):
        ECOWITT_API_URL = f'https://api.ecowitt.net/api/v3/device/real_time?application_key={app_key}&api_key={api_key}&mac={mac_address}&call_back=all&temp_unitid={temp_unitid}&pressure_unitid={pressure_unitid}'

        log(4, f"INFO: Reading Ecowitt API from - {ECOWITT_API_URL}")
        try:
            response = requests.get(ECOWITT_API_URL)
            if response.status_code == 200:
                raw_data = response.json()

                result['outdoor']['temperature'] = _get_nested_value(raw_data, 'data.outdoor.temperature.value', float)
                result['outdoor']['feels_like'] = _get_nested_value(raw_data, 'data.outdoor.feels_like.value', float)
                result['outdoor']['humidity'] = _get_nested_value(raw_data, 'data.outdoor.humidity.value', float)
                result['outdoor']['app_temp'] = _get_nested_value(raw_data, 'data.outdoor.app_temp.value', float)
                result['outdoor']['dew_point'] = _get_nested_value(raw_data, 'data.outdoor.dew_point.value', float)

                result['indoor']['temperature'] = _get_nested_value(raw_data, 'data.indoor.temperature.value', float)
                result['indoor']['humidity'] = _get_nested_value(raw_data, 'data.indoor.humidity.value', float)

                result['solar_and_uvi']['solar'] = _get_nested_value(raw_data, 'data.solar_and_uvi.solar.value', float)
                result['solar_and_uvi']['uvi'] = _get_nested_value(raw_data, 'data.solar_and_uvi.uvi.value', float)

                result['rainfall']['rain_rate'] = _get_nested_value(raw_data, 'data.rainfall.rain_rate.value', float)
                result['rainfall']['daily'] = _get_nested_value(raw_data, 'data.rainfall.daily.value', float)
                result['rainfall']['event'] = _get_nested_value(raw_data, 'data.rainfall.event.value', float)
                result['rainfall']['hourly'] = _get_nested_value(raw_data, 'data.rainfall.hourly.value', float)
                result['rainfall']['weekly'] = _get_nested_value(raw_data, 'data.rainfall.weekly.value', float)
                result['rainfall']['monthly'] = _get_nested_value(raw_data, 'data.rainfall.monthly.value', float)
                result['rainfall']['yearly'] = _get_nested_value(raw_data, 'data.rainfall.yearly.value', float)

                result['wind']['wind_speed'] = _get_nested_value(raw_data, 'data.wind.wind_speed.value', float)
                result['wind']['wind_gust'] = _get_nested_value(raw_data, 'data.wind.wind_gust.value', float)
                result['wind']['wind_direction'] = _get_nested_value(raw_data, 'data.wind.wind_direction.value', int)

                result['pressure']['relative'] = _get_nested_value(raw_data, 'data.pressure.relative.value', float)
                result['pressure']['absolute'] = _get_nested_value(raw_data, 'data.pressure.absolute.value', float)

                result['lightning']['distance'] = _get_nested_value(raw_data, 'data.lightning.distance.value', float)
                result['lightning']['count'] = _get_nested_value(raw_data, 'data.lightning.count.value', int)

                log(1, f'INFO: Data read from Ecowitt API')
            else:
                result = f'Got error from the Ecowitt API. Response code {response.status_code}'
                log(0, f'ERROR: {result}')
        except Exception as e:
            me = os.path.basename(__file__)
            eType, eObject, eTraceback = sys.exc_info()
            log(0, f'ERROR: Failed to read data from Ecowitt on line {eTraceback.tb_lineno} in {me} - {e}')
    else:
        log(0, 'ERROR: Missing Ecowitt Application Key, API Key or MAC Address')

    return result

get_ecowitt_local_data(address, password=None)

Fetch live weather data directly from a local Ecowitt gateway.

This variant talks to the gateway's local HTTP API and parses a variety of metrics such as temperatures, humidity, rainfall, wind, pressure and lightning. Values are returned in a nested dict with None defaults.

Units are parsed from the API response and temperature values are converted to Celsius when needed.

Parameters:

Name	Type	Description	Default
address	str	Base URL or IP address of the Ecowitt gateway.	required
password	str	Reserved for password-protected gateways (currently unused).	None

Returns:

Name	Type	Description
dict		Nested dictionary of parsed values.

Source code in scripts/modules/allsky_shared.py

def get_ecowitt_local_data(address, password=None):
    """
    Fetch live weather data directly from a local Ecowitt gateway.

    This variant talks to the gateway's local HTTP API and parses a variety
    of metrics such as temperatures, humidity, rainfall, wind, pressure and
    lightning. Values are returned in a nested dict with None defaults.

    Units are parsed from the API response and temperature values are
    converted to Celsius when needed.

    Args:
        address (str):
            Base URL or IP address of the Ecowitt gateway.
        password (str, optional):
            Reserved for password-protected gateways (currently unused).

    Returns:
        dict:
            Nested dictionary of parsed values.
    """
    '''
    Temp 0 - C, 1 - F
    Pressure 0 - hPA, 1 - inHg, 2 - mmHg
    Wind 0 - m/s, 1 - km/h, 2 - mph, 3 - knots, 5 - Beaufort
    Rain 0 - mm, 1 - in
    Irradiance 0 - Klux, 1 - W/m2, 2 - Kfc
    Capacity 0 - L, 1 - m3, 2 - Gal
    '''

    result = {
        'outdoor': {
            'temperature': None,
            'feels_like': None,
            'humidity': None,
            'app_temp': None,
            'dew_point': None
        },
        'indoor': {
            'temperature': None,
            'humidity': None
        },
        'solar_and_uvi': {
            'solar': None,
            'uvi': None
        },
        'rainfall': {
            'rain_rate': None,
            'daily': None,
            'event': None,
            'hourly': None,
            'weekly': None,
            'monthly': None,
            'yearly': None
        },
        'wind': {
            'wind_speed': None,
            'wind_gust': None,
            'wind_direction': None
        },
        'pressure': {
            'relative': None,
            'absolute': None
        },
        'lightning': {
            'distance': None,
            'count': None
        }
    }

    def parse_val(val, as_type=float, unit=None):
        """
        Extract numeric part from a value string and optionally convert °F to °C.

        Args:
            val (Any):
                Raw value from the Ecowitt JSON (may contain unit text).
            as_type (callable):
                Type to cast the numeric part to (e.g. float, int).
            unit (str | None):
                Optional unit string; when reported as Fahrenheit, values are
                converted to Celsius.

        Returns:
            Any | None:
                Parsed and optionally converted value, or None on failure.
        """
        if val is None:
            return None
        try:
            num_str = str(val).strip().split()[0].strip('%')
            value = as_type(num_str)
            if unit:
                unit = unit.lower()
                if unit in ['f', '°f']:
                    value = round((value - 32) * 5 / 9, 2)
            return value
        except (ValueError, TypeError):
            return None

    def get_val_and_unit(data_list, target_id):
        """
        Helper to locate a record in an Ecowitt list and extract (value, unit).

        Args:
            data_list (list[dict]):
                List of readings as returned by the Ecowitt gateway.
            target_id (str):
                Identifier to match in each dict's ``"id"`` field.

        Returns:
            tuple:
                ``(value, unit)`` where both may be None if the ID is not found.
        """
        for item in data_list:
            if item.get("id") == target_id:
                return item.get("val"), item.get("unit", None)
        return None, None

    LIVE_URL = f'{address}/get_livedata_info?'

    try:
        response = requests.get(LIVE_URL)
        if response.status_code == 200:
            live_data = response.json()

            common = live_data.get("common_list", [])
            val, unit = get_val_and_unit(common, "0x02")
            result['outdoor']['temperature'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x07")
            result['outdoor']['humidity'] = parse_val(val, int, unit)

            val, unit = get_val_and_unit(common, "3")
            result['outdoor']['feels_like'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x03")
            result['outdoor']['dew_point'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x0B")
            result['wind']['wind_speed'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x0C")
            result['wind']['wind_gust'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x0A")
            result['wind']['wind_direction'] = parse_val(val, int, unit)

            val, unit = get_val_and_unit(common, "0x15")
            result['solar_and_uvi']['solar'] = parse_val(val, float, unit)

            val, unit = get_val_and_unit(common, "0x17")
            result['solar_and_uvi']['uvi'] = parse_val(val, int, unit)

            # --- Rain ---
            rain = live_data.get("rain", [])
            for rid, key in {
                "0x0D": "event",
                "0x0E": "rain_rate",
                "0x10": "hourly",
                "0x11": "daily",
                "0x12": "weekly",
                "0x13": "yearly"
            }.items():
                val, unit = get_val_and_unit(rain, rid)
                result['rainfall'][key] = parse_val(val, float, unit)

            # --- WH25 Indoor Sensor ---
            wh25 = live_data.get("wh25", [{}])[0]
            result['indoor']['temperature'] = parse_val(wh25.get("intemp"), float, wh25.get("unit"))
            result['indoor']['humidity'] = parse_val(wh25.get("inhumi"), int)

            result['pressure']['absolute'] = parse_val(wh25.get("abs"), float)
            result['pressure']['relative'] = parse_val(wh25.get("rel"), float)

            # --- Lightning ---
            lightning = live_data.get("lightning", [{}])[0]
            result['lightning']['distance'] = parse_val(lightning.get("distance"), float)
            result['lightning']['count'] = parse_val(lightning.get("count"), int)

    except Exception as e:
        me = os.path.basename(__file__)
        eType, eObject, eTraceback = sys.exc_info()
        log(0, f'ERROR: Failed to read live data from the local Ecowitt gateway on line {eTraceback.tb_lineno} in {me} - {e}')

    return result

get_environment_variable(name, fatal=False, debug=False, try_allsky_debug_file=False)

Helper for reading an environment variable.

This is the modern, snake_case wrapper around the legacy :func:getEnvironmentVariable implementation. New code should call this function rather than the camelCase version.

Parameters:

Name	Type	Description	Default
name		Name of the environment variable to read.	required
fatal		If True and the variable cannot be resolved, the process will terminate with an error.	False
debug		If True, values are read from the debug database instead of the real environment.	False
try_allsky_debug_file		When False (default), the function will fall back to loading variables from variables.json. When True, it will instead try to look up the value in the overlay debug data file.	False

Returns:

Type	Description
	The resolved value as a string, or None if not found (and fatal
	is False).

Source code in scripts/modules/allsky_shared.py

def get_environment_variable(name, fatal=False, debug=False, try_allsky_debug_file=False):
    """
    Helper for reading an environment variable.

    This is the modern, snake_case wrapper around the legacy
    :func:`getEnvironmentVariable` implementation. New code should call this
    function rather than the camelCase version.

    Args:
        name: Name of the environment variable to read.
        fatal: If True and the variable cannot be resolved, the process will
            terminate with an error.
        debug: If True, values are read from the debug database instead of
            the real environment.
        try_allsky_debug_file: When False (default), the function will fall
            back to loading variables from ``variables.json``. When True,
            it will instead try to look up the value in the overlay debug
            data file.

    Returns:
        The resolved value as a string, or None if not found (and ``fatal``
        is False).
    """
    return getEnvironmentVariable(name, fatal, debug)

get_extra_dir(current_only=False)

Helper to get the "extra data" directory or directories.

This simply calls :func:getExtraDir. New code should use this snake_case name.

Source code in scripts/modules/allsky_shared.py

def get_extra_dir(current_only:bool = False) -> list[str] | str:
    """
    Helper to get the "extra data" directory or directories.

    This simply calls :func:`getExtraDir`. New code should use this
    snake_case name.
    """
    return getExtraDir(current_only)

get_flows_with_module(module_name)

Scan module flow files and return those containing a given module.

Only postprocessing_*.json files that are not debug variants are considered. Files that fail to parse are quietly ignored.

Parameters:

Name	Type	Description	Default
module_name	str	Name of the module to search for in the flow definitions.	required

Returns:

Name	Type	Description
dict		Mapping of filename to parsed JSON content for flows that contain the given module.

Source code in scripts/modules/allsky_shared.py

def get_flows_with_module(module_name):
    """
    Scan module flow files and return those containing a given module.

    Only ``postprocessing_*.json`` files that are not debug variants are
    considered. Files that fail to parse are quietly ignored.

    Args:
        module_name (str):
            Name of the module to search for in the flow definitions.

    Returns:
        dict:
            Mapping of filename to parsed JSON content for flows that contain
            the given module.
    """
    folder = Path(ALLSKY_MODULES)
    found: Dict[str, Any] = {}

    for file in folder.glob("*.json"):
        if not file.name.endswith("-debug.json"):
            if file.name.startswith("postprocessing_"):
                try:
                    with file.open("r", encoding="utf-8") as f:
                        data = json.load(f)

                    if module_name in data:
                        found[file.name] = data

                except (json.JSONDecodeError, OSError) as e:
                    pass

    return found

get_gpio_pin(gpio_pin, pi=None, show_errors=False)

Read the logical state of a GPIO pin via the Allsky API (legacy alias).

This definition simply calls :func:read_gpio_pin. It is kept for backward compatibility with existing code that expects get_gpio_pin to return a pin value rather than a board pin object.

Parameters:

Name	Type	Description	Default
gpio_pin	int | str	Pin identifier understood by the Allsky API.	required
pi	Any	Unused placeholder kept for interface compatibility.	None
show_errors	bool	Currently unused; kept for interface compatibility.	False

Returns:

Name	Type	Description
bool		True if the GPIO is reported as "on", False otherwise.

Source code in scripts/modules/allsky_shared.py

def get_gpio_pin(gpio_pin, pi=None, show_errors=False):
    """
    Read the logical state of a GPIO pin via the Allsky API (legacy alias).

    This definition simply calls :func:`read_gpio_pin`. It is kept for
    backward compatibility with existing code that expects ``get_gpio_pin``
    to return a pin value rather than a board pin object.

    Args:
        gpio_pin (int | str):
            Pin identifier understood by the Allsky API.
        pi (Any, optional):
            Unused placeholder kept for interface compatibility.
        show_errors (bool, optional):
            Currently unused; kept for interface compatibility.

    Returns:
        bool:
            True if the GPIO is reported as ``"on"``, False otherwise.
    """
    return read_gpio_pin(gpio_pin, pi=None, show_errors=False)

get_gpio_pin_details(pin)

Get the CircuitPython board pin object for a given numeric pin.

This is a convenience wrapper around :func:getGPIOPin and should be used by new code.

Parameters:

Name	Type	Description	Default
pin	int	Numeric pin index (0–27) corresponding to a board pin.	required

Returns:

Name	Type	Description
Any		The matching board.Dx constant, or None if the pin is unknown.

Source code in scripts/modules/allsky_shared.py

def get_gpio_pin_details(pin):
    """
    Get the CircuitPython ``board`` pin object for a given numeric pin.

    This is a convenience wrapper around :func:`getGPIOPin` and should be
    used by new code.

    Args:
        pin (int):
            Numeric pin index (0–27) corresponding to a board pin.

    Returns:
        Any:
            The matching ``board.Dx`` constant, or None if the pin is unknown.
    """
    return getGPIOPin(pin)

get_hass_sensor_value(ha_url, ha_ltt, ha_sensor)

Query a Home Assistant sensor and return its numeric state.

A GET request is sent to the Home Assistant REST API using the supplied long-lived token. The sensor's state is parsed as a float on success.

Parameters:

Name	Type	Description	Default
ha_url	str	Base URL of the Home Assistant instance (e.g. "http://host:8123").	required
ha_ltt	str	Long-lived access token for Home Assistant.	required
ha_sensor	str	Entity ID of the sensor (e.g. "sensor.outdoor_temp").	required

Returns:

Type	Description
	float | None: The sensor state as a float, or None if the sensor cannot be read.

Source code in scripts/modules/allsky_shared.py

def get_hass_sensor_value(ha_url, ha_ltt, ha_sensor):
    """
    Query a Home Assistant sensor and return its numeric state.

    A GET request is sent to the Home Assistant REST API using the supplied
    long-lived token. The sensor's state is parsed as a float on success.

    Args:
        ha_url (str):
            Base URL of the Home Assistant instance (e.g. ``"http://host:8123"``).
        ha_ltt (str):
            Long-lived access token for Home Assistant.
        ha_sensor (str):
            Entity ID of the sensor (e.g. ``"sensor.outdoor_temp"``).

    Returns:
        float | None:
            The sensor state as a float, or None if the sensor cannot be read.
    """
    result = None

    headers = {
        'Authorization': f'Bearer {ha_ltt}',
        'Content-Type': 'application/json',
    }

    try:
        response = requests.get(f'{ha_url}/api/states/{ha_sensor}', headers=headers)

        if response.status_code == 200:
            result = float(response.json().get('state'))
        else:
            if response.status_code == 404:
                log(0, f'ERROR: Unable to read {ha_sensor} from {ha_url}. homeassistant reports the sensor does not exist')
            else:
                if response.status_code == 401:
                    log(0, f'ERROR: Unable to read {ha_sensor} from {ha_url}. homeassistant reports the token is unauthorised')
                else:
                    log(0, f'ERROR: Unable to read {ha_sensor} from {ha_url}. Error code {response.status_code}')

    except Exception as e:
        me = os.path.basename(__file__)
        eType, eObject, eTraceback = sys.exc_info()
        log(0, f'ERROR: Failed to read data from Homeassistant {eTraceback.tb_lineno} in {me} - {e}')
        result = None

    return result

get_lat_lon()

Read latitude and longitude from settings and return them as floats.

The settings latitude and longitude may be stored in a variety of formats supported by :func:convert_lat_lon (for example, 51.5N or -0.13). If a value is empty, the corresponding return value is None.

Returns:

Type	Description
	Tuple of (lat, lon) where each element is either a float or
	None if not defined.

Source code in scripts/modules/allsky_shared.py

def get_lat_lon():
    """
    Read latitude and longitude from settings and return them as floats.

    The settings ``latitude`` and ``longitude`` may be stored in a variety
    of formats supported by :func:`convert_lat_lon` (for example,
    ``51.5N`` or ``-0.13``). If a value is empty, the corresponding
    return value is ``None``.

    Returns:
        Tuple of ``(lat, lon)`` where each element is either a float or
        None if not defined.
    """
    lat = None
    lon = None

    temp_lat = get_setting('latitude')
    if temp_lat != '':
        lat = convert_lat_lon(temp_lat)
    temp_lon = get_setting('longitude')
    if temp_lon != '':
        lon = convert_lat_lon(temp_lon)

    return lat, lon

get_pi_info(info)

Query simple hardware details about the Raspberry Pi.

Parameters:

Name	Type	Description	Default
info		One of the predefined constants: PI_INFO_MODEL – return the board model string. Pi_INFO_CPU_TEMPERATURE – return the CPU temperature in °C.	required

Returns:

Type	Description
	The requested value, or None if info does not match a
	supported constant.

Source code in scripts/modules/allsky_shared.py

def get_pi_info(info):
    """
    Query simple hardware details about the Raspberry Pi.

    Args:
        info:
            One of the predefined constants:

            - ``PI_INFO_MODEL`` – return the board model string.
            - ``Pi_INFO_CPU_TEMPERATURE`` – return the CPU temperature in °C.

    Returns:
        The requested value, or None if ``info`` does not match a
        supported constant.
    """
    from gpiozero import CPUTemperature
    result = None

    if info == PI_INFO_MODEL:
        pi_info = get_pi_board_info()
        if pi_info is not None:
            result = pi_info.model

    if info == Pi_INFO_CPU_TEMPERATURE:
        result = CPUTemperature().temperature

    return result

get_rpi_meta_value(key)

Read a single value from the Raspberry Pi camera metadata file.

The metadata file format can be either JSON or simple key=value text. This helper tries JSON first and falls back to line-based parsing.

Parameters:

Name	Type	Description	Default
key	str	Metadata key to retrieve.	required

Returns:

Name	Type	Description
Any		The value if found, or 0 if the file is missing, unreadable, or the key is not present.

Source code in scripts/modules/allsky_shared.py

def get_rpi_meta_value(key):
    """
    Read a single value from the Raspberry Pi camera metadata file.

    The metadata file format can be either JSON or simple ``key=value`` text.
    This helper tries JSON first and falls back to line-based parsing.

    Args:
        key (str):
            Metadata key to retrieve.

    Returns:
        Any:
            The value if found, or 0 if the file is missing, unreadable, or the
            key is not present.
    """
    result = None
    metadata_path = get_rpi_metadata()

    if metadata_path is not None:
        try:
            with open(metadata_path, 'r') as file:
                metadata = json.load(file)
                if key in metadata:
                    result = metadata[key]
        except json.JSONDecodeError as e:
            with open(metadata_path, 'r') as f:
                for line in f:
                    if line.startswith(key + "="):
                        result = line.split("=", 1)[1].strip()
        except FileNotFoundError as e:
            result = 0
        except Exception as e:
            result = 0

    return result

get_rpi_metadata()

Determine the path to the Raspberry Pi camera metadata file.

The metadata path is extracted from the extraargs in the main settings file if a --metadata argument is present. If no explicit path is found, a default of metadata.txt in the current directory is used.

Returns:

Name	Type	Description
str		Path to the metadata file.

Source code in scripts/modules/allsky_shared.py

def get_rpi_metadata():
    """
    Determine the path to the Raspberry Pi camera metadata file.

    The metadata path is extracted from the ``extraargs`` in the main
    settings file if a ``--metadata`` argument is present. If no explicit
    path is found, a default of ``metadata.txt`` in the current directory
    is used.

    Returns:
        str:
            Path to the metadata file.
    """
    with open(ALLSKY_SETTINGS_FILE) as file:
        config = json.load(file)

    extraargs = config.get('extraargs', '')
    args = shlex.split(extraargs)

    metadata_path = None
    for i, arg in enumerate(args):
        if arg == '--metadata' and i + 1 < len(args):
            metadata_path = args[i + 1]
            break
    if metadata_path is None:
        metadata_path = os.path.join(ALLSKY_CURRENT_DIR, 'metadata.txt')

    return metadata_path

get_secrets(keys)

Load one or more secrets from env.json in ALLSKYPATH.

The file is expected to contain a flat JSON object mapping key names to secret values. Only the requested keys are returned; missing keys are silently ignored.

Parameters:

Name	Type	Description	Default
keys	Union[str, List[str]]	Either a single key (string) or a list of key names to fetch.	required

Returns:

Type	Description
Union[str, Dict[str, str], None]	If a single key was requested, the value as a string (or None if
Union[str, Dict[str, str], None]	it is not present).
Union[str, Dict[str, str], None]	If multiple keys were requested, a dict mapping each key to
Union[str, Dict[str, str], None]	its value. Keys that are not found are omitted from the mapping.
Union[str, Dict[str, str], None]	If the file cannot be read or parsed, returns None (single key) or
Union[str, Dict[str, str], None]	an empty dict (multiple keys).

Source code in scripts/modules/allsky_shared.py

def get_secrets(keys: Union[str, List[str]]) -> Union[str, Dict[str, str], None]:
    """
    Load one or more secrets from ``env.json`` in ``ALLSKYPATH``.

    The file is expected to contain a flat JSON object mapping key names
    to secret values. Only the requested keys are returned; missing keys
    are silently ignored.

    Args:
        keys:
            Either a single key (string) or a list of key names to fetch.

    Returns:
        If a single key was requested, the value as a string (or None if
        it is not present).

        If multiple keys were requested, a ``dict`` mapping each key to
        its value. Keys that are not found are omitted from the mapping.

        If the file cannot be read or parsed, returns None (single key) or
        an empty dict (multiple keys).
    """
    single = isinstance(keys, str)
    if single:
        keys = [keys]

    try:
        file_path = os.path.join(ALLSKYPATH, 'env.json')
        with open(file_path, 'r') as f:
            secrets = json.load(f)

        results = {k: secrets[k] for k in keys if k in secrets}

        if single:
            return results.get(keys[0])
        return results

    except (IOError, json.JSONDecodeError) as e:
        print(f"Error reading secrets file: {e}")
        return None if single else {}

get_sensor_temperature()

Get the current sensor temperature for the active camera.

For Raspberry Pi cameras, this reads the SensorTemperature value from the Pi metadata. For other camera types it uses the AS_TEMPERATURE_C environment variable.

If no temperature can be determined, 0.0 is returned.

Returns:

Name	Type	Description
float		Sensor temperature in °C.

Source code in scripts/modules/allsky_shared.py

def get_sensor_temperature():
    """
    Get the current sensor temperature for the active camera.

    For Raspberry Pi cameras, this reads the ``SensorTemperature`` value
    from the Pi metadata. For other camera types it uses the
    ``AS_TEMPERATURE_C`` environment variable.

    If no temperature can be determined, 0.0 is returned.

    Returns:
        float:
            Sensor temperature in °C.
    """
    temperature = 0
    camera_type = get_camera_type()

    if camera_type == 'rpi':
        temperature = get_rpi_meta_value('SensorTemperature')
    else:
        temperature = get_environment_variable('AS_TEMPERATURE_C')

    if temperature == None:
        temperature = 0

    return float(temperature)

get_setting(settingName)

Helper for reading a setting from the loaded settings JSON.

This wraps the legacy :func:getSetting and should be used in new code. See :func:getSetting for details.

Source code in scripts/modules/allsky_shared.py

def get_setting(settingName):
    """
    Helper for reading a setting from the loaded settings JSON.

    This wraps the legacy :func:`getSetting` and should be used in new
    code. See :func:`getSetting` for details.
    """
    return getSetting(settingName)

get_value_from_debug_data(key)

Look up a key in the overlay debug file.

This function is used when running in debug mode to retrieve values that would normally be supplied via environment variables.

Parameters:

Name	Type	Description	Default
key	str	Name of the value to retrieve.	required

Returns:

Type	Description
str | None	The corresponding value as a string with whitespace removed, or
str | None	None if the file does not exist, cannot be read, or the key is
str | None	not present.

Source code in scripts/modules/allsky_shared.py

def get_value_from_debug_data(key: str) -> str | None:
    """
    Look up a key in the overlay debug file.

    This function is used when running in debug mode to retrieve values
    that would normally be supplied via environment variables.

    Args:
        key: Name of the value to retrieve.

    Returns:
        The corresponding value as a string with whitespace removed, or
        None if the file does not exist, cannot be read, or the key is
        not present.
    """
    setup_for_command_line()

    try:
        allsky_tmp = os.environ["ALLSKY_TMP"]
        file_path = os.path.join(allsky_tmp, "overlaydebug.txt")

        try:
            with open(file_path, "r", encoding="utf-8") as f:
                for line in f:
                    if not line.strip() or line.strip().startswith("#"):
                        continue

                    parts = line.split(maxsplit=1)
                    if len(parts) == 2 and parts[0] == key:
                        return "".join(parts[1].split())
        except FileNotFoundError:
            return None
    except:
        return None
    return None

load_extra_data_file(file_name, type='')

Load an extra data file from one or more ALLSKY_EXTRA directories.

This helper looks for the given file name in all configured extra-data directories (via :func:get_extra_dir). If it finds a readable file, it will attempt to parse it according to its extension.

Currently only JSON files are parsed. Text files are recognised but not yet processed (the block is a placeholder).

Parameters:

Name	Type	Description	Default
file_name	str	Name of the extra data file to load (e.g. "extra.json").	required
type	str	Force the file type. If set to "json", the file is parsed as JSON regardless of its extension.	''

Returns:

Name	Type	Description
dict		Parsed JSON data if successful; otherwise an empty dict.

Source code in scripts/modules/allsky_shared.py

def load_extra_data_file(file_name, type=''):
    """
    Load an extra data file from one or more ALLSKY_EXTRA directories.

    This helper looks for the given file name in all configured extra-data
    directories (via :func:`get_extra_dir`). If it finds a readable file,
    it will attempt to parse it according to its extension.

    Currently only JSON files are parsed. Text files are recognised but
    not yet processed (the block is a placeholder).

    Args:
        file_name (str):
            Name of the extra data file to load (e.g. ``"extra.json"``).
        type (str, optional):
            Force the file type. If set to ``"json"``, the file is parsed
            as JSON regardless of its extension.

    Returns:
        dict:
            Parsed JSON data if successful; otherwise an empty dict.
    """
    result = {}
    extra_data_paths = get_extra_dir()
    for extra_data_path in extra_data_paths:
        if extra_data_path is not None:               # it should never be None
            extra_data_filename = os.path.join(extra_data_path, file_name)
            file_path = Path(extra_data_filename)
            if file_path.is_file() and isFileReadable(file_path):
                file_extension = Path(file_path).suffix

                if file_extension == '.json' or type == 'json':
                    try:
                        with open(extra_data_filename, 'r') as file:
                            result = json.load(file)
                    except json.JSONDecodeError:
                        log(0, f'ERROR: cannot read {extra_data_filename}.')

                if file_extension == '.txt':
                    pass

    return result

load_json_file(path)

Load a JSON file and return its parsed contents.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the JSON file.	required

Returns:

Type	Description
	The parsed JSON data (dict or list). If the file does not exist,
	cannot be read, or contains invalid JSON, an empty dict is
	returned.

Source code in scripts/modules/allsky_shared.py

def load_json_file(path: str | Path):
    """
    Load a JSON file and return its parsed contents.

    Args:
        path: Path to the JSON file.

    Returns:
        The parsed JSON data (dict or list). If the file does not exist,
        cannot be read, or contains invalid JSON, an empty dict is
        returned.
    """
    try:
        file_path = Path(path)
        if not file_path.is_file():
            return {}

        with file_path.open("r", encoding="utf-8") as file:
            return json.load(file)

    except (OSError, json.JSONDecodeError):
        return {}

load_mask(mask_file_name, target_image)

Load a grayscale mask image and resize it to match a target image.

The mask is loaded from ALLSKY_OVERLAY/images/<mask_file_name> and converted to a float mask in the range [0, 1]. If the mask dimensions do not match the target image, it is resized accordingly.

Parameters:

Name	Type	Description	Default
mask_file_name	str	Name of the mask file to load.	required
target_image	ndarray	Target image whose shape is used for resizing.	required

Returns:

Type	Description
	numpy.ndarray | None: Float mask array in the range [0, 1], or None if the mask could not be loaded.

Source code in scripts/modules/allsky_shared.py

def load_mask(mask_file_name, target_image):
    """
    Load a grayscale mask image and resize it to match a target image.

    The mask is loaded from ``ALLSKY_OVERLAY/images/<mask_file_name>`` and
    converted to a float mask in the range [0, 1]. If the mask dimensions do
    not match the target image, it is resized accordingly.

    Args:
        mask_file_name (str):
            Name of the mask file to load.
        target_image (numpy.ndarray):
            Target image whose shape is used for resizing.

    Returns:
        numpy.ndarray | None:
            Float mask array in the range [0, 1], or None if the mask could
            not be loaded.
    """
    import cv2
    mask = None

    mask_path = os.path.join(ALLSKY_OVERLAY, 'images', mask_file_name)
    target_shape = target_image.shape[:2]

    mask = cv2.imread(mask_path, cv2.IMREAD_GRAYSCALE)
    if mask is not None:
        if (mask.shape[0] != target_shape[0]) or (mask.shape[1] != target_shape[1]):
            mask = cv2.resize(mask, (target_shape[1], target_shape[0]))
        mask = mask.astype(np.float32) / 255.0

    return mask

load_secrets_file()

Load environment-style secrets from env.json in the Allsky home directory.

Any JSON decoding errors are treated as an empty file.

Returns:

Name	Type	Description
dict	Dict[str, Any]	Parsed secrets dictionary, or an empty dict if missing/invalid.

Source code in scripts/modules/allsky_shared.py

def load_secrets_file() -> Dict[str, Any]:
    """
    Load environment-style secrets from ``env.json`` in the Allsky home directory.

    Any JSON decoding errors are treated as an empty file.

    Returns:
        dict:
            Parsed secrets dictionary, or an empty dict if missing/invalid.
    """
    file_path = Path(os.path.join(ALLSKYPATH, 'env.json'))
    env_data: Dict[str, Any] = {}
    if file_path.is_file():
        with file_path.open("r", encoding="utf-8") as f:
            try:
                env_data = json.load(f) or {}
            except json.JSONDecodeError:
                env_data = {}

    return env_data

log(level, text, preventNewline=False, exitCode=None, sendToAllsky=False)

Very simple method to log data if in verbose mode

Log a message to stdout (depending on log level) and optionally forward it to the Allsky WebUI.

Parameters:

Name	Type	Description	Default
level		Numeric log level. The message is printed if the global LOGLEVEL is greater than or equal to this value. Level 0 is treated as an error.	required
text		The message to log.	required
preventNewline		If True, the message is printed without a trailing newline.	False
exitCode		If not None, the process exits with this code after logging the message.	None
sendToAllsky		If True, the message is also passed to the Allsky WebUI via addMessage.sh. Level 0 messages are always sent as errors.	False

Notes

The function does not raise exceptions. If the WebUI message script fails, the error is silently ignored.

Source code in scripts/modules/allsky_shared.py

def log(level, text, preventNewline = False, exitCode=None, sendToAllsky=False):
    """ Very simple method to log data if in verbose mode

    Log a message to stdout (depending on log level) and optionally
    forward it to the Allsky WebUI.

    Args:
        level:
            Numeric log level. The message is printed if the global
            ``LOGLEVEL`` is greater than or equal to this value. Level 0
            is treated as an error.
        text:
            The message to log.
        preventNewline:
            If True, the message is printed without a trailing newline.
        exitCode:
            If not None, the process exits with this code after logging
            the message.
        sendToAllsky:
            If True, the message is also passed to the Allsky WebUI via
            ``addMessage.sh``. Level 0 messages are always sent as
            errors.

    Notes:
        The function does not raise exceptions. If the WebUI message
        script fails, the error is silently ignored.
    """
    global LOGLEVEL, ALLSKY_SCRIPTS

    if LOGLEVEL >= level:
        if preventNewline:
            print(text, end="")
        else:
            print(text)

    if sendToAllsky or level == 0:
        if level == 0:
            type = "error"
        else:
            type = "warning"
        # Need to escape single quotes in {text}.
        doubleQuote = '"'
        text = text.replace("'", f"'{doubleQuote}'{doubleQuote}'")
        command = os.path.join(ALLSKY_SCRIPTS, f"addMessage.sh --type {type} --msg '{text}'")
        os.system(command)

    if exitCode is not None:
        sys.exit(exitCode)

mask_image(image, mask_file_name='', log_info=False)

Apply a mask to an image, returning a masked copy.

The mask is loaded via :func:load_mask and applied either directly (for grayscale images) or per-channel (for colour images). The result is clipped and converted back to uint8.

Parameters:

Name	Type	Description	Default
image	ndarray	Input image (grayscale or BGR).	required
mask_file_name	str	Name of the mask image file. If empty, no masking is performed and None is returned.	''
log_info	bool	If True, log a message at level 4 when a mask is applied.	False

Returns:

Type	Description
	numpy.ndarray | None: Masked image, or None if no mask is applied or an error occurs.

Source code in scripts/modules/allsky_shared.py

def mask_image(image, mask_file_name='', log_info=False):
    """
    Apply a mask to an image, returning a masked copy.

    The mask is loaded via :func:`load_mask` and applied either directly
    (for grayscale images) or per-channel (for colour images). The result
    is clipped and converted back to ``uint8``.

    Args:
        image (numpy.ndarray):
            Input image (grayscale or BGR).
        mask_file_name (str, optional):
            Name of the mask image file. If empty, no masking is performed
            and None is returned.
        log_info (bool, optional):
            If True, log a message at level 4 when a mask is applied.

    Returns:
        numpy.ndarray | None:
            Masked image, or None if no mask is applied or an error occurs.
    """
    output = None
    try:
        if mask_file_name != '':
            mask = load_mask(mask_file_name, image)
            if len(image.shape) == 2:
                image = image.astype(np.float32)
                output = image * mask
            else:
                image = image.astype(np.float32)
                if mask.ndim == 2:
                    mask = mask[..., np.newaxis]
                output = image * mask

            output = np.clip(output, 0, 255).astype(np.uint8)

            if log_info:
                log(4, f'INFO: Mask {mask_file_name} applied')

    except Exception as e:
        me = os.path.basename(__file__)
        eType, eObject, eTraceback = sys.exc_info()
        log(0, f'ERROR: mask_image failed on line {eTraceback.tb_lineno} in {me} - {e}')

    return output

normalise_on_off(value)

Normalise an on/off style value to the strings "on" or "off".

Parameters:

Name	Type	Description	Default
value	Any	Raw value (e.g. "on", "1", "off", 0).	required

Returns:

Name	Type	Description
str		"on" if the input looks like an enabled value, otherwise "off".

Source code in scripts/modules/allsky_shared.py

def normalise_on_off(value):
    """
    Normalise an on/off style value to the strings ``"on"`` or ``"off"``.

    Args:
        value (Any):
            Raw value (e.g. ``"on"``, ``"1"``, ``"off"``, ``0``).

    Returns:
        str:
            ``"on"`` if the input looks like an enabled value, otherwise ``"off"``.
    """
    if str(value).strip().lower() == 'on' or str(value).strip() == '1':
        return 'on'
    return 'off'

obfuscate_password(password)

Obfuscate a password, leaving the first and last characters visible.

This is used when logging configuration without exposing full credentials.

Parameters:

Name	Type	Description	Default
password	str	Original password string.	required

Returns:

Type	Description
str	Obfuscated password. Very short passwords are completely masked.

Source code in scripts/modules/allsky_shared.py

def obfuscate_password(password: str) -> str:
    """
    Obfuscate a password, leaving the first and last characters visible.

    This is used when logging configuration without exposing full
    credentials.

    Args:
        password: Original password string.

    Returns:
        Obfuscated password. Very short passwords are completely masked.
    """
    if not password:
        return ""
    if len(password) <= 2:
        return "*" * len(password)
    return password[0] + "*" * (len(password) - 2) + password[-1]

read_environment_variable(name)

Read an environment variable without any Allsky-specific fallback.

This is a very thin wrapper around os.environ access and does not attempt to pull values from variables.json or any debug store.

Parameters:

Name	Type	Description	Default
name		Environment variable name.	required

Returns:

Type	Description
	The value as a string, or None if the variable is not defined.

Source code in scripts/modules/allsky_shared.py

def read_environment_variable(name):
    """
    Read an environment variable without any Allsky-specific fallback.

    This is a very thin wrapper around ``os.environ`` access and does not
    attempt to pull values from ``variables.json`` or any debug store.

    Args:
        name: Environment variable name.

    Returns:
        The value as a string, or None if the variable is not defined.
    """
    result = None
    try:
        result = os.environ[name]
    except KeyError:
        result = None        

    return result

read_gpio_pin(gpio_pin, pi=None, show_errors=False)

Read the logical state of a GPIO pin via the Allsky HTTP API.

A GET request is sent to the Allsky API, and the returned JSON is expected to contain a "value" field with the string "on" or "off".

Parameters:

Name	Type	Description	Default
gpio_pin	int | str	Pin identifier understood by the Allsky API.	required
pi	Any	Unused placeholder for compatibility with other GPIO interfaces.	None
show_errors	bool	Currently unused; errors are propagated via exceptions.	False

Returns:

Name	Type	Description
bool		True if the GPIO value is "on", False otherwise.

Raises:

Type	Description
HTTPError	If the HTTP request fails (via raise_for_status).

Source code in scripts/modules/allsky_shared.py

def read_gpio_pin(gpio_pin, pi=None, show_errors=False):
    """
    Read the logical state of a GPIO pin via the Allsky HTTP API.

    A GET request is sent to the Allsky API, and the returned JSON is
    expected to contain a ``"value"`` field with the string ``"on"`` or
    ``"off"``.

    Args:
        gpio_pin (int | str):
            Pin identifier understood by the Allsky API.
        pi (Any, optional):
            Unused placeholder for compatibility with other GPIO interfaces.
        show_errors (bool, optional):
            Currently unused; errors are propagated via exceptions.

    Returns:
        bool:
            True if the GPIO value is ``"on"``, False otherwise.

    Raises:
        requests.HTTPError:
            If the HTTP request fails (via ``raise_for_status``).
    """
    api_url = get_api_url()
    response = requests.get(
        f'{api_url}/gpio/digital/{gpio_pin}',
        timeout=2
    )
    response.raise_for_status()
    data = response.json()

    return data.get('value') == 'on'

run_python_script(script, args=None, cwd=None)

Run a Python script using the same interpreter as the current process (e.g., inside a venv).

This function ensures the target script is executed with the current Python interpreter (sys.executable), so that packages installed in the active virtual environment are available.

Parameters:

Name	Type	Description	Default
script	str	Path to the Python script to execute.	required
args	Optional[List[str]]	Additional arguments to pass to the script. Defaults to None.	None
cwd	Optional[str]	Working directory in which to run the script. If None, uses the current directory.	None

Returns:

Type	Description
Tuple[int, str]	Tuple[int, str]: A tuple containing: - return code (int): The process's exit code, or 127 if the script is not found. - output (str): Combined standard output and standard error from the script, stripped of trailing whitespace.

Example

code, output = run_python_script("myscript.py", ["--option", "value"]) print(code, output) 0 Script ran successfully

Source code in scripts/modules/allsky_shared.py

def run_python_script(script: str, args: Optional[List[str]] = None, cwd: Optional[str] = None) -> Tuple[int, str]:
    """
    Run a Python script using the same interpreter as the current process (e.g., inside a venv).

    This function ensures the target script is executed with the current Python interpreter
    (`sys.executable`), so that packages installed in the active virtual environment are available.

    Args:
        script (str): Path to the Python script to execute.
        args (Optional[List[str]]): Additional arguments to pass to the script. Defaults to None.
        cwd (Optional[str]): Working directory in which to run the script. If None, uses the current directory.

    Returns:
        Tuple[int, str]: A tuple containing:
            - return code (int): The process's exit code, or 127 if the script is not found.
            - output (str): Combined standard output and standard error from the script, stripped of trailing whitespace.

    Example:
        >>> code, output = run_python_script("myscript.py", ["--option", "value"])
        >>> print(code, output)
        0 Script ran successfully
    """
    args = args or []
    try:
        proc = subprocess.run(
            [sys.executable, script, *args],
            capture_output=True,
            text=True,
            check=False,
            cwd=cwd,
        )
        output = (proc.stdout or "") + (proc.stderr or "")
        return proc.returncode, output.strip()
    except FileNotFoundError:
        return 127, f"Script not found: {script}"

run_script(script)

Run an arbitrary executable script and capture its output.

Parameters:

Name	Type	Description	Default
script	str	Path to the script or binary to execute.	required

Returns:

Type	Description
int	Tuple (returncode, output) where:
str	returncode is the process exit code, or 127 if the script was not found.
Tuple[int, str]	output is the combined stdout and stderr as a single string.

Source code in scripts/modules/allsky_shared.py

def run_script(script: str) -> Tuple[int, str]:
    """
    Run an arbitrary executable script and capture its output.

    Args:
        script: Path to the script or binary to execute.

    Returns:
        Tuple ``(returncode, output)`` where:

        - ``returncode`` is the process exit code, or 127 if the script
          was not found.
        - ``output`` is the combined stdout and stderr as a single
          string.
    """
    try:
        result = subprocess.run(
            [script],
            capture_output=True,
            text=True,
            check=False
        )
        output = result.stdout + result.stderr
        return result.returncode, output.strip()
    except FileNotFoundError:
        return 127, f"Script not found: {script}"

save_extra_data(file_name='', extra_data={}, source='', structure={}, custom_fields={}, event='postcapture')

Persist "extra data" for use by Allsky overlay modules.

This function writes the provided data to a file inside the current ALLSKY_EXTRA directory (resolved via get_extra_dir(True)), using a temporary file in ALLSKY_TMP and an atomic move to avoid partial writes. It ensures the destination directory exists and is web-server accessible, applies final permissions, and (optionally) updates a database when the structure indicates one is in use.

Behavior

1) Ensure extra data directory exists (checkAndCreateDirectory) and enable web access (create_file_web_server_access). 2) If the target filename ends with .json, normalize/shape the payload via format_extra_data_json(extra_data, structure, source). 3) Merge any custom_fields into the payload (overrides existing keys). 4) Serialize to JSON (pretty-printed) and write to a temp file created in ALLSKY_TMP, then atomically move it to the final path. 5) Set mode 0o770 and call set_permissions() for owner/group alignment. 6) If structure contains a "database" key, call update_database().

Parameters:

Name	Type	Description	Default
file_name	str	File name (with extension) to write into the extra data directory.	''
extra_data	Any	Data to persist. If file_name ends with .json, this should be JSON-serializable. Non-JSON targets are written as the JSON string.	{}
source	str	Context or origin tag passed to the JSON formatter. Default: ''.	''
structure	dict	Schema/metadata guiding JSON formatting and optional DB updates. If it contains "database", update_database() will be invoked. Default: {}.	{}
custom_fields	dict	Extra key/values to inject into the payload before serialization. Keys here override the same keys in extra_data. Default: {}.	{}
event	str	Event type (e.g. 'postcapture', 'periodic') used when deciding if database updates should occur.	'postcapture'

Returns:

Type	Description
	None

Side Effects

• Creates/updates a file in ALLSKY_EXTRA.
• Applies filesystem permissions to the output file.
• May perform a database update if requested by structure.

Error Handling

Any exception is (currently) allowed to propagate only into the surrounding code; earlier versions logged and swallowed errors.

Source code in scripts/modules/allsky_shared.py

def save_extra_data(file_name: str = '', extra_data: dict = {}, source: str = '', structure: dict = {}, custom_fields: dict = {}, event: str = 'postcapture'):
    """
    Persist "extra data" for use by Allsky overlay modules.

    This function writes the provided data to a file inside the current
    ALLSKY_EXTRA directory (resolved via `get_extra_dir(True)`), using a
    temporary file in ALLSKY_TMP and an atomic move to avoid partial writes.
    It ensures the destination directory exists and is web-server accessible,
    applies final permissions, and (optionally) updates a database when the
    `structure` indicates one is in use.

    Behavior:
      1) Ensure extra data directory exists (`checkAndCreateDirectory`) and
         enable web access (`create_file_web_server_access`).
      2) If the target filename ends with `.json`, normalize/shape the payload
         via `format_extra_data_json(extra_data, structure, source)`.
      3) Merge any `custom_fields` into the payload (overrides existing keys).
      4) Serialize to JSON (pretty-printed) and write to a temp file created in
         `ALLSKY_TMP`, then atomically move it to the final path.
      5) Set mode 0o770 and call `set_permissions()` for owner/group alignment.
      6) If `structure` contains a `"database"` key, call `update_database()`.

    Args:
        file_name (str):
            File name (with extension) to write into the extra data directory.
        extra_data (Any):
            Data to persist. If `file_name` ends with `.json`, this should be
            JSON-serializable. Non-JSON targets are written as the JSON string.
        source (str, optional):
            Context or origin tag passed to the JSON formatter. Default: ''.
        structure (dict, optional):
            Schema/metadata guiding JSON formatting and optional DB updates.
            If it contains `"database"`, `update_database()` will be invoked.
            Default: {}.
        custom_fields (dict, optional):
            Extra key/values to inject into the payload before serialization.
            Keys here override the same keys in `extra_data`. Default: {}.
        event (str, optional):
            Event type (e.g. ``'postcapture'``, ``'periodic'``) used when
            deciding if database updates should occur.

    Returns:
        None

    Side Effects:
        - Creates/updates a file in ALLSKY_EXTRA.
        - Applies filesystem permissions to the output file.
        - May perform a database update if requested by `structure`.

    Error Handling:
        Any exception is (currently) allowed to propagate only into the
        surrounding code; earlier versions logged and swallowed errors.
    """
    saveExtraData(file_name, extra_data, source, structure, custom_fields, event)

save_json_file(data, filename)

Save a dictionary to a JSON file with pretty formatting.

Parameters:

Name	Type	Description	Default
data	dict	Dictionary to save. Must be JSON-serializable.	required
filename	Union[str, Path]	Path or string of the file to write.	required

Returns:

Type	Description
None	True if the file could be written successfully, False otherwise.

Source code in scripts/modules/allsky_shared.py

def save_json_file(data: dict, filename: Union[str, Path]) -> None:
    """
    Save a dictionary to a JSON file with pretty formatting.

    Args:
        data: Dictionary to save. Must be JSON-serializable.
        filename: Path or string of the file to write.

    Returns:
        True if the file could be written successfully, False otherwise.
    """
    file_path = Path(filename)

    try:
        with file_path.open('w', encoding='utf-8') as file:
            json.dump(data, file, ensure_ascii=False, indent=4)
    except:
        return False

    return True

set_gpio_pin(gpio_pin, state, name='', pi=None, show_errors=False)

Set the logical state of a GPIO pin via the Allsky HTTP API.

Parameters:

Name	Type	Description	Default
gpio_pin	int | str	Pin identifier understood by the Allsky API.	required
state	Any	Desired state; normalised using :func:normalise_on_off to "on" or "off".	required
pi	Any	Unused placeholder for compatibility with other GPIO interfaces.	None
show_errors	bool	Currently unused; errors are propagated via exceptions.	False

Returns:

Name	Type	Description
dict		Parsed JSON response from the API.

Raises:

Type	Description
HTTPError	If the HTTP request fails (via raise_for_status).

Source code in scripts/modules/allsky_shared.py

def set_gpio_pin(gpio_pin, state, name="", pi=None, show_errors=False):
    """
    Set the logical state of a GPIO pin via the Allsky HTTP API.

    Args:
        gpio_pin (int | str):
            Pin identifier understood by the Allsky API.
        state (Any):
            Desired state; normalised using :func:`normalise_on_off` to
            ``"on"`` or ``"off"``.
        pi (Any, optional):
            Unused placeholder for compatibility with other GPIO interfaces.
        show_errors (bool, optional):
            Currently unused; errors are propagated via exceptions.

    Returns:
        dict:
            Parsed JSON response from the API.

    Raises:
        requests.HTTPError:
            If the HTTP request fails (via ``raise_for_status``).
    """
    api_url = get_api_url()
    state = normalise_on_off(state)
    response = requests.post(
        f'{api_url}/gpio/digital',
        json={
            'pin': str(gpio_pin),
            'state': state.lower(),
            'name': name
        },
        timeout=2
    )
    response.raise_for_status()
    return response.json()

set_last_run(module)

Helper to record that a module has just run.

This function simply calls the legacy :func:setLastRun. New code should use this name; the camelCase variant is kept for older code.

Source code in scripts/modules/allsky_shared.py

def set_last_run(module):
    """
    Helper to record that a module has just run.

    This function simply calls the legacy :func:`setLastRun`. New code
    should use this name; the camelCase variant is kept for older code.
    """
    setLastRun(module)

set_pwm(gpio_pin, duty_cycle, name='', pi=None, show_errors=False)

Set PWM output on a GPIO pin via the Allsky HTTP API.

This helper posts the requested duty cycle (and a fixed frequency of 1000Hz) to the API.

Parameters:

Name	Type	Description	Default
gpio_pin	int | str	Pin identifier understood by the Allsky API.	required
duty_cycle	int | float	Duty cycle value; interpreted by the remote API.	required
pi	Any	Unused placeholder for compatibility with other GPIO interfaces.	None
show_errors	bool	Currently unused; errors are propagated via exceptions.	False

Returns:

Name	Type	Description
dict		Parsed JSON response from the API.

Raises:

Type	Description
HTTPError	If the HTTP request fails (via raise_for_status).

Source code in scripts/modules/allsky_shared.py

def set_pwm(gpio_pin, duty_cycle, name="", pi=None, show_errors=False):
    """
    Set PWM output on a GPIO pin via the Allsky HTTP API.

    This helper posts the requested duty cycle (and a fixed frequency of
    1000Hz) to the API.

    Args:
        gpio_pin (int | str):
            Pin identifier understood by the Allsky API.
        duty_cycle (int | float):
            Duty cycle value; interpreted by the remote API.
        pi (Any, optional):
            Unused placeholder for compatibility with other GPIO interfaces.
        show_errors (bool, optional):
            Currently unused; errors are propagated via exceptions.

    Returns:
        dict:
            Parsed JSON response from the API.

    Raises:
        requests.HTTPError:
            If the HTTP request fails (via ``raise_for_status``).
    """
    api_url = get_api_url()
    frequency = 1000
    response = requests.post(
        f'{api_url}/gpio/pwm',
        json={
            'pin': str(gpio_pin),
            'duty': duty_cycle,
            'frequency': frequency,
            'name': name
        },
        timeout=2
    )
    response.raise_for_status()
    return response.json()

should_run(module, period)

Helper to check whether a module should run again.

This wrapper simply calls the legacy :func:shouldRun. New code should use this snake_case name; the camelCase version is retained for backwards compatibility.

See :func:shouldRun for details.

Source code in scripts/modules/allsky_shared.py

def should_run(module, period):
    """
    Helper to check whether a module should run again.

    This wrapper simply calls the legacy :func:`shouldRun`. New code
    should use this snake_case name; the camelCase version is retained
    for backwards compatibility.

    See :func:`shouldRun` for details.
    """
    return shouldRun(module, period)

stop_pwm(gpio_pin)

Stop PWM output on a GPIO pin via the Allsky HTTP API.

This is implemented by sending a PWM request with 0% duty cycle.

Parameters:

Name	Type	Description	Default
gpio_pin	int | str	Pin identifier understood by the Allsky API.	required

Returns:

Name	Type	Description
dict		Parsed JSON response from the API.

Raises:

Type	Description
HTTPError	If the HTTP request fails (via raise_for_status).

Source code in scripts/modules/allsky_shared.py

def stop_pwm(gpio_pin):
    """
    Stop PWM output on a GPIO pin via the Allsky HTTP API.

    This is implemented by sending a PWM request with 0% duty cycle.

    Args:
        gpio_pin (int | str):
            Pin identifier understood by the Allsky API.

    Returns:
        dict:
            Parsed JSON response from the API.

    Raises:
        requests.HTTPError:
            If the HTTP request fails (via ``raise_for_status``).
    """
    api_url = get_api_url()
    frequency = 1000
    duty_cycle = 0
    response = requests.post(
        f'{api_url}/gpio/pwm',
        json={
            'pin': str(gpio_pin),
            'duty': duty_cycle,
            'frequency': frequency
        },
        timeout=2
    )
    response.raise_for_status()
    return response.json()

to_bool(v)

Normalise a value to a boolean, supporting several truthy strings.

This version differs slightly from the earlier :func:to_bool in this file: it treats "true", "1", "yes" and "y" (case- insensitive) as True, and everything else as False. It is used when normalising configuration dictionaries.

Parameters:

Name	Type	Description	Default
v	bool | str | None	Input value.	required

Returns:

Name	Type	Description
bool	bool	Normalised boolean value.

Source code in scripts/modules/allsky_shared.py

def to_bool(v: bool | str) -> bool:
    """
    Normalise a value to a boolean, supporting several truthy strings.

    This version differs slightly from the earlier :func:`to_bool` in this
    file: it treats ``"true"``, ``"1"``, ``"yes"`` and ``"y"`` (case-
    insensitive) as True, and everything else as False. It is used when
    normalising configuration dictionaries.

    Args:
        v (bool | str | None):
            Input value.

    Returns:
        bool:
            Normalised boolean value.
    """
    if isinstance(v, bool):
        return v
    if v is None:
        return False
    return str(v).strip().lower() in ("true", "1", "yes", "y")

update_setting(values)

Helper to update one or more settings.

This wraps the legacy :func:updateSetting. New code should use this snake_case name.

Source code in scripts/modules/allsky_shared.py

def update_setting(values):
    """
    Helper to update one or more settings.

    This wraps the legacy :func:`updateSetting`. New code should use
    this snake_case name.
    """
    updateSetting(values)