API Reference

def main() -> None:
    """Entry point when running this file as a Python script.

    This is where the instructions begin.

    Arguments: None.
    Returns: None.
    """
    log_header(LOG, "P02")

    LOG.info("===================================")
    LOG.info("START main()")
    LOG.info("===================================")

    LOG.info("-------------------------------")
    LOG.info("01. LOAD the data.")
    LOG.info("-------------------------------")

    # Use the imported privacy-preserving log_path() function
    # To indicate where we will look for the data file.
    log_path(LOG, "data file", path=DATA_FILE_PATH)

    # Call the built-in pandas `read_csv` function.
    # Store the tabular pandas DataFrame returned
    # in a local variable named `df`.

    df: pd.DataFrame = pd.read_csv(DATA_FILE_PATH)

    LOG.info("Data loaded successfully.")

    LOG.info("-------------------------------")
    LOG.info("02. INSPECT the data.")
    LOG.info("-------------------------------")

    # Call the inspect() function to get a string
    # with basic information about the DataFrame.
    # Pass in the pandas DataFrame (df)
    # The grain (what one row represents)
    # And the log so it knows where to send messages.

    inspection_string: str = inspect(df=df, grain=GRAIN, log=LOG)

    LOG.info(inspection_string)

    LOG.info("-------------------------------")
    LOG.info("03. REPEAT logic using a for loop.")
    LOG.info("-------------------------------")

    # Get a list of all column names in the DataFrame.
    # Use the DataFrame's `columns` attribute and convert it to a list
    # with the built-in tolist() method.
    column_names: list[str] = df.columns.tolist()

    # For each name in the column names list, log its name.
    # Note that we must use a colon at the end of the for loop line.
    # And we must indent the body of the for loop correctly.
    for name in column_names:
        LOG.info(f"Column name: {name}")

    # Up above, we choose a column to group by and log the reason for choosing it.
    LOG.info(f"Selected group column: {GROUP_COLUMN}")
    LOG.info(f"Reason for choosing this group: {WHY_THIS_GROUP}")

    # Now, let us use Python to get the unique values in the selected group column.
    # Use the df[column name] to get a one-dimensional array of values
    # by passing in the exact column name as a string (in quotes).
    # NOTE: The entry above must exactly match
    # the column name in the CSV file, including case and spaces.
    # Once we have that, we can apply the .unique() method to get the unique values.
    # Once we have that, we can apply the .tolist() method to convert
    # the array of unique values into a Python list of strings.
    unique_list: list[str] = df[GROUP_COLUMN].unique().tolist()

    # For each unique item in the list, log the value.
    for item in unique_list:
        LOG.info(f"Item: {item}")

    LOG.info("-------------------------------")
    LOG.info("04. TRANSFORM one list to another list.")
    LOG.info("-------------------------------")

    # Python uses something called a "list comprehension"
    # to transform one list into another when the transformation is simple.
    # It is often more concise and readable than using a for loop.
    # The list comprehension syntax is:
    # [expression for item in iterable]
    # where the expression is a simple transformation applied to each item.

    # Common simple string transformations include:
    # - converting strings to uppercase. e.g., name.upper()
    # - converting strings to lowercase. e.g., name.lower()
    # - stripping whitespace, e.g., name.strip()

    capitalized_column_names: list[str] = [name.upper() for name in column_names]
    LOG.info(f"Capitalized column names: {capitalized_column_names}")

    LOG.info("-------------------------------")
    LOG.info("05. BRANCH based on conditions.")
    LOG.info("-------------------------------")

    # Log the selected measurement column and the reason for choosing it.
    LOG.info(f"Selected measurement column: {MEASUREMENT_COLUMN}")
    LOG.info(f"Reason for choosing this measurement: {WHY_THIS_MEASUREMENT}")

    minimum: float = df[MEASUREMENT_COLUMN].min()
    maximum: float = df[MEASUREMENT_COLUMN].max()
    mean: float = df[MEASUREMENT_COLUMN].mean()
    LOG.info(f"{MEASUREMENT_COLUMN} - Minimum: {minimum}")
    LOG.info(f"{MEASUREMENT_COLUMN} - Maximum:  {maximum}")
    LOG.info(f"{MEASUREMENT_COLUMN} - Mean:     {mean}")
    LOG.info("-------------------------------")

    # Get the selected measurement for the first row in the DataFrame.
    # Provide the exact column name as a string to access its values
    # as an array-like object, from which we can select specific rows using iloc.
    # iloc stands for "index location" and is used to select rows by their integer index.
    # Python starts counting at 0, so iloc[0] refers to the first row.
    # If it helps, you can think of it as 0 as "different from the list start".
    # There is no difference between the first item and the start of the list so
    # its offset or index is 0,
    # and it can be accessed using iloc[0]
    # The second item is one away from the start,
    # so it can be accessed using iloc[1].
    sample_index: int = 0
    sample_reading: float = df[MEASUREMENT_COLUMN].iloc[sample_index]
    LOG.info(f"Sample {MEASUREMENT_COLUMN}: {sample_reading}")

    LOG.info(f"Short threshold multiplier: {SHORT_THRESHOLD_MULTIPLIER}")
    LOG.info(f"Long threshold multiplier:  {LONG_THRESHOLD_MULTIPLIER}")

    short_threshold: float = SHORT_THRESHOLD_MULTIPLIER * mean
    long_threshold: float = LONG_THRESHOLD_MULTIPLIER * mean

    LOG.info(f"Short threshold: {short_threshold}")
    LOG.info(f"Long threshold:  {long_threshold}")

    # Use the Python keywords if, elif, and else
    # to classify the selected measurement based on the calculated thresholds.
    # elif means "else if"
    if sample_reading < short_threshold:
        classification_string: str = "SHORT"
    elif sample_reading > long_threshold:
        classification_string: str = "LONG"
    else:
        classification_string: str = "MEDIUM"

    LOG.info(f"First row {MEASUREMENT_COLUMN} classification: {classification_string}")

    LOG.info("-------------------------------")
    LOG.info("06. REPEAT while a condition is true.")
    LOG.info("-------------------------------")

    # We can also perform logic repeatedly using a while loop.
    # This is often used for streaming data or continuously monitoring a condition.
    # In this example, we simulate streaming data by repeatedly processing
    # one measurement from the CSV file
    # every so many seconds, for a total of MAX_RECORDS measurements.

    # Constant values used by the while loop.
    MAX_RECORDS: Final[int] = 10  # CUSTOM: change this from 10.
    STREAM_WAIT_SECONDS: Final[int] = 1  # CUSTOM: Change this from 1 second.

    LOG.info("Starting to process measurements periodically...")
    LOG.info(f"Max records to process: {MAX_RECORDS}")
    LOG.info(f"Stream wait seconds: {STREAM_WAIT_SECONDS}")

    # Initialize the count variable used by the while loop.
    # By convention, counting starts at 0, so the first pass reads row 0.
    count: int = 0
    LOG.info(f"Current count: {count}")

    # Start the while loop to process measurements periodically
    # while the count is less than the maximum number of records.
    while count < MAX_RECORDS:
        # Get the measurement from row `count`, which advances each pass.
        current_measurement: float = df[MEASUREMENT_COLUMN].iloc[count]
        LOG.info(f"Current {MEASUREMENT_COLUMN}: {current_measurement}")

        count += 1
        LOG.info(f"Updated count: {count}")

        time.sleep(STREAM_WAIT_SECONDS)

    LOG.info("-------------------------------")
    LOG.info("07. VISUALIZE the selected measurement.")
    LOG.info("-------------------------------")

    LOG.info("Creating a chart to visualize the selected measurement.")
    LOG.info("We selected one numeric column, so let's look at the distribution.")

    # Define a path to save the distribution plot.
    # REQUIRED: Use the "docs/images" folder to store generated charts.
    CHART_PATH = Path("docs/images/measurement-distribution.png")

    # Call an imported function that will show a distribution plot
    # Pass in the pandas DataFrame (df) along with the selected measurement column.
    # It will return a matplotlib Axes object representing the distribution plot.
    ax = show_numeric_distribution(
        df,
        column=MEASUREMENT_COLUMN,
    )

    # call the save_chart() function and pass in the Axes and the path
    save_chart(ax, CHART_PATH)
    LOG.info(f"Chart saved successfully at {CHART_PATH}.")

    LOG.info(
        "IMPORTANT: Close chart window to continue by clicking its X or close button."
    )
    plt.show()

    LOG.info("===================================")
    LOG.info("END main() - Executed successfully!")
    LOG.info("===================================")

def get_analyst_description(
    grain: str,
    target: str,
    feature: str,
    why: str,
    log: logging.Logger,
) -> str:
    """Get a formatted summary string of the analyst description.

    These are the analyst's declarations, written after looking at the data.
    This is critical analyst work: look, then say what one row means,
    which of the columns might be a target we could predict,
    which of the columns might be a feature we could use if we
    were to build a model to predict the target,
    and why we think that feature might be related to the target.

    Arguments:
        grain: what one row means.
        target: a thing we could try to predict.
        feature: a feature (clue / indicator / column) that might help.
        why: why this feature might help predict the target.
        log: the logger to write progress to.

    Returns:
        a formatted multi-line string.
    """
    log.info("START get_analyst_description")

    summary_string: str = f"""
--------------------------------------------------------
Analyst Data Description (and Possible Prediction Plan):
--------------------------------------------------------
    A row represents:           {grain}
    A target we might predict:  {target}
    A feature that might help:  {feature}
    Why the feature might help: {why}
"""

    log.info("END get_analyst_description. Returning summary_string.")
    return summary_string

def inspect(df: pd.DataFrame, grain: str, log: logging.Logger) -> str:
    """Get a formatted inspection string from the data.

    Ask the data about itself.
    No need to type column names by hand - the data knows.

    Arguments:
        df: the loaded pandas DataFrame (a 2-dimensional table like an Excel sheet).
        grain: what one row represents.
        log: the logger to send progress messages to.

    Returns:
        a formatted multi-line string.
    """
    log.info("START inspect")

    # Get the count of rows.
    row_count: int = len(df)

    # Get the count of columns.
    column_count: int = len(df.columns)

    # Get a list of column names.
    column_names: list[str] = list(df.columns)

    # Get the first few rows of data
    first_rows: pd.DataFrame = df.head()

    # Log the facts Python discovered about the data.
    log.info("   row_count:    %s", row_count)
    log.info("   column_count: %s", column_count)
    log.info("   column_names: %s", column_names)

    # Build a readable string to return to the calling code.
    # Use a multi-line string (triple quotes) to make it easy to read.
    # Use a formatted string (f before the opening) so we can pass in information
    inspection_string: str = f"""
----------------
Data Inspection:
----------------
    One row means: {grain}
    Row count: {row_count}
    Column count: {column_count}
    Column names: {column_names}

---------------
First Few Rows:
---------------
{first_rows}
"""

    log.info("END inspect. Returning inspection_string.")
    return inspection_string