pldag 5.0.1

A DAG-based combinatorial-model framework.
Documentation
import random
from dataclasses import dataclass
from typing import List

@dataclass
class ProductPeriod:
    id: str
    _from: int
    _to: int


@dataclass
class Item:
    period: ProductPeriod
    value: float

    @staticmethod
    def random() -> 'Item':
        # Generate random product ID
        product_id = str(random.randint(100000, 999999))

        # Generate random period using integers (e.g., days from epoch)
        # All periods within a 120-day window
        start = random.randint(0, 10)

        # End is 20-60 days after start (shorter periods for more overlaps)
        duration = random.randint(0, 5)
        end = start + duration

        # Random value between 0 and 1000
        value = random.uniform(0, 10)

        return Item(
            period=ProductPeriod(
                id=product_id,
                _from=start,
                _to=end
            ),
            value=value
        )

    @staticmethod
    def random_non_overlapping(count: int) -> List['Item']:
        """Generate a list of non-overlapping items with random gaps (some back-to-back)."""
        items = []
        current_day = 0

        for _ in range(count):
            product_id = str(random.randint(100000, 999999))

            # Sometimes add a gap (0-3 days), sometimes back-to-back
            # 50% chance of being back-to-back (gap=0), 50% chance of gap 1-3
            if random.random() < 0.5:
                gap = 0
            else:
                gap = random.randint(1, 3)

            start = current_day + gap

            # Duration between 2-5 days
            duration = random.randint(2, 5)
            end = start + duration

            value = random.uniform(0, 10)

            items.append(Item(
                period=ProductPeriod(
                    id=product_id,
                    _from=start,
                    _to=end
                ),
                value=value
            ))

            # Move to next period (end of current period)
            current_day = end

        return items

    @staticmethod
    def random_overlapping(count: int) -> List['Item']:
        """Generate a list of items with intentional overlaps."""
        items = []
        current_day = 0

        for _ in range(count):
            product_id = str(random.randint(100000, 999999))

            # Start can be before the current_day ends to create overlap
            # or after with a small gap
            start = current_day + random.randint(-3, 2)
            start = max(0, start)  # Ensure non-negative

            # Duration between 2-5 days
            duration = random.randint(2, 5)
            end = start + duration

            value = random.uniform(0, 10)

            items.append(Item(
                period=ProductPeriod(
                    id=product_id,
                    _from=start,
                    _to=end
                ),
                value=value
            ))

            # Move forward by a small amount (not necessarily to end)
            current_day = start + random.randint(1, 3)

        return items

    @staticmethod
    def same_price_back_to_back(same_price_count: int, same_price_value: float = 5.0) -> List['Item']:
        """Generate multiple back-to-back periods with the same price, then a latest period with different price."""
        items = []
        current_day = 0

        # Create multiple back-to-back periods with the same price
        for _ in range(same_price_count):
            product_id = str(random.randint(100000, 999999))

            duration = random.randint(2, 5)
            start = current_day
            end = start + duration

            items.append(Item(
                period=ProductPeriod(
                    id=product_id,
                    _from=start,
                    _to=end
                ),
                value=same_price_value
            ))

            current_day = end

        # Add the latest period with a different price
        product_id = str(random.randint(100000, 999999))
        duration = random.randint(2, 5)
        start = current_day
        end = start + duration

        # Different price for the latest period
        different_price = random.uniform(6.0, 9.0)

        items.append(Item(
            period=ProductPeriod(
                id=product_id,
                _from=start,
                _to=end
            ),
            value=different_price
        ))

        return items

    @staticmethod
    def grouped_back_to_back(groups: List[tuple[int, float]]) -> List['Item']:
        """Generate groups of back-to-back periods with specified counts and prices.

        Args:
            groups: List of tuples (count, price) where count is the number of
                   back-to-back periods and price is the value for that group.

        Example:
            Item.grouped_back_to_back([(3, 5.0), (2, 7.5), (1, 3.2)])
            Creates 3 periods at 5.0, then 2 periods at 7.5, then 1 period at 3.2
        """
        items = []
        current_day = 0

        for count, price in groups:
            for _ in range(count):
                product_id = str(random.randint(100000, 999999))

                duration = random.randint(2, 5)
                start = current_day
                end = start + duration

                items.append(Item(
                    period=ProductPeriod(
                        id=product_id,
                        _from=start,
                        _to=end
                    ),
                    value=price
                ))

                current_day = end

        return items

    @staticmethod
    def grouped_with_gaps(groups: List[tuple[int, float]]) -> List['Item']:
        """Generate groups of non-overlapping periods (with gaps) sharing the same price.

        Args:
            groups: List of tuples (count, price) where count is the number of
                   periods with that price.

        Example:
            Item.grouped_with_gaps([(3, 5.0), (2, 7.5), (1, 3.2)])
            Creates 3 periods at 5.0, then 2 periods at 7.5, then 1 period at 3.2
            Some may be back-to-back, some may have gaps
        """
        items = []
        current_day = 0

        for count, price in groups:
            for _ in range(count):
                product_id = str(random.randint(100000, 999999))

                # Sometimes add a gap (0-3 days), sometimes back-to-back
                # 50% chance of being back-to-back (gap=0), 50% chance of gap 1-3
                if random.random() < 0.5:
                    gap = 0
                else:
                    gap = random.randint(1, 3)

                start = current_day + gap
                duration = random.randint(2, 5)
                end = start + duration

                items.append(Item(
                    period=ProductPeriod(
                        id=product_id,
                        _from=start,
                        _to=end
                    ),
                    value=price
                ))

                current_day = end

        return items

def pil(collection: List[Item]) -> tuple[Item, List[Item], ProductPeriod]:

    # FROM THE FOLLOWING LOGIC:
    # var current = collections.First();

    # // Hämta det första element som är skilt ifrån första sales total
    # // Det är ifrån det element vi ska kolla 30 dagar bakåt
    # var latestTotaltBeforeCurrent = collections
    #     .Skip(1)
    #     .FirstOrDefault(c => c.Pricing.SalesTotal.Amount != current.Pricing.SalesTotal.Amount);

    # // Vi skapar en tillfällig period som sträcker sig ifrån senaste förändringsprisets till-datum
    # // och 30 dagar bakåt
    # var pilPeriod = new ProductPeriod(
    #     new ProductPeriodId("pil"),
    #     latestTotaltBeforeCurrent.ProductPeriod.To.AddDays(-30),
    #     latestTotaltBeforeCurrent.ProductPeriod.To);

    # // Pilpriset är nu det lägsta pris som konfigurationen haft ifrån
    # // latestTotaltBeforeCurrent och 30 dagar bakåt.
    # var pilPrice = collections
    #     .Where(c => c.ProductPeriod.Overlaps(pilPeriod))
    #     .MinBy(c => c.Pricing.SalesTotal.Amount);

    # return pilPrice.Pricing;

    # Sort collection by period start date (descending - most recent first)
    sorted_collection = sorted(collection, key=lambda x: x.period._from, reverse=True)

    current = sorted_collection[0]

    # Find the index of the first item with a different value
    latest_total_before_current_idx = next(
        (i for i, c in enumerate(sorted_collection[1:], start=1) if c.value != current.value),
        None
    )

    if latest_total_before_current_idx is None:
        # No PIL period to calculate
        empty_period = ProductPeriod(id="pil", _from=0, _to=0)
        return current, sorted_collection, empty_period

    latest_total_before_current = sorted_collection[latest_total_before_current_idx-1]

    # Create PIL period: 30 days before the end date of latest_total_before_current
    pil_period = ProductPeriod(
        id="pil",
        _from=latest_total_before_current.period._from - 10,
        _to=latest_total_before_current.period._from
    )

    # Find items that overlap with the PIL period
    overlapping_items = [
        c for c in sorted_collection
        if periods_overlap(c.period, pil_period)
    ]

    # Return the item with the minimum value
    if overlapping_items:
        return min(overlapping_items, key=lambda c: c.value), sorted_collection, pil_period

    return current, sorted_collection, pil_period


def periods_overlap(p1: ProductPeriod, p2: ProductPeriod) -> bool:
    """Check if two periods overlap."""
    # Special case: if either period has same from and to, no overlap
    if p1._from == p1._to or p2._from == p2._to:
        return False

    # Determine which period starts earlier
    earliest_period, latest_period = (p1, p2) if p1._from < p2._from else (p2, p1)
    return earliest_period._to >= latest_period._from


def visualize_period(from_day: int, to_day: int, max_days: int = 200) -> str:
    """Create a visual representation of a period where each character = 1 day."""
    # Create the timeline
    timeline = [' '] * max_days

    # Fill the period with dashes (from inclusive, to exclusive)
    for i in range(from_day, min(to_day, max_days)):
        timeline[i] = '-'

    # Mark start boundary
    if from_day < max_days:
        timeline[from_day] = '|'

    # Mark end boundary (at to position, which is exclusive from the range)
    # This way if a.to = b.from, they share the same | at that position
    if to_day < max_days and to_day > from_day:
        timeline[to_day] = '|'

    return ''.join(timeline[:max_days])


def print_item(item: Item, label: str = "", highlight: bool = False, show_visual: bool = True):
    """Print an item in a readable format."""
    marker = ">>> " if highlight else "    "
    duration = item.period._to - item.period._from
    print(f"{marker}{label}")
    print(f"{marker}  Period ID: {item.period.id}")
    print(f"{marker}  From:      {item.period._from}")
    print(f"{marker}  To:        {item.period._to}")
    print(f"{marker}  Duration:  {duration} days")
    print(f"{marker}  Value:     {item.value:.2f}")
    if show_visual:
        visual = visualize_period(item.period._from, item.period._to)
        print(f"{marker}  Timeline:  {visual}")
    print()


def print_collection_visual(pil_result: Item, collection: List[Item], pil_period: ProductPeriod):
    """Print all items with aligned visual timelines."""
    # Find the maximum day needed
    max_day = max(item.period._to for item in collection)

    # Print PIL period in red at the top if it exists
    if pil_period and pil_period._from != pil_period._to:
        visual = visualize_period(pil_period._from, pil_period._to, max_day + 1)
        print(f"\033[91m{visual}  PIL PERIOD \033[0m")
        print()

    for i, item in enumerate(collection):
        visual = visualize_period(item.period._from, item.period._to, max_day + 1)
        marker = " <-- PIL" if pil_result and item == pil_result else ""
        print(f"{visual}  {item.value:.2f} {marker}")


if __name__ == "__main__":
    for _ in range(15):
        print("-" * 50)
        random_collection = Item.grouped_with_gaps([(3, 8.0), (2, 7.5), (5, 5)])
        print_collection_visual(*pil(random_collection))