micromet package

Subpackages

Submodules

micromet.pipeline module

Complete pipeline for processing micrometeorological data with Micromet.

This module provides high-level orchestration for the complete data processing workflow, from raw data files to cleaned, validated, and analyzed datasets.

Classes

Pipeline : Main orchestration class for data processing PipelineConfig : Configuration container for pipeline settings ProcessingResult : Container for processing results and metadata

Functions

run_pipeline : Convenience function to run complete pipeline process_station : Process a single station’s data batch_process : Process multiple stations

Examples

Basic usage:

>>> from micromet.pipeline import Pipeline
>>>
>>> # Process a single file
>>> pipeline = Pipeline()
>>> result = pipeline.process_file(
...     'data/US-UTW_Flux.dat',
...     site_id='US-UTW'
... )
>>>
>>> # Batch process all stations
>>> results = pipeline.batch_process(
...     input_dir='./raw_data',
...     output_dir='./processed_data'
... )

Command-line usage:

$ python -m micromet.pipeline –site US-UTW –input data/ –output results/ $ python -m micromet.pipeline –batch –input data/ –output results/

class micromet.pipeline.Pipeline(config=None, logger=None)[source]

Bases: object

Main orchestration class for micrometeorological data processing.

This class coordinates the complete workflow from raw data files to cleaned, validated, and analyzed datasets.

Parameters:

config (Optional[PipelineConfig]) – Configuration settings for the pipeline.
logger (Optional[Logger]) – Logger instance for tracking progress.

config

Pipeline configuration.

Type:: PipelineConfig

logger

Logger instance.

Type:: logging.Logger

reader

Data reader instance.

Type:: AmerifluxDataProcessor

__init__(config=None, logger=None)[source]: Initialize the Pipeline.

batch_process(input_dir, output_dir, pattern='*Flux*.dat', data_type='eddy')[source]

Process multiple files in a directory.

Parameters:

input_dir (Union[str, Path]) – Directory containing input files.
output_dir (Union[str, Path]) – Directory for output files.
pattern (str) – Glob pattern for finding input files. Defaults to “Flux.dat”.
data_type (str) – Type of data (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

Results for all processed files.

Return type:

List[ProcessingResult]

process_file(input_file, site_id=None, output_dir=None, data_type='eddy')[source]

Process a single data file through the complete pipeline.

Parameters:

input_file (Union[str, Path]) – Path to input data file.
site_id (Optional[str]) – Station identifier. If None, attempts to extract from filename.
output_dir (Union[str, Path, None]) – Directory for output files. If None, uses input file directory.
data_type (str) – Type of data (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

Container with processing results and metadata.

Return type:

ProcessingResult

process_station(site_id, input_dir, output_dir, data_types=['eddy', 'met'])[source]

Process all data types for a single station.

Parameters:

site_id (str) – Station identifier (e.g., ‘US-UTW’).
input_dir (Union[str, Path]) – Directory containing input files.
output_dir (Union[str, Path]) – Directory for output files.
data_types (List[str]) – Data types to process. Defaults to [‘eddy’, ‘met’].

Returns:

Dictionary mapping data_type to ProcessingResult.

Return type:

Dict[str, ProcessingResult]

class micromet.pipeline.PipelineConfig(check_timestamps=True, drop_soil=True, generate_reports=True, generate_plots=False, save_intermediate=False, var_limits_csv=None, expected_freq='30min', output_format='csv')[source]

Bases: object

Configuration settings for the data processing pipeline.

check_timestamps

Whether to perform timestamp alignment analysis (slower but thorough).

Type:: bool

drop_soil

Whether to drop extra soil sensor columns.

Type:: bool

generate_reports

Whether to generate validation and gap reports.

Type:: bool

generate_plots

Whether to generate diagnostic plots.

Type:: bool

save_intermediate

Whether to save intermediate processing steps.

Type:: bool

var_limits_csv

Path to custom variable limits CSV file.

Type:: Path or None

expected_freq

Expected data frequency (e.g., ‘30min’).

Type:: str

output_format

Output file format (‘csv’, ‘parquet’, ‘feather’).

Type:: str

check_timestamps: bool = True

drop_soil: bool = True

expected_freq: str = '30min'

generate_plots: bool = False

generate_reports: bool = True

output_format: str = 'csv'

save_intermediate: bool = False

to_dict()[source]

Convert configuration to dictionary.

Return type:: dict

var_limits_csv: Path | None = None

class micromet.pipeline.ProcessingResult(site_id, success, input_file, output_file=None, n_records_input=0, n_records_output=0, n_flagged=0, processing_time=0.0, timestamp_issues=None, error_message=None, reports=<factory>)[source]

Bases: object

Container for processing results and metadata.

site_id

Station identifier.

Type:: str

success

Whether processing completed successfully.

Type:: bool

input_file

Path to input file.

Type:: Path

output_file

Path to output file (if saved).

Type:: Path or None

n_records_input

Number of records in input data.

Type:: int

n_records_output

Number of records in output data.

Type:: int

n_flagged

Number of records flagged during QA/QC.

Type:: int

processing_time

Processing time in seconds.

Type:: float

timestamp_issues

Detected timestamp alignment issues.

Type:: dict or None

error_message

Error message if processing failed.

Type:: str or None

reports

Dictionary of generated reports.

Type:: dict

error_message: str | None = None

input_file: Path

n_flagged: int = 0

n_records_input: int = 0

n_records_output: int = 0

output_file: Path | None = None

processing_time: float = 0.0

reports: Dict

site_id: str

success: bool

summary()[source]

Generate a human-readable summary.

Return type:: str

timestamp_issues: Dict | None = None

to_dict()[source]

Convert result to dictionary.

Return type:: dict

micromet.pipeline.batch_process(input_dir, output_dir, **kwargs)[source]

Convenience function for batch processing.

Parameters:

input_dir (Union[str, Path]) – Input directory.
output_dir (Union[str, Path]) – Output directory.
**kwargs – Additional arguments passed to Pipeline constructor.

Returns:

Results for all processed files.

Return type:

List[ProcessingResult]

micromet.pipeline.main()[source]: Command-line interface for the pipeline.

micromet.pipeline.process_station(site_id, input_dir, output_dir, **kwargs)[source]

Convenience function to process a single station.

Parameters:

site_id (str) – Station identifier.
input_dir (Union[str, Path]) – Input directory.
output_dir (Union[str, Path]) – Output directory.
**kwargs – Additional arguments passed to Pipeline constructor.

Returns:

Processing results for each data type.

Return type:

Dict[str, ProcessingResult]

micromet.reader module

This module provides the AmerifluxDataProcessor class for reading and parsing AmeriFlux-style CSV files (TOA5 or AmeriFlux output) into a pandas DataFrame.

class micromet.reader.AmerifluxDataProcessor(logger=None)[source]

Bases: object

A class for reading and parsing AmeriFlux-style CSV files.

This class is designed to handle Campbell Scientific TOA5 files or standard AmeriFlux output files, parsing them into a pandas DataFrame.

Parameters:: logger (Logger) – A logger for tracking the data processing. If not provided, a default logger is used.

logger

The logger used for logging messages.

Type:: logging.Logger

skip_rows

The number of rows to skip at the beginning of the file.

Type:: int or list of int

names

The column names for the DataFrame.

Type:: list of str

NA_VALUES = ['-9999', 'NAN', 'NaN', 'nan', nan, -9999.0]

__init__(logger=None)[source]

Initialize the AmerifluxDataProcessor.

Parameters:: logger (Logger) – A logger for tracking the data processing. If not provided, a default logger is used.

iterate_through_stations()[source]

Iterate through all stations and compile their data.

This method iterates through a predefined list of stations, compiles the data for each station, and returns a dictionary of DataFrames.

Returns:: A dictionary where keys are station IDs and values are DataFrames of the compiled data for each station.
Return type:: dict

raw_file_compile(main_dir, station_folder_name, search_str='*Flux_AmeriFluxFormat*.dat')[source]

Compile raw AmeriFlux datalogger files into a single DataFrame.

This method searches for files matching a given pattern within a station’s directory, processes each file, and concatenates them into a single DataFrame.

Parameters:

main_dir (Union[str, Path]) – The main directory containing the station folders.
station_folder_name (Union[str, Path]) – The name of the station folder.
search_str (str) – The search string (glob pattern) for finding files to compile. Defaults to “Flux_AmeriFluxFormat.dat”.

Returns:

A DataFrame containing the compiled data, or None if no valid files were found.

Return type:

Optional[DataFrame]

to_dataframe(file)[source]

Read an AmeriFlux-style CSV file and return it as a pandas DataFrame.

This method first determines the header structure of the file and then reads the data into a DataFrame, handling missing values.

Parameters:: file (Union[str, Path]) – The path to the CSV file to be read.
Returns:: A DataFrame containing the parsed data from the file.
Return type:: DataFrame

micromet.station_data_pull module

class micromet.station_data_pull.StationDataDownloader(config, logger=None)[source]

Bases: object

A class to manage downloading data from a station’s logger.

This class handles the connection and data download from a Campbell Scientific data logger via its web API.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object containing station details and credentials.
logger (Logger) – A logger for logging messages. If None, a new logger is created.

config

The configuration object.

Type:: configparser.ConfigParser or dict

logger

The logger instance.

Type:: logging.Logger

logger_credentials

The authentication credentials for the logger.

Type:: requests.auth.HTTPBasicAuth

__init__(config, logger=None)[source]

Initialize the StationDataDownloader.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object containing station details and credentials.
logger (Logger) – A logger for logging messages. If None, a new logger is created.

download_from_station(station, loggertype='eddy', mode='since-time', p1='0', p2='0')[source]

Download data from a station’s logger.

This method constructs a request to the station’s web API to download data based on the specified parameters.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.
mode (str) – The data query mode (‘since-time’, ‘most-recent’, etc.). Defaults to ‘since-time’.
p1 (str) – The primary parameter for the query (e.g., start time). Defaults to “0”.
p2 (str) – The secondary parameter for the query (e.g., end time). Defaults to “0”.

Returns:

A tuple containing the downloaded data as a DataFrame, the size of the data packet in MB, and the HTTP status code.

Return type:

tuple[pd.DataFrame | None, float | None, int]

static get_station_id(stationid)[source]

Extract the station ID from a full station identifier string.

Parameters:: stationid (str) – The full station identifier (e.g., ‘US-ABC’).
Returns:: The extracted station ID (e.g., ‘ABC’).
Return type:: str

get_times(station, loggertype='eddy')[source]

Retrieve the current time from the logger and the system.

This method queries a station’s logger for its current time and also gets the current system time for comparison.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

A tuple containing the logger’s current time as a string and the system’s current time as a string.

Return type:

Tuple[Optional[str], str]

class micromet.station_data_pull.StationDataProcessor(config, engine, logger=None)[source]

Bases: StationDataDownloader

A class for processing and managing station data.

This class extends StationDataDownloader to add functionality for reformatting data, interacting with a database, and managing the overall data processing workflow.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object with station details.
engine (Engine) – A SQLAlchemy engine for database connections.
logger (Logger) – A logger for logging messages.

engine

The SQLAlchemy engine instance.

Type:: sqlalchemy.engine.base.Engine

__init__(config, engine, logger=None)[source]

Initialize the StationDataProcessor.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object with station details.
engine (Engine) – A SQLAlchemy engine for database connections.
logger (Logger) – A logger for logging messages.

compare_sql_to_station(df, station, field='timestamp_end', loggertype='eddy')[source]

Compare station data with records in the database and filter new entries.

Parameters:

df (DataFrame) – The DataFrame containing the station data.
station (str) – The identifier for the station.
field (str) – The field to use for comparison. Defaults to “timestamp_end”.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

A DataFrame containing only the new records.

Return type:

DataFrame

database_columns(dat)[source]

Get the list of column names for a given database table.

Parameters:: dat (str) – The type of data (‘eddy’ or ‘met’), which corresponds to the table name.
Returns:: A list of column names in the specified table.
Return type:: list

get_max_date(station, loggertype='eddy')[source]

Get the maximum timestamp from the station’s data in the database.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

The latest timestamp found in the database for the station.

Return type:

datetime

get_station_data(station, reformat=True, loggertype='eddy', config_path='./data/reformatter_vars.yml', var_limits_csv='./data/extreme_values.csv', drop_soil=False)[source]

Fetch and process data for a single station.

This method downloads data from a station, optionally reformats it, and returns the processed data.

Parameters:

station (str) – The identifier for the station.
reformat (bool) – Whether to reformat the downloaded data. Defaults to True.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.
config_path (str) – The path to the reformatter configuration file.
var_limits_csv (str) – The path to the variable limits CSV file.
drop_soil (bool) – Whether to drop soil-related data. Defaults to False.

Returns:

A tuple containing the processed DataFrame and the size of the downloaded data packet in MB.

Return type:

Tuple[Optional[DataFrame], Optional[float]]

process_station_data(site_folders, config_path='./data/reformatter_vars.yml', var_limits_csv='./data/extreme_values.csv')[source]

Process and upload data for all specified stations.

This method iterates through a dictionary of site folders, fetches data for each station, processes it, and uploads it to the database.

Parameters:

site_folders (dict) – A dictionary mapping station IDs to folder names.
config_path (str) – The path to the reformatter configuration file. Defaults to “./data/reformatter_vars.yml”.
var_limits_csv (str) – The path to the variable limits CSV file. Defaults to “./data/extreme_values.csv”.

Return type:

None

static remove_existing_records(df, column_to_check, values_to_remove, logger=None)[source]

Remove rows from a DataFrame that already exist in the database.

Parameters:

df (DataFrame) – The input DataFrame.
column_to_check (str) – The name of the column to check for existing values.
values_to_remove (list) – A list of values to be removed from the DataFrame.
logger (Logger) – A logger for logging messages. Defaults to None.

Returns:

The DataFrame with existing records removed.

Return type:

DataFrame

micromet.station_info module

This module contains station-specific information, such as site folders and logger IDs.

micromet.utils module

Utility functions for the micromet package.

micromet.utils.create_reformatter_from_site(site_id, config_dir='src/micromet/data', check_timestamps=True, **reformatter_kwargs)[source]

Create a Reformatter instance with site configuration loaded from .ini file.

This is a convenience factory function that reads the site configuration and creates a properly configured Reformatter instance.

Parameters:

site_id (str) – The site identifier (e.g., ‘US-CdM’, ‘US-UTW’).
config_dir (Path | str) – Directory containing the .ini files. Defaults to ‘src/micromet/data’.
check_timestamps (bool) – Whether to enable timestamp checking. Defaults to True.
**reformatter_kwargs – Additional keyword arguments passed to Reformatter (e.g., drop_soil, var_limits_csv).

Returns:

A configured Reformatter instance.

Return type:

Reformatter

Examples

>>> reformatter = create_reformatter_from_site('US-CdM')
>>> df_clean, report, ts_results = reformatter.process(raw_data)

>>> # With additional options
>>> reformatter = create_reformatter_from_site(
...     'US-UTW',
...     drop_soil=False,
...     check_timestamps=True
... )

>>> # Disable timestamp checking for speed
>>> reformatter = create_reformatter_from_site(
...     'US-UTB',
...     check_timestamps=False
... )

micromet.utils.extract_config_for_reformatter(site_id, config_dir='src/micromet/data')[source]

Extract only the values needed for Reformatter from a site config.

This is a convenience function that returns just the three values needed to initialize a Reformatter with timestamp checking.

Parameters:

site_id (str) – The site identifier (e.g., ‘US-CdM’).
config_dir (Path | str) – Directory containing the .ini files.

Returns:

A tuple of (site_lat, site_lon, site_utc_offset).

Return type:

Tuple[float, float, float]

Examples

>>> lat, lon, utc = extract_config_for_reformatter('US-CdM')
>>> lat, lon, utc
(37.5241, -109.7471, -7.0)

micromet.utils.get_all_site_configs(config_dir='src/micromet/data')[source]

Read all site configurations from .ini files in a directory.

Parameters:: config_dir (Path | str) – Directory containing the .ini files. Defaults to ‘src/micromet/data’.
Returns:: Dictionary mapping site_id to configuration dictionaries.
Return type:: Dict[str, Dict[str, float | str]]

Examples

>>> all_configs = get_all_site_configs()
>>> all_configs['US-CdM']['site_lat']
37.5241
>>> list(all_configs.keys())
['US-CdM', 'US-UTB', 'US-UTD', ...]

micromet.utils.load_yaml(path)[source]

Load a YAML file and return its contents as a dictionary.

Parameters:: path (Path | str) – The path to the YAML file.
Returns:: The contents of the YAML file as a dictionary.
Return type:: Dict
Raises:: FileNotFoundError – If the specified file does not exist.

micromet.utils.logger_check(logger)[source]

Initialize and return a logger instance if none is provided.

This function checks if a logger object is provided. If not, it creates and configures a new logger.

Parameters:: logger (Logger | None) – An existing logger instance.
Returns:: A configured logger instance.
Return type:: Logger

micromet.utils.read_site_config(site_id, config_dir='src/micromet/data')[source]

Read site configuration from an .ini file.

Parameters:

site_id (str) – The site identifier (e.g., ‘US-CdM’, ‘US-UTW’).
config_dir (Path | str) – Directory containing the .ini files. Defaults to ‘src/micromet/data’.

Returns:

Dictionary with keys: - ‘site_lat’: float - Station latitude - ‘site_lon’: float - Station longitude - ‘site_utc_offset’: float - UTC offset in hours - ‘site_elevation’: float - Station elevation in meters - ‘site_name’: str - Full station name - ‘site_id’: str - Station identifier

Return type:

Dict[str, float | str]

Raises:

FileNotFoundError – If the .ini file for the site is not found.
KeyError – If required metadata fields are missing.

Examples

>>> config = read_site_config('US-CdM')
>>> config['site_lat']
37.5241
>>> config['site_utc_offset']
-7.0

Module contents

Micromet: A package for processing and analyzing micrometeorological data.

This package provides a collection of tools for reading, reformatting, performing quality control, and generating reports from micrometeorological and flux data, particularly from AmeriFlux-style data sources.

The main components of the package are: - AmerifluxDataProcessor: For reading and parsing data files. - Reformatter: For cleaning and standardizing data. - tools: A collection of utility functions for analysis. - graphs: For creating various plots and visualizations. - StationDataDownloader: For downloading data from stations. - StationDataProcessor: For processing and managing station data.

class micromet.AmerifluxDataProcessor(logger=None)[source]

Bases: object

A class for reading and parsing AmeriFlux-style CSV files.

This class is designed to handle Campbell Scientific TOA5 files or standard AmeriFlux output files, parsing them into a pandas DataFrame.

Parameters:: logger (Logger) – A logger for tracking the data processing. If not provided, a default logger is used.

logger

The logger used for logging messages.

Type:: logging.Logger

skip_rows

The number of rows to skip at the beginning of the file.

Type:: int or list of int

names

The column names for the DataFrame.

Type:: list of str

NA_VALUES = ['-9999', 'NAN', 'NaN', 'nan', nan, -9999.0]

__init__(logger=None)[source]

Initialize the AmerifluxDataProcessor.

Parameters:: logger (Logger) – A logger for tracking the data processing. If not provided, a default logger is used.

iterate_through_stations()[source]

Iterate through all stations and compile their data.

This method iterates through a predefined list of stations, compiles the data for each station, and returns a dictionary of DataFrames.

Returns:: A dictionary where keys are station IDs and values are DataFrames of the compiled data for each station.
Return type:: dict

raw_file_compile(main_dir, station_folder_name, search_str='*Flux_AmeriFluxFormat*.dat')[source]

Compile raw AmeriFlux datalogger files into a single DataFrame.

This method searches for files matching a given pattern within a station’s directory, processes each file, and concatenates them into a single DataFrame.

Parameters:

main_dir (Union[str, Path]) – The main directory containing the station folders.
station_folder_name (Union[str, Path]) – The name of the station folder.
search_str (str) – The search string (glob pattern) for finding files to compile. Defaults to “Flux_AmeriFluxFormat.dat”.

Returns:

A DataFrame containing the compiled data, or None if no valid files were found.

Return type:

Optional[DataFrame]

to_dataframe(file)[source]

Read an AmeriFlux-style CSV file and return it as a pandas DataFrame.

This method first determines the header structure of the file and then reads the data into a DataFrame, handling missing values.

Parameters:: file (Union[str, Path]) – The path to the CSV file to be read.
Returns:: A DataFrame containing the parsed data from the file.
Return type:: DataFrame

class micromet.Reformatter(var_limits_csv=None, drop_soil=True, check_timestamps=False, site_lat=None, site_lon=None, site_utc_offset=-7, logger=None)[source]

Bases: object

A class to clean and standardize station data for flux/met processing.

This class provides a pipeline for preparing raw station data by applying a series of transformations, including fixing timestamps, renaming columns, applying physical limits, and checking timestamp alignment.

Parameters:

var_limits_csv (str | Path | None) – Path to a CSV file containing variable limits. If not provided, default limits are used.
drop_soil (bool) – If True, extra soil-related columns are dropped. Defaults to True.
check_timestamps (bool) – If True, perform timestamp alignment analysis on radiation data. Defaults to False.
site_lat (float | None) – Latitude of the site (required if check_timestamps=True).
site_lon (float | None) – Longitude of the site (required if check_timestamps=True).
site_utc_offset (int) – UTC offset in hours for the site (required if check_timestamps=True).
logger (Logger | None) – A logger for tracking the reformatting process. If not provided, a default logger is used.

logger

The logger used for logging messages.

Type:: logging.Logger

config

A dictionary of configuration parameters for the reformatting process.

Type:: dict

varlimits

A DataFrame containing the physical limits for each variable.

Type:: pd.DataFrame

drop_soil

A flag indicating whether to drop extra soil columns.

Type:: bool

check_timestamps

A flag indicating whether to perform timestamp alignment checks.

Type:: bool

site_lat

The latitude of the site.

Type:: float

site_lon

The longitude of the site.

Type:: float

site_utc_offset

The UTC offset of the site in hours.

Type:: float

__init__(var_limits_csv=None, drop_soil=True, check_timestamps=False, site_lat=None, site_lon=None, site_utc_offset=-7, logger=None)[source]

Initialize the Reformatter.

Parameters:

var_limits_csv (str | Path | None) – Path to a CSV file containing variable limits.
drop_soil (bool) – If True, extra soil-related columns are dropped. Defaults to True.
check_timestamps (bool) – If True, perform timestamp alignment analysis. Defaults to False.
site_lat (float | None) – Latitude of the site (required if check_timestamps=True).
site_lon (float | None) – Longitude of the site (required if check_timestamps=True).
site_utc_offset (int) – UTC offset in hours (required if check_timestamps=True).
logger (Logger | None) – A logger for tracking the reformatting process.

finalize(df)[source]: Finalize the data by applying cleaning and standardization steps.

prepare(df, interval=30, data_type='eddy')[source]: Current method - keep for backward compatibility

preprocess(df, data_type='eddy', interval=30)[source]: Preprocess the data by applying initial cleaning and standardization steps.

process(df, interval, data_type='eddy')[source]

Prepare the data by applying a series of cleaning and standardization steps.

This method takes a DataFrame of station data and applies a pipeline of transformations to clean and standardize it. The steps include fixing timestamps, renaming columns, setting numeric types, resampling, applying physical limits, and optionally checking timestamp alignment.

Parameters:

df (DataFrame) – The input DataFrame of station data.
data_type (str) – The type of data being processed (e.g., ‘eddy’, ‘met’). This is used to determine which column renaming map to use. Defaults to ‘eddy’.
interval (int) – The sampling interval used with the data; must be either 30 or 60 minutes

Returns:

A tuple containing: - The prepared DataFrame with standardized and cleaned data. - A report DataFrame detailing the changes made during the

application of physical limits.

A dictionary with timestamp alignment results (if check_timestamps=True), or None otherwise. Contains keys: ‘summary’, ‘composites’, ‘flags’.

Return type:

Tuple[DataFrame, DataFrame, Optional[Dict]]

class micromet.StationDataDownloader(config, logger=None)[source]

Bases: object

A class to manage downloading data from a station’s logger.

This class handles the connection and data download from a Campbell Scientific data logger via its web API.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object containing station details and credentials.
logger (Logger) – A logger for logging messages. If None, a new logger is created.

config

The configuration object.

Type:: configparser.ConfigParser or dict

logger

The logger instance.

Type:: logging.Logger

logger_credentials

The authentication credentials for the logger.

Type:: requests.auth.HTTPBasicAuth

__init__(config, logger=None)[source]

Initialize the StationDataDownloader.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object containing station details and credentials.
logger (Logger) – A logger for logging messages. If None, a new logger is created.

download_from_station(station, loggertype='eddy', mode='since-time', p1='0', p2='0')[source]

Download data from a station’s logger.

This method constructs a request to the station’s web API to download data based on the specified parameters.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.
mode (str) – The data query mode (‘since-time’, ‘most-recent’, etc.). Defaults to ‘since-time’.
p1 (str) – The primary parameter for the query (e.g., start time). Defaults to “0”.
p2 (str) – The secondary parameter for the query (e.g., end time). Defaults to “0”.

Returns:

A tuple containing the downloaded data as a DataFrame, the size of the data packet in MB, and the HTTP status code.

Return type:

tuple[pd.DataFrame | None, float | None, int]

static get_station_id(stationid)[source]

Extract the station ID from a full station identifier string.

Parameters:: stationid (str) – The full station identifier (e.g., ‘US-ABC’).
Returns:: The extracted station ID (e.g., ‘ABC’).
Return type:: str

get_times(station, loggertype='eddy')[source]

Retrieve the current time from the logger and the system.

This method queries a station’s logger for its current time and also gets the current system time for comparison.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

A tuple containing the logger’s current time as a string and the system’s current time as a string.

Return type:

Tuple[Optional[str], str]

class micromet.StationDataProcessor(config, engine, logger=None)[source]

Bases: StationDataDownloader

A class for processing and managing station data.

This class extends StationDataDownloader to add functionality for reformatting data, interacting with a database, and managing the overall data processing workflow.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object with station details.
engine (Engine) – A SQLAlchemy engine for database connections.
logger (Logger) – A logger for logging messages.

engine

The SQLAlchemy engine instance.

Type:: sqlalchemy.engine.base.Engine

__init__(config, engine, logger=None)[source]

Initialize the StationDataProcessor.

Parameters:

config (Union[ConfigParser, dict]) – A configuration object with station details.
engine (Engine) – A SQLAlchemy engine for database connections.
logger (Logger) – A logger for logging messages.

compare_sql_to_station(df, station, field='timestamp_end', loggertype='eddy')[source]

Compare station data with records in the database and filter new entries.

Parameters:

df (DataFrame) – The DataFrame containing the station data.
station (str) – The identifier for the station.
field (str) – The field to use for comparison. Defaults to “timestamp_end”.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

A DataFrame containing only the new records.

Return type:

DataFrame

database_columns(dat)[source]

Get the list of column names for a given database table.

Parameters:: dat (str) – The type of data (‘eddy’ or ‘met’), which corresponds to the table name.
Returns:: A list of column names in the specified table.
Return type:: list

get_max_date(station, loggertype='eddy')[source]

Get the maximum timestamp from the station’s data in the database.

Parameters:

station (str) – The identifier for the station.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.

Returns:

The latest timestamp found in the database for the station.

Return type:

datetime

get_station_data(station, reformat=True, loggertype='eddy', config_path='./data/reformatter_vars.yml', var_limits_csv='./data/extreme_values.csv', drop_soil=False)[source]

Fetch and process data for a single station.

This method downloads data from a station, optionally reformats it, and returns the processed data.

Parameters:

station (str) – The identifier for the station.
reformat (bool) – Whether to reformat the downloaded data. Defaults to True.
loggertype (str) – The type of logger (‘eddy’ or ‘met’). Defaults to ‘eddy’.
config_path (str) – The path to the reformatter configuration file.
var_limits_csv (str) – The path to the variable limits CSV file.
drop_soil (bool) – Whether to drop soil-related data. Defaults to False.

Returns:

A tuple containing the processed DataFrame and the size of the downloaded data packet in MB.

Return type:

Tuple[Optional[DataFrame], Optional[float]]

process_station_data(site_folders, config_path='./data/reformatter_vars.yml', var_limits_csv='./data/extreme_values.csv')[source]

Process and upload data for all specified stations.

This method iterates through a dictionary of site folders, fetches data for each station, processes it, and uploads it to the database.

Parameters:

site_folders (dict) – A dictionary mapping station IDs to folder names.
config_path (str) – The path to the reformatter configuration file. Defaults to “./data/reformatter_vars.yml”.
var_limits_csv (str) – The path to the variable limits CSV file. Defaults to “./data/extreme_values.csv”.

Return type:

None

static remove_existing_records(df, column_to_check, values_to_remove, logger=None)[source]

Remove rows from a DataFrame that already exist in the database.

Parameters:

df (DataFrame) – The input DataFrame.
column_to_check (str) – The name of the column to check for existing values.
values_to_remove (list) – A list of values to be removed from the DataFrame.
logger (Logger) – A logger for logging messages. Defaults to None.

Returns:

The DataFrame with existing records removed.

Return type:

DataFrame