Refactor Pen.get_manure_stream()

README.md

@@ -1,7 +1,7 @@
 [![Flake8](https://img.shields.io/badge/Flake8-passed-brightgreen)](https://github.com/RuminantFarmSystems/MASM/actions/workflows/combined_format_lint_test_mypy.yml)
 [![Pytest](https://img.shields.io/badge/Pytest-passed-brightgreen)](https://github.com/RuminantFarmSystems/MASM/actions/workflows/combined_format_lint_test_mypy.yml)
 [![Coverage](https://img.shields.io/badge/Coverage-99%25-brightgreen)](https://github.com/RuminantFarmSystems/MASM/actions/workflows/combined_format_lint_test_mypy.yml)
-[![Mypy](https://img.shields.io/badge/Mypy-1214%20errors-red)](https://github.com/RuminantFarmSystems/MASM/actions/workflows/combined_format_lint_test_mypy.yml)
+[![Mypy](https://img.shields.io/badge/Mypy-1215%20errors-red)](https://github.com/RuminantFarmSystems/MASM/actions/workflows/combined_format_lint_test_mypy.yml)
 
 
 # RuFaS: Ruminant Farm Systems

RUFAS/biophysical/animal/pen.py

@@ -651,32 +651,75 @@ def get_manure_streams(self) -> dict[str, ManureStream]:
 
         """
         animal_manure_streams: dict[str, ManureStream] = {}
-        if self.animal_combination == AnimalCombination.GROWING_AND_CLOSE_UP:
-            total_animals_in_pen = len(self.animals_in_pen)
-            num_growing = len(
-                [
-                    animal
-                    for animal in self.animals_in_pen.values()
-                    if animal.animal_type in [AnimalType.HEIFER_I, AnimalType.HEIFER_II]
-                ]
-            )
-            num_close_up = len(
-                [
-                    animal
-                    for animal in self.animals_in_pen.values()
-                    if animal.animal_type in [AnimalType.HEIFER_III, AnimalType.DRY_COW]
-                ]
-            )
-            methane_production_potential = (
-                (0.17 * num_growing / total_animals_in_pen + 0.24 * num_close_up / total_animals_in_pen)
-                if total_animals_in_pen > 0
-                else 0.0
-            )
-        else:
-            methane_production_potential = (
-                0.17 if self.animal_combination in [AnimalCombination.CALF, AnimalCombination.GROWING] else 0.24
+        total_stream = self._calculate_total_pen_manure_stream()
+
+        general_stream_proportion, parlor_stream_proportion, parlor_stream = self._handle_parlor_stream(total_stream)
+        if parlor_stream_proportion is not None or general_stream_proportion < 1.0 or parlor_stream is not None:
+            base_parlor_stream_name = f"{self.parlor_stream_name}" if self.parlor_stream_name else "parlor_stream"
+            animal_manure_streams[f"{base_parlor_stream_name}_PEN_{self.id}"] = parlor_stream
+
+        self._validate_general_manure_stream_proportions()
+        for stream_config in self.manure_streams:
+            manure_stream = self._split_general_manure_stream(
+                general_stream_proportion, parlor_stream_proportion, stream_config, total_stream
             )
+            if manure_stream.pen_manure_data is not None:
+                manure_stream.pen_manure_data.set_first_processor(str(stream_config.get("first_processor")))
+            manure_stream = self._apply_bedding(manure_stream, str(stream_config.get("bedding_name")))
+            stream_name = f"{str(stream_config.get('stream_name'))}_{self.animal_combination.name}_PEN_{self.id}"
+            animal_manure_streams[stream_name] = manure_stream
+
+        return animal_manure_streams
+
+    def _split_general_manure_stream(
+        self,
+        general_stream_proportion: float,
+        parlor_stream_proportion: Any | None,
+        stream_config: dict[str, str | float],
+        total_stream: ManureStream,
+    ) -> ManureStream:
+        """
+        Split the total manure stream by the given general stream proportion..
+
+        Parameters
+        ----------
+        general_stream_proportion : float
+            The proportion of the total manure stream to be assigned to the general stream.
+        parlor_stream_proportion : float | None
+            The proportion of the total manure stream to be assigned to the parlor stream.
+        stream_config : dict[str, str | float]
+            The config of the target stream to be split.
+        total_stream : ManureStream
+            The total manure stream to be split from.
+
+        Returns
+        -------
+        ManureStream
+            The split manure stream.
+
+        """
+        general_substream_proportion = float(stream_config.get("stream_proportion", 1.0))
+        split_ratio = general_substream_proportion * general_stream_proportion
+        manure_stream_deposit_split = (
+            general_substream_proportion if parlor_stream_proportion is not None else split_ratio
+        )
+        manure_stream = total_stream.split_stream(
+            split_ratio=split_ratio,
+            stream_type=StreamType.GENERAL,
+            manure_stream_deposition_split=manure_stream_deposit_split,
+        )
+        return manure_stream
 
+    def _calculate_total_pen_manure_stream(self) -> ManureStream:
+        """
+        Calculates and returns the total manure stream for the pen.
+
+        Returns
+        -------
+        ManureStream
+            The aggregate manure stream for the pen.
+        """
+        methane_production_potential = self._calculate_methane_production_potential()
         pen_animal_excretions = self.total_manure_excretion
         total_pen_manure_data = PenManureData(
             num_animals=len(self.animals_in_pen),
@@ -687,7 +730,6 @@ def get_manure_streams(self) -> dict[str, ManureStream]:
             manure_urine_nitrogen=pen_animal_excretions.urine_nitrogen,
             stream_type=StreamType.GENERAL,
         )
-
         total_stream = ManureStream(
             water=pen_animal_excretions.manure_mass - pen_animal_excretions.total_solids,
             ammoniacal_nitrogen=pen_animal_excretions.manure_total_ammoniacal_nitrogen,
@@ -703,8 +745,30 @@ def get_manure_streams(self) -> dict[str, ManureStream]:
             pen_manure_data=total_pen_manure_data,
             bedding_non_degradable_volatile_solids=0.0,
         )
+        return total_stream
+
+    def _handle_parlor_stream(self, total_stream: ManureStream) -> tuple[float, float | None, ManureStream | None]:
+        """
+        Handles the processing of the parlor stream based on the type of animal combination and the given total stream.
+        If the animal combination indicates lactating cows, the parlor stream is split from the total stream using the
+        specified milking time proportion.
 
+        Parameters
+        ----------
+        total_stream : ManureStream
+            The complete manure stream that is available for processing. The function processes this stream to
+            derive the parlor stream proportion and data if applicable.
+
+        Returns
+        -------
+        tuple[float, float | None, ManureStream | None]
+            A tuple containing:
+            - general_stream_proportion (float): The proportion of the general stream that remains.
+            - parlor_stream_proportion (float or None): The proportion of the parlor stream, or None if not applicable.
+            - parlor_stream (ManureStream or None): The resulting parlor stream, or None if not applicable.
+        """
         parlor_stream_proportion = None
+        parlor_stream = None
         if self.animal_combination == AnimalCombination.LAC_COW:
             parlor_stream_proportion = self.minutes_away_for_milking / 1440
             general_stream_proportion = 1 - parlor_stream_proportion
@@ -715,31 +779,70 @@ def get_manure_streams(self) -> dict[str, ManureStream]:
             )
             if parlor_stream.pen_manure_data is not None and self.first_parlor_processor is not None:
                 parlor_stream.pen_manure_data.set_first_processor(self.first_parlor_processor)
-            base_parlor_stream_name = f"{self.parlor_stream_name}" if self.parlor_stream_name else "parlor_stream"
-            parlor_stream_name = f"{base_parlor_stream_name}_PEN_{self.id}"
-            animal_manure_streams[parlor_stream_name] = parlor_stream
         else:
             general_stream_proportion = 1.0
+        self._validate_parlor_stream_proportion(general_stream_proportion, parlor_stream, parlor_stream_proportion)
+        return general_stream_proportion, parlor_stream_proportion, parlor_stream
 
-        self._validate_general_manure_stream_proportions()
-        for stream in self.manure_streams:
-            general_substream_proportion = float(stream.get("stream_proportion", 1.0))
-            split_ratio = general_substream_proportion * general_stream_proportion
-            manure_stream_deposit_split = (
-                general_substream_proportion if parlor_stream_proportion is not None else split_ratio
+    def _validate_parlor_stream_proportion(
+        self,
+        general_stream_proportion: float,
+        parlor_stream: ManureStream | None,
+        parlor_stream_proportion: float | None,
+    ) -> None:
+        """
+        Validates that the parlor stream proportion is within the expected range.
+
+        Parameters
+        ----------
+        general_stream_proportion:  float
+            The proportion of the general stream that remains.
+        parlor_stream_proportion: float | None
+            The proportion of the parlor stream, or None if not applicable.
+        parlor_stream : ManureStream | None
+            The resulting parlor stream, or None if not applicable.
+        """
+        if parlor_stream_proportion is not None or general_stream_proportion < 1.0 or parlor_stream is not None:
+            assert parlor_stream is not None, "Parlor stream should not be None if parlor stream proportion is not None"
+            assert (
+                parlor_stream_proportion is not None
+            ), "Parlor stream proportion should not be None if parlor stream is not None"
+
+    def _calculate_methane_production_potential(self) -> float:
+        """
+        Calculates the methane production potential based on the combination and type of animals in the pen.
+
+        Returns
+        -------
+        float
+            The methane production potential for the animals in the pen.
+        """
+        if self.animal_combination == AnimalCombination.GROWING_AND_CLOSE_UP:
+            total_animals_in_pen = len(self.animals_in_pen)
+            num_growing = len(
+                [
+                    animal
+                    for animal in self.animals_in_pen.values()
+                    if animal.animal_type in [AnimalType.HEIFER_I, AnimalType.HEIFER_II]
+                ]
+            )
+            num_close_up = len(
+                [
+                    animal
+                    for animal in self.animals_in_pen.values()
+                    if animal.animal_type in [AnimalType.HEIFER_III, AnimalType.DRY_COW]
+                ]
             )
-            manure_stream = total_stream.split_stream(
-                split_ratio=split_ratio,
-                stream_type=StreamType.GENERAL,
-                manure_stream_deposition_split=manure_stream_deposit_split,
+            methane_production_potential = (
+                (0.17 * num_growing / total_animals_in_pen + 0.24 * num_close_up / total_animals_in_pen)
+                if total_animals_in_pen > 0
+                else 0.0
             )
-            if manure_stream.pen_manure_data is not None:
-                manure_stream.pen_manure_data.set_first_processor(str(stream.get("first_processor")))
-            manure_stream = self._apply_bedding(manure_stream, str(stream.get("bedding_name")))
-            stream_name = f"{str(stream.get('stream_name'))}_{self.animal_combination.name}_PEN_{self.id}"
-            animal_manure_streams[stream_name] = manure_stream
-
-        return animal_manure_streams
+        else:
+            methane_production_potential = (
+                0.17 if self.animal_combination in [AnimalCombination.CALF, AnimalCombination.GROWING] else 0.24
+            )
+        return methane_production_potential
 
     def _apply_bedding(self, manure_stream: ManureStream, bedding_name: str) -> ManureStream:
         """

changelog.md

@@ -49,6 +49,7 @@ v1.0.0
 - [2863](https://github.com/RuminantFarmSystems/MASM/pull/2863) - [minor change] [NoInputChange] [NoOutputChange] Updates TaskManager to avoid using multiprocessing when running single tasks.
 - [2854](https://github.com/RuminantFarmSystems/MASM/pull/2854) - [minor change] [NoInputChange] [NoOutputChange] Update `emissions.py` filtering process and remove `use_filter_key_name` option in the OM filter.
 - [2872](https://github.com/RuminantFarmSystems/RuFaS/pull/2872) - [minor change] [NoInputChange] [NoOutputChange] Adds information and links for onboarding videos.
+- [2850](https://github.com/RuminantFarmSystems/MASM/pull/2850) - [minor change] [NoInputChange] [NoOutputChange] Refactor `Pen.get_manure_stream()`.
 
 ### v1.0.0