RUFAS.biophysical.animal.reproduction.reproduction module#

class RUFAS.biophysical.animal.reproduction.reproduction.Reproduction(heifer_reproduction_program: HeiferReproductionProtocol | None = None, heifer_reproduction_sub_program: HeiferTAISubProtocol | HeiferSynchEDSubProtocol | None = None, cow_reproduction_program: CowReproductionProtocol | None = None, cow_presynch_program: CowPreSynchSubProtocol | None = None, cow_ovsynch_program: CowTAISubProtocol | None = None, cow_resynch_program: CowReSynchSubProtocol | None = None, ai_day: int = 0, estrus_day: int = 0, abortion_day: int = 0, breeding_to_preg_time: int = 0, conception_rate: float = 0.0, cow_TAI_conception_rate: float = 0.0, num_conception_rate_decreases: int = 0, hormone_schedule: dict[int, dict[str, Any]] | None = None, gestation_length: int = 0, conceptus_weight: float = 0.0, calf_birth_weight: float = 0.0, calves: int = 0, calving_interval: int = 400, calving_interval_history: list[int] | None = None, body_weight_at_calving: float = 0.0, do_not_breed: bool = False, estrus_count: int = 0)#

Bases: object

Manages the reproduction protocol for heifers and cows, including artificial insemination (AI), hormone delivery, pregnancy checks, estrus detection, and pregnancy management.

Parameters#

heifer_reproduction_programHeiferReproductionProtocol, optional: The reproduction program for heifers, by default None.
heifer_reproduction_sub_programHEIFER_REPRODUCTION_SUB_PROTOCOLS, optional: The sub-program for heifer reproduction, by default None.
cow_reproduction_programCowReproductionProtocol, optional: The reproduction program for cows, by default None.
ai_dayint, optional: The day of artificial insemination, by default 0.
estrus_dayint, optional: The day estrus is expected, by default 0.
abortion_dayint, optional: The day of abortion, by default 0.
breeding_to_preg_timeint, optional: Time taken from breeding to pregnancy, by default 0.
conception_ratefloat, optional: Conception rate of the animal, by default 0.0.
TAI_conception_ratefloat, optional: Timed artificial insemination (TAI) conception rate, by default 0.0. We will only initialize the TAI conception rate for cows if this value is passed in as an input. The TAI conception rate for heifers will be dynamically determined later on.
num_conception_rate_decreasesint, optional: Number of times the conception rate decreases, by default 0.
hormone_scheduledict[int, dict[str, Any]], optional: The hormone schedule for the reproduction protocol, by default None.
gestation_lengthint, optional: Length of the gestation period, by default 0.
conceptus_weightfloat, optional: Weight of the conceptus, by default 0.0.
calf_birth_weightfloat, optional: Birth weight of the calf, by default 0.0.
calvesint, optional: Number of calves, by default 0.
calving_intervalint, optional: Interval between calvings, by default 0.
calving_interval_historylist[int], optional: History of calving intervals, by default None.
body_weight_at_calvingfloat, optional: Body weight of the animal at calving, by default 0.0.
do_not_breedbool, optional: Flag indicating if the animal should not breed, by default False.

Notes#

This class mutates a ReproductionDataStream object during the update process, relying on the side effect. This could lead to potential unexpected behaviors.

__init__(heifer_reproduction_program: HeiferReproductionProtocol | None = None, heifer_reproduction_sub_program: HeiferTAISubProtocol | HeiferSynchEDSubProtocol | None = None, cow_reproduction_program: CowReproductionProtocol | None = None, cow_presynch_program: CowPreSynchSubProtocol | None = None, cow_ovsynch_program: CowTAISubProtocol | None = None, cow_resynch_program: CowReSynchSubProtocol | None = None, ai_day: int = 0, estrus_day: int = 0, abortion_day: int = 0, breeding_to_preg_time: int = 0, conception_rate: float = 0.0, cow_TAI_conception_rate: float = 0.0, num_conception_rate_decreases: int = 0, hormone_schedule: dict[int, dict[str, Any]] | None = None, gestation_length: int = 0, conceptus_weight: float = 0.0, calf_birth_weight: float = 0.0, calves: int = 0, calving_interval: int = 400, calving_interval_history: list[int] | None = None, body_weight_at_calving: float = 0.0, do_not_breed: bool = False, estrus_count: int = 0) → None#

reproduction_update(reproduction_inputs: ReproductionInputs, time: RufasTime) → ReproductionOutputs#

Update the reproduction status of the animal.

Parameters#

reproduction_inputsReproductionInputs: The inputs for the reproduction protocol.
timeRufasTime: The current time in the simulation.

Returns#

ReproductionOutputs: Updated reproduction outputs for the animal.

heiferII_reproduction_update(reproduction_data_stream: ReproductionDataStream, time: RufasTime) → ReproductionDataStream#

Update reproduction status for heiferII based on the protocol.

Parameters#

reproduction_data_streamReproductionDataStream: Current reproduction datastream.
timeRufasTime: The current time in the simulation.

Returns#

ReproductionDataStream: Updated reproduction datastream for the heifer.

cow_reproduction_update(reproduction_data_stream: ReproductionDataStream, time: RufasTime) → ReproductionDataStream#

Update reproduction status for cows based on the protocol.

Parameters#

reproduction_data_streamReproductionDataStream: Current reproduction datastream.
timeRufasTime: The current time in the simulation.

Returns#

ReproductionDataStream: Updated reproduction datastream for the cow.

_validate_cow_reproduction_program() → None#: Ensures the cow reproduction program is one of the valid protocols.

_update_cow_repro_program_and_log_repro_stats_if_needed(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Updates the cow reproduction program to match configuration, logging any pre-existing state.

_execute_cow_reproduction_protocols(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Executes the relevant cow reproduction protocol methods based on current program and state.

_handle_cow_ai_day(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Performs AI and related updates if the current day matches the cow’s AI day.

_execute_heifer_reproduction_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Executes the appropriate heifer reproduction protocol based on the current reproduction program.

_add_cow_give_birth_events(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Adds the new_birth event and records the parity for cow giving birth.

_update_cow_repro_program_and_reset_repro_state_if_needed(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Updates the cow’s reproduction program and resets reproduction state if it differs from the configured program.

_simulate_estrus_if_eligible(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Simulates estrus if the cow reproduction program is ED or ED_TAI.

cow_give_birth(reproduction_data_stream: ReproductionDataStream, time: RufasTime) → ReproductionDataStream#: Handle the birth of a calf, resetting reproduction states and updating outputs.

_set_heifer_reproduction_program(reproduction_data_stream: ReproductionDataStream, simulation_day: int, repro_program: HeiferReproductionProtocol) → ReproductionDataStream#: Set the reproduction program for a heifer if it does not match the current program.

_set_cow_reproduction_program(reproduction_data_stream: ReproductionDataStream, simulation_day: int, repro_program: CowReproductionProtocol) → ReproductionDataStream#: Set the reproduction program for a cow if it does not match the current program.

_simulate_first_estrus(reproduction_data_stream: ReproductionDataStream, start_day: int, simulation_day: int, estrus_note: str, avg_estrus_cycle: float, max_cycle_length: float = inf) → ReproductionDataStream#

Calculate and set first next estrus day for an heiferII.

Parameters#

start_dayint: The day to begin the estrus cycle calculation.
simulation_dayint: The current simulation day.
estrus_notestr: Note explaining the reason for estrus simulation.
avg_estrus_cyclefloat: Average length of the estrus cycle.
max_cycle_lengthfloat, optional: Maximum allowable length for the estrus cycle, by default inf.

Returns#

ReproductionDataStream: Updated reproduction datastream after estrus simulation.

_simulate_estrus(reproduction_data_stream: ReproductionDataStream, start_day: int, simulation_day: int, estrus_note: str, avg_estrus_cycle: float, std_estrus_cycle: float, max_cycle_length: float = inf) → ReproductionDataStream#

Calculate and set the next estrus day for the animal.

Parameters#

start_dayint: The day to begin the estrus cycle calculation.
simulation_dayint: The current simulation day.
estrus_notestr: Note explaining the reason for estrus simulation.
avg_estrus_cyclefloat: Average length of the estrus cycle.
std_estrus_cyclefloat: Standard deviation of the estrus cycle length.
max_cycle_lengthfloat, optional: Maximum allowable length for the estrus cycle, by default inf.

Returns#

ReproductionDataStream: Updated reproduction datastream after estrus simulation.

_detect_estrus(detection_rate: float) → bool#: Determine if estrus is detected based on a randomized comparison with a detection rate.

execute_heifer_ed_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the estrus detection (ED) protocol for heifers.

_handle_generic_estrus_detection(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Perform general estrus detection, using specific functions for detected and non-detected cases.

_handle_estrus_detection(reproduction_data_stream: ReproductionDataStream, simulation_day: int, on_estrus_detected: Callable[[ReproductionDataStream, int], ReproductionDataStream], on_estrus_not_detected: Callable[[ReproductionDataStream, int], ReproductionDataStream]) → ReproductionDataStream#: Handle estrus detection and call specific functions when detected and not detected.

_handle_estrus_detected(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Perform actions associated with estrus detection in the ED protocol.

_handle_estrus_not_detected(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Perform actions when estrus is not detected in the ED protocol.

_deliver_hormones(reproduction_data_stream: ReproductionDataStream, hormones: list[str], delivery_day: int, simulation_day: int) → ReproductionDataStream#: Deliver hormones on the specified day.

_execute_hormone_delivery_schedule(reproduction_data_stream: ReproductionDataStream, simulation_day: int, schedule: dict[int, dict[str, Any]]) → ReproductionDataStream#: Execute hormone delivery schedule on the specified days.

execute_heifer_tai_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the Timed Artificial Insemination (TAI) protocol for heifers.

execute_heifer_synch_ed_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the SynchED protocol for heifers.

_set_up_hormone_schedule(reproduction_data_stream: ReproductionDataStream, start_from: int, reproduction_sub_protocol: str) → ReproductionDataStream#: Set up the hormone delivery schedule for heifers or cows.

_handle_synch_ed_hormone_delivery_and_set_estrus_day(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Deliver hormones and set the estrus day for heifers in the SynchED protocol.

_handle_synch_ed_estrus_detection(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle estrus detection for heifers in the SynchED program.

_handle_estrus_not_detected_in_synch_ed(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#

Handle the case where estrus is not detected for heifers in the SynchED program.

If estrus is not detected, this method updates the reproduction program and sets up the next steps according to the fallback protocol.

open_heifer(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle the status of an open heifer (post-abortion or pregnancy loss).

_perform_ai(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Perform artificial insemination (AI) on the animal.

_increment_heifer_ai_counts(reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Increment the AI counts for heifers.

_increment_successful_heifer_conceptions(reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Increment the counts of successful conceptions for heifers.

_handle_successful_heifer_conception(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle a successful conception event for a heifer.

_handle_failed_heifer_conception(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle a failed conception event for a heifer.

_calculate_gestation_length() → int#: Calculate the gestation length for the heifer.

_calculate_calf_birth_weight(breed: Breed) → float#

Calculates the birth weight of the calf based on the breed.

Notes#

[AN.BWT.6]

breed: Breed
The breed of the animal.

float
The birth weight for the calf (kg).

_initialize_pregnancy_parameters(reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Initialize parameters related to pregnancy for the heifer.

heifer_pregnancy_update(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Update the pregnancy status for a heifer.

_handle_heifer_pregnancy_check(reproduction_data_stream: ReproductionDataStream, pregnancy_check_config: PregnancyCheckConfig, simulation_day: int) → ReproductionDataStream#: Handle pregnancy checks for heifers and take appropriate actions based on results.

_handle_cow_pregnancy_check(reproduction_data_stream: ReproductionDataStream, pregnancy_check_config: PregnancyCheckConfig, simulation_day: int) → ReproductionDataStream#: Handle a pregnancy check for cows and update the state based on the outcome.

_terminate_pregnancy(reproduction_data_stream: ReproductionDataStream, preg_loss_const: str, simulation_day: int) → ReproductionDataStream#: Terminate pregnancy and reset related parameters if pregnancy is lost.

_calculate_conception_rate_on_ai_day() → None#: Adjust conception rate on the AI day based on breeding history and parity.

execute_cow_ed_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the estrus detection (ED) protocol for cows.

_enter_ovsynch_repro_state(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Updates the reproduction state to ‘IN_OVSYNCH’

_repeat_estrus_simulation_before_vwp(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Repeat estrus simulation for cows before the voluntary waiting period (VWP).

_setup_ai_day_after_estrus_detected(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Set up the AI day for cows when estrus is detected.

_simulate_full_estrus_cycle(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Simulate a full estrus cycle when estrus is not detected.

_simulate_full_estrus_cycle_before_ovsynch(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Simulate an estrus cycle before the OvSynch program in the ED-TAI protocol.

_execute_cow_hormone_delivery_schedule(reproduction_data_stream: ReproductionDataStream, simulation_day: int, schedule: dict[int, dict[str, Any]]) → ReproductionDataStream#: Execute hormone delivery schedule for cows.

execute_cow_tai_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the TAI protocol for cows, setting up hormone delivery schedules for presynch and ovsynch programs.

_reset_ai_day_if_needed(program_to_enter: str, reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Resets the pre-existing AI day for an animal to enter PreSynch or OvSynch.

_setup_presynch_on_presynch_start_day_if_valid(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Set up the presynch program for cows on the start day if applicable.

_enter_fresh_state_if_in_empty_state(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Enter the fresh state for cows if the current reproduction state is empty.

_setup_ovsynch_on_ovsynch_start_day_if_valid(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Set up an OvSynch program for cows on the OvSynch start day if applicable.

_should_set_up_hormone_delivery_for_presynch(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → tuple[bool, ReproductionDataStream]#: Determine if hormone delivery should be set up for presynch based on current status.

_should_set_up_hormone_delivery_for_ovsynch(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → tuple[bool, ReproductionDataStream]#: Determine if hormone delivery should be set up for OvSynch based on current status.

_increment_cow_ai_counts(reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Increment AI counts for cows.

_increment_successful_cow_conceptions(reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Increment successful conception counts for cows.

execute_cow_ed_tai_protocol(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Execute the ED-TAI protocol for cows, monitoring days in milk and scheduling AI or estrus detection.

_handle_estrus_not_detected_before_ovsynch_start_day(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Redirect cow to OvSynch if estrus was not detected before the OvSynch start day.

_decrease_conception_rate_by_parity(calves: int, conception_rate: float) → float#: Adjust conception rate based on cow parity.

_handle_successful_cow_conception(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle successful conception for cows, setting pregnancy parameters and scheduling resynch as needed.

_handle_failed_cow_conception(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle failed conception for cows, scheduling appropriate repro actions based on protocol.

cow_pregnancy_update(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Update pregnancy status for cows, performing checks on designated pregnancy check days.

_check_do_not_breed_flag(simulation_day: int, reproduction_data_stream: ReproductionDataStream) → ReproductionDataStream#: Check if cow should be marked as do-not-breed if not pregnant beyond breeding window.

open_cow(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Handle an open cow’s status, determining next steps based on reproduction protocol and resynch program.

_handle_open_cow_in_pgf_at_pd_resynch(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Deliver a PGF injection and simulate a short estrus cycle for an open cow in the PGFatPD protocol.

_handle_open_cow_in_tai_before_pd_resynch(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Redirect an open cow in the TAIbeforePD protocol to the OvSynch program and stop estrus detection.

_schedule_ovsynch_program_in_advance(reproduction_data_stream: ReproductionDataStream, simulation_day: int, days_before_first_preg_check: int = 6) → ReproductionDataStream#: Schedule an OvSynch program in advance for cows in the TAIbeforePD resynch protocol.

_exit_ovsynch_program_early_when_first_preg_check_passed_or_estrus_detected(reproduction_data_stream: ReproductionDataStream, simulation_day: int) → ReproductionDataStream#: Exit the OvSynch program early in the TAIbeforePD protocol when a pregnancy check is passed or estrus is detected.