RUFAS.biophysical.animal.herd_manager module#

class RUFAS.biophysical.animal.herd_manager.HerdManager(weather: Weather, time: RufasTime, is_ration_defined_by_user: bool, available_feeds: list[Feed], feed_emissions_estimator: PurchasedFeedEmissionsEstimator | None = None)#

Bases: object

DEFAULT_NUM_STALLS_BY_COMBINATION = {AnimalCombination.CALF: 110, AnimalCombination.CLOSE_UP: 200, AnimalCombination.GROWING: 800, AnimalCombination.GROWING_AND_CLOSE_UP: 500, AnimalCombination.LAC_COW: 850}#

ANIMAL_GROUPING_SCENARIO: AnimalGroupingScenario#

classmethod set_animal_grouping_scenario(scenario: AnimalGroupingScenario) → None#

Sets the animal grouping scenario to the given scenario.

Parameters#

scenarioAnimalGroupingScenario: The scenario to set the animal grouping scenario to.

__init__(weather: Weather, time: RufasTime, is_ration_defined_by_user: bool, available_feeds: list[Feed], feed_emissions_estimator: PurchasedFeedEmissionsEstimator | None = None) → None#

Initializes the pens and the animal herd in the simulation with data from user inputs.

Parameters#

weatherWeather: instance of the Weather class
timeRufasTime: instance of the RufasTime class
is_ration_defined_by_userbool: True if user-defined rations are used for the herd, otherwise false.
available_feedslist[Feed]: Nutrition information of feeds available to formulate animals rations with.
feed_emissions_estimatorPurchasedFeedEmissionsEstimator, default=None: Instance of the PurchasedFeedEmissionsEstimator class.

property animals_by_type: dict[AnimalType, list[Animal]]#

Group animals by type.

Returns#

dict[AnimalType, list[Animal]]: A dictionary where each key corresponds to an AnimalType enum and each value is a list of Animal objects belonging to that type.

property animals_by_combination: dict[AnimalCombination, list[Animal]]#

Group animals by combination.

Returns#

dict[AnimalCombination, list[Animal]]: A dictionary where the keys are instances of AnimalCombination, and the values are lists of Animal instances belonging to the corresponding combination.

property pens_by_animal_combination: dict[AnimalCombination, list[Pen]]#

Group a list of pens by animal combination.

Returns#

Dict[AnimalCombination, list[Pen]]: Dictionary of pens grouped by animal combination.

property phosphorus_concentration_by_animal_class: dict[AnimalType, float]#

Retrieves the phosphorus concentration for each animal type.

Returns#

dict[AnimalType, float]: A dictionary mapping each animal type to its corresponding phosphorus concentration. If the total body weight of an animal class is zero, the phosphorus concentration is set to 0.0 for that class.

Notes#

These variables are the P concentrations of each class of animal. They are calculated daily and are used when an animal is added to the herd, whether by birth or replacement herd purchase. They are calculated by dividing the total P in the animals of the class by the total body weight of the animals, on a per-animal basis.

property current_herd_size: int#

Calculates the current size of the herd based on the number of heiferIIIs and cows.

Returns#

int: The current size of the herd.

collect_daily_feed_request() → RequestedFeed#

Collects total amount of feeds needed for all animals on the current day.

Returns#

dict[RUFAS_ID, float]: Mapping of the feed ID’s requested to the amounts of feed (kg dry matter).

print_herd_snapshot(txt: str) → None#

Prints a formatted snapshot of the herd, showing the current count of different categories of cattle such as calves, heifer groups, and cows.

Parameters#

txtstr: A descriptive prefix to the snapshot summary, typically used to identify the context or timestamp of the snapshot.

_print_animal_num_warnings(herd_data: dict[str, Any]) → None#

If simulate_animals is false, creates warnings if there are more than 0 animals for any of the animal types,: and logs how many warnings were generated

Otherwise, if simulate_animals is true, logs that it is true

Parameters#

herd_dataDict[str, Any]: dictionary containing information about the herd

_reset_daily_statistics() → None#: Reset the daily herd statistics.

_update_sold_animal_statistics(sold_newborn_calves: list[Animal], sold_heiferIIs: list[Animal], sold_and_died_cows: list[Animal]) → None#: Call the corresponding functions to update the statistics for sold animals

_perform_daily_routines_for_animals(time: RufasTime, animals: list[Animal]) → tuple[list[Animal], list[Animal], list[Animal], list[Animal]]#: Perform daily routines for a given list of animals.

_update_herd_structure(graduated_animals: list[Animal], newborn_calves: list[Animal], newly_added_animals: list[Animal], removed_animals: list[Animal], available_feeds: list[Feed], current_day_conditions: CurrentDayConditions, total_inventory: TotalInventory) → None#: Call the corresponding functions to update the herd structure and reassign animals to new pens.

daily_routines(available_feeds: list[Feed], time: RufasTime, weather: Weather, total_inventory: TotalInventory) → dict[str, ManureStream]#

Perform daily routines for managing animal herds and updating associated data.

This method handles all daily activities related to the management of animal herds, including animal transitions (graduation, removal), sales, births, updates to herd statistics, and manure data collection.

Parameters#

available_feedslist[Feed]: A list of feed resources available for the day.
timeRufasTime: An instance of the RufasTime object representing the current time and simulation day.
weatherWeather: An object providing weather conditions affecting herd activities.
total_inventoryTotalInventory: Object representing the total inventory of herd-related resources.

Returns#

dict[str, ManureStream]: A list of dictionaries containing manure data for each pen in the herd.

_create_newborn_calf(newborn_calf_config: NewBornCalfValuesTypedDict, simulation_day: int) → Animal#

Creates a new newborn calf instance and records its entry event in the herd if it is not sold.

Parameters#

newborn_calf_configNewBornCalfValuesTypedDict: Configuration for the newborn calf containing its attributes.
simulation_dayint: The current day in the simulation.

Returns#

Animal: An instance of the Animal class representing the newly created newborn calf.

_check_if_heifers_need_to_be_sold(simulation_day: int) → list[Animal]#

Checks if surplus heifers need to be sold based on herd size.

This method evaluates if the current number of heifers and cows exceeds a specified threshold (defined as 3% over the herd statistics’ target herd size). If the threshold is surpassed, heiferIIIs are removed from the herd until the herd size falls within the acceptable range.

Parameters#

simulation_dayint: The simulation day on which the check and potential sale is conducted.

Returns#

list[Animal]: A list of heiferIIIs to be sold.

_check_if_replacement_heifers_needed(time: RufasTime) → list[Animal]#

Checks if replacement heifers are needed to maintain the herd size.

This function determines whether additional heiferIIIs need to be added to the herd based on the current herd size, purchase thresholds, and the availability of heifers in the replacement market.

Parameters#

timeRufasTime: An instance of the RufasTime class providing the current simulation day and date.

Returns#

list[Animal]: A list of heiferIIIs bought.

_remove_animal_from_current_array(animal: Animal) → None#

Remove an animal object from the current array that it belongs to.

Parameters#

animalAnimal: The animal instance to be removed from its current array.

_add_animal_to_new_array(animal: Animal) → None#

Adds an animal to the appropriate array based on its type.

Parameters#

animalAnimal: The animal object to be added to the respective array based on its animal_type.

_update_animal_array(animal: Animal) → None#

Updates the internal animal array by removing the given animal from its current array and adding it to a new array.

Parameters#

animalAnimal: The animal object to update in the internal arrays.

_handle_graduated_animals(graduated_animals: list[Animal], available_feeds: list[Feed], current_day_conditions: CurrentDayConditions, total_inventory: TotalInventory) → None#

Reassigns animals that have graduated to a new pen, and updates the pen id map.

Parameters#

graduated_animalslist[Animal]: List of animals that have graduated and need to be reassigned to a new pen.
available_feedslist[Feed]: Nutrition information of feeds available to formulate animals rations with.
current_day_conditionsCurrentDayConditions: Object representing the current conditions of the day.
total_inventoryTotalInventory: Inventory currently available or projected to be available at a future date.

_handle_newly_added_animals(new_animals: list[Animal], available_feeds: list[Feed], current_day_conditions: CurrentDayConditions, total_inventory: TotalInventory) → None#

Adds newly added animals to their appropriate pen and updates the pen id map.

Parameters#

new_animalslist[Animal]: List of newly added animals.
available_feedslist[Feed]: Nutrition information of feeds available to formulate animals rations with.
current_day_conditionsCurrentDayConditions: Object representing the current conditions of the day.
total_inventoryTotalInventory: Inventory currently available or projected to be available at a future date.

_remove_animal_from_pen_and_id_map(animal: Animal) → None#

Removes animal from its current pen, and removes it from the pen id map.

Parameters#

animalAnimal: The animal to be removed from its current pen and the pen id map.

_add_animal_to_pen_and_id_map(animal: Animal, available_feeds: list[Feed], current_day_conditions: CurrentDayConditions, total_inventory: TotalInventory) → None#

Adds animal to pen with the lowest stocking density, and updates the pen id map accordingly.

Parameters#

animalAnimal: The animal to be added to a pen.
available_feedslist[Feed]: Nutrition information of feeds available to formulate animals rations with.
current_day_conditionsCurrentDayConditions: Object representing the current conditions of the day.
total_inventoryTotalInventory: Inventory currently available or projected to be available at a future date.

initialize_pens(all_pen_data: list[dict[str, Any]]) → None#

Populates the list of pens with the information from the input json file.

Parameters#

all_pen_data: list[dict[str, Any]]: List containing information about the pens.

allocate_animals_to_pens(simulation_day: int) → None#: Allocate animals to pens based on the current animal population and the number of pens available. This method distributes the animals among the pens, ensuring that the animal density of each pen matches the overall density as closely as possible.

_plan_animal_allocation(num_animals: int, max_spaces_in_pens: list[int]) → list[int]#

Make an allocation plan to distribute animals across pens based on overall pen density, allowing controlled overstocking if the number of animals exceeds total pen capacity.

General rules: 1. Animals are allocated proportionally across pens based on overall density, ensuring even distribution relative to pen capacity. 2. Each pen receives animals up to a calculated allocation limit: ceil(overall_density * pen_capacity). 3. If the total number of animals exceeds the sum of all pen capacities, the excess animals are distributed proportionally, allowing pens to exceed capacity. 4. Warnings are logged for any pen that becomes overstocked. 5. All animals are guaranteed to be allocated.

Notes#

This allocation strategy prioritizes proportional and fair distribution by calculating an overall density and applying it to each pen’s capacity. The result ensures that pen densities remain consistent even under overstocking scenarios.

Pens are sorted by allocation limit, and animals are allocated in that order. The final pen receives any remaining animals to guarantee full allocation.

Overstocking is permitted when necessary and is handled fairly based on capacity-derived allocation limits. Logging ensures that overstocked pens are tracked for review.

Parameters#

num_animalsint: The total number of animals to allocate. Must be a non-negative integer.
max_spaces_in_penslist[int]: A list of integers representing the maximum number of animals each pen can accommodate without overstocking. Each integer must be positive.

Returns#

list[int]: A list of integers representing the number of animals allocated to each pen. Each value may exceed the pen’s capacity if overstocking occurs.

Raises#

AssertionError: If the total number of allocated animals does not match the number of animals provided.

Examples#

>>> _plan_animal_allocation(num_animals=90, max_spaces_in_pens=[30, 30, 30], simulation_day=1)
[30, 30, 30]

>>> _plan_animal_allocation(num_animals=95, max_spaces_in_pens=[30, 30, 30], simulation_day=1)
[32, 32, 31]

>>> _plan_animal_allocation(num_animals=47, max_spaces_in_pens=[20, 15, 10], simulation_day=1)
[20, 16, 11]  # Overstocked due to animal count exceeding total capacity

>>> _plan_animal_allocation(num_animals=70, max_spaces_in_pens=[50, 30, 20], simulation_day=1)
[35, 21, 14]

_execute_allocation_plan(allocation_plan: list[int], animals: list[Animal], animal_pens: list[Pen]) → None#

Execute an allocation plan to distribute animals into pens according to the given plan.

This method iterates over the provided allocation plan and updates each pen with the specified number of animals.

Parameters#

allocation_planlist[int]: A list of integers representing the number of animals to be allocated to each pen. The length of the allocation_plan list must match the number of pens in animal_pens.
animalslist[Animal]: A list of animals to be allocated among the pens.
animal_penslist[Pen]: A list of Pen objects representing the pens to which animals will be allocated.

Raises#

ValueError: If the length of the allocation plan does not match the number of pens. If the sum of the allocation plan does not match the number of animals.

_sort_cows_before_allocation() → None#: Sort cows by days_in_milk in increasing order.

_calculate_max_animal_spaces_per_pen(num_stalls: int, max_stocking_density: float) → int#

Calculate the maximum number of animal spaces available per pen based on the user density.

Parameters#

num_stallsint: The number of stalls in the pen. Must be greater than or equal to 0.
max_stocking_densityfloat: The maximum stocking density for the pen. Must be greater than or equal to 0.

Returns#

int: The maximum number of animal spaces available in the pen.

Raises#

ValueError: If the number of stalls or maximum stocking density is less than 0.

Examples#

>>> HerdManager._calculate_max_animal_spaces_per_pen(num_stalls=10, max_stocking_density=1.5)
15
>>> HerdManager._calculate_max_animal_spaces_per_pen(num_stalls=5, max_stocking_density=2.0)
10

_allocate_animals_to_pens_helper(animals: list[Animal], pens: list[Pen]) → None#

Allocate animals to pens based on overall density while preventing overcrowding.

This method distributes the animals among the available pens, ensuring that the density in each pen matches the overall density as closely as possible.

Parameters#

animalsList[Union[Calf, HeiferI, HeiferII, HeiferIII, Cow]]: A list of animal to be allocated to pens.
pensList[Pen]: A list of Pen objects representing the available pens. All these pens should have the same animal combination.

fully_update_animal_to_pen_id_map(simulation_day: int) → None#: Updates the entire animal_to_pen_id_map dictionary so that each animal’s ID is associated with the pen that animal is in.

_gather_pen_history(animal_type_list: list[Animal], simulation_day: int) → None#

Updates pen history data for a given animal type.

Checks the current pen ID and pen composition of all animals for a given animal class, and then updates the pen history for that type using the update_pen_history() method.

Parameters#

animal_type_listList[Animal]: List of animals.
simulation_dayint: The current simulation day.

record_pen_history(simulation_day: int) → None#

Records the pen history of all the animals.

Parameters#

simulation_dayint: The current simulation day.

clear_pens() → None#: Removes animals from pens for re-allocation. This is part of the routines that happen every ration interval.

end_ration_interval(simulation_day: int) → bool#

Checks if a new ration should be formulated for the current simulation_day.

Returns#

bool: True if today is the day a new ration has to be formulated, False otherwise.

set_milk_type_in_calf_ration_manager() → None#: Sets the milk type of calves to be either whole or replacement depending on the diet configured by the user.

initialize_nutrient_requirements(weather: Weather, time: RufasTime, available_feeds: list[Feed]) → None#

Calculates initial nutrient requirements at the beginning of the simulation for initial pen allocation.

Parameters#

weatherWeather: instance of the Weather class
timeRufasTime: instance of the RufasTime class
available_feedslist[Feed]: Nutrition information of feeds available to formulate animals rations with.

update_all_max_daily_feeds(total_inventory: TotalInventory, next_harvest_dates: dict[int, date], time: RufasTime) → IdealFeeds#

Updates the max feeds of all available feeds types based on the current total inventory.

Parameters#

total_inventoryTotalInventory: The total inventory of all available feeds.
next_harvest_datesDict[RUFAS_ID, date]: The next harvest date for each applicable feed type.
timeRufasTime: RufasTime object.

Returns#

IdealFeeds: The maximum daily feeds for each feed type.

_update_single_max_daily_feed(rufas_id: int, next_harvest: date, total_inventory: TotalInventory, time: RufasTime) → None#

Updates a single max daily feed based on the current amount available, number of animals, and next harvest date.

Parameters#

rufas_idRUFAS_ID: The RuFaS Feed ID of the max daily feed to be updated.
next_harvestdate: When next harvest of the given RuFaS feed will be.
total_inventoryTotalInventory: Total amounts of feeds in inventory.
timeRufasTime: RufasTime object.

_find_pen_available_feeds(all_available_feeds: list[Feed], user_defined_ration_feed_ids: list[int]) → list[Feed]#: Find the available feeds for the pen.

formulate_rations(available_feeds: list[Feed], current_temperature: float, ration_interval_length: int, total_inventory: TotalInventory, simulation_day: int) → RequestedFeed#

Formulates rations for all pens.

Parameters#

available_feedsList[Feed]: List of available feeds.
current_temperaturefloat: Current temperature (C).
ration_interval_lengthint: Length of the ration interval (days).
total_inventoryTotalInventory: The total inventory of all available feeds.

Returns#

RequestedFeed: Feeds requested to be purchased for the newly formulated rations.

_reformulate_ration_single_pen(pen: Pen, pen_available_feeds: list[Feed], current_temperature: float, total_inventory: TotalInventory) → None#

Reformulates ration for a single pen.

Parameters#

penPen: Pen that requires ration reformulation.
pen_available_feedsList[Feed]: List of available feeds in this pen.
current_temperaturefloat: Current temperature (C).
total_inventoryTotalInventory: Inventory currently available or projected to be available at a future date.

update_herd_statistics() → None#: Calculates and updates herd statistics.

_calculate_herd_percentages() → None#: Calculates and updates the herd percentages for different animal types.

_calculate_cow_percentages() → None#: Calculates percentages of various cow categories within the herd and updates the corresponding attributes of the herd_statistics object.

_calculate_cull_reason_percentages() → None#: Calculates the percentage distribution for each culling reason in the herd statistics.

_update_cow_parity_statistics() → None#

Updates statistics related to the parity of cows in the herd.

Parity-related statistics include:

the number of cows for each parity level
the average age of cows at different parities
the average calving age for each parity
the average time between calving and subsequent pregnancy

This method also calculates the percentage distribution of cows across parity levels relative to the herd population. All computed statistics are stored in the herd_statistics attribute of the class.

_update_cow_milking_statistics() → None#

Updates the herd’s milking statistics.

This method performs calculations for both lactating and dry cows and updates the herd_statistics. The metrics include:

average days in milk

daily milk production

milk fat content

milk protein content

voluntary waiting period statistics.

Raises#

ValueError: If any milk production, fat content, or protein content is detected from dry cows. An error entry is also added to the error log with details on the issue.

_update_cow_pregnancy_statistics() → None#

Updates the pregnancy statistics for the cows in the herd.

This method calculates and updates the statistics related to pregnant cows, open (non-pregnant) cows, and the average number of days in pregnancy.

_update_sold_and_died_cow_statistics(sold_and_died_cows: list[Animal]) → None#

Updates the herd statistics with details of cows that are sold or have died. This method records the culling age, updates statistics related to culled cows, and categorizes the cows based on specific attributes such as cull reason and parity.

Parameters#

sold_and_died_cowslist[Animal]: A list of cows that were either sold or died.

_update_sold_heiferII_statistics(sold_heiferIIs: list[Animal]) → None#

Updates sold heiferII statistics in the herd statistics.

This method updates the herd’s statistical values relating to sold heiferIIs. The updates include incrementing the number of sold heiferIIs, appending details about each sold heiferII, and calculating the average heiferII culling age.

Parameters#

sold_heiferIIslist[Animal]: A list of heiferII animals that have been sold.

_update_sold_newborn_calf_statistics(sold_newborn_calves: list[Animal]) → None#

Updates the statistics of sold newborn calves in the herd statistics. It increments the count of sold calves and appends detailed information about each sold newborn calf to the corresponding statistics.

Parameters#

sold_newborn_calveslist[Animal]: A list of newborn calves that were sold.

_update_cow_reproduction_statistics() → None#: Updates the reproduction statistics of cows in the herd.

_update_heifer_reproduction_statistics() → None#: Updates the reproduction statistics of heifers in the herd.

_update_average_mature_body_weight() → None#: Updates the average mature body weight of the animals in the herd.

_update_average_cow_body_weight() → None#: Updates the average body weight of cows in the herd.

_update_average_cow_parity() → None#: Updates the average cow parity number in the herd statistics.