text:
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

from typing import Tuple

import numpy as np
import pandas

from modin.error_message import ErrorMessage
from modin.data_management.utils import compute_chunksize, _get_nan_block_id

class BaseBlockPartitions(object):
    """Abstract Class that manages a set of `BaseRemotePartition` objects, and
        structures them into a 2D numpy array. This object will interact with
        each of these objects through the `BaseRemotePartition` API.

    Note: See the Abstract Methods and Fields section immediately below this
        for a list of requirements for subclassing this object.
    """

    # Abstract Methods and Fields: Must implement in children classes
    # In some cases, there you may be able to use the same implementation for
    # some of these abstract methods, but for the sake of generality they are
    # treated differently.
    def __init__(self, partitions):
        """Init must accept a parameter `partitions` that is a 2D numpy array
            of type `_partition_class` (defined below). This method will be
            called from a factory.

        Args:
            partitions: A 2D numpy array of the type defined in
                `_partition_class`.
        """
        raise NotImplementedError("Must be implemented in children classes")

    @property
    def __constructor__(self):
        """Convenience method for creating new objects.

        Note: This is used by the abstract class to ensure the return type is the same
            as the child subclassing it.
        """
        return type(self)

    # Partition class is the class to use for storing each partition. It must
    # extend the `BaseRemotePartition` class.
    _partition_class = None
    # Column partitions class is the class to use to create the column partitions.
    _column_partitions_class = None
    # Row partitions class is the class to use to create row partitions.
    _row_partition_class = None
    # Whether or not we have already filtered out the empty partitions.
    _filtered_empties = False

    def _get_partitions(self):
        if not self._filtered_empties:
            self._partitions_cache = np.array(
                [
                    row
                    for row in [
                        [
                            self._partitions_cache[i][j]
                            for j in range(len(self._partitions_cache[i]))
                            if self.block_lengths[i] != 0 and self.block_widths[j] != 0
                        ]
                        for i in range(len(self._partitions_cache))
                    ]
                    if len(row)
                ]
            )
            self._remove_empty_blocks()
            self._filtered_empties = True
        return self._partitions_cache

    def _set_partitions(self, new_partitions):
        self._filtered_empties = False
        self._partitions_cache = new_partitions

    partitions = property(_get_partitions, _set_partitions)

    def preprocess_func(self, map_func):
        """Preprocess a function to be applied to `BaseRemotePartition` objects.

        Note: If your `BaseRemotePartition` objects assume that a function provided
            is serialized or wrapped or in some other format, this is the place
            to add that logic. It is possible that this can also just return
            `map_func` if the `apply` method of the `BaseRemotePartition` object
            you are using does not require any modification to a given
            function.

        Args:
            map_func: The function to be preprocessed.

        Returns
            The preprocessed version of the `map_func` provided. Note: This
            does not require any specific format, only that the
            `BaseRemotePartition.apply` method will recognize it (For the subclass
            being used).
        """
        return self._partition_class.preprocess_func(map_func)

    # END Abstract Methods

    @property
    def column_partitions(self):
        """A list of `BaseAxisPartition` objects.

        Note: Each value in this list will be an `BaseAxisPartition` object.
            `BaseAxisPartition` is located in the `base_remote_partition.py` file.

        Returns a list of `BaseAxisPartition` objects.
        """
        return [self._column_partitions_class(col) for col in self.partitions.T]

    @property
    def row_partitions(self):
        """A list of `BaseAxisPartition` objects, represents column partitions.

        Note: Each value in this list will an `BaseAxisPartition` object.
            `BaseAxisPartition` is located in the `base_remote_partition.py` file.

        Returns a list of `BaseAxisPartition` objects.
        """
        return [self._row_partition_class(row) for row in self.partitions]

    # Lengths of the blocks
    _lengths_cache = None

    # These are set up as properties so that we only use them when we need
    # them. We also do not want to trigger this computation on object creation.
    @property
    def block_lengths(self):
        """Gets the lengths of the blocks.

        Note: This works with the property structure `_lengths_cache` to avoid
            having to recompute these values each time they are needed.
        """
        if self._lengths_cache is None:
            # The first column will have the correct lengths. We have an
            # invariant that requires that all blocks be the same length in a
            # row of blocks.
            self._lengths_cache = np.array(
                [obj.length().get() for obj in self._partitions_cache.T[0]]
                if len(self._partitions_cache.T) > 0
                else []
            )
        return self._lengths_cache

    # Widths of the blocks
    _widths_cache = None

    @property
    def block_widths(self):
        """Gets the widths of the blocks.

        Note: This works with the property structure `_widths_cache` to avoid
            having to recompute these values each time they are needed.
        """
        if self._widths_cache is None:
            # The first column will have the correct lengths. We have an
            # invariant that requires that all blocks be the same width in a
            # column of blocks.
            self._widths_cache = np.array(
                [obj.width().get() for obj in self._partitions_cache[0]]
                if len(self._partitions_cache) > 0
                else []
            )
        return self._widths_cache

    def _remove_empty_blocks(self):
        if self._widths_cache is not None:
            self._widths_cache = [width for width in self._widths_cache if width != 0]
        if self._lengths_cache is not None:
            self._lengths_cache = np.array(
                [length for length in self._lengths_cache if length != 0]
            )

    @property
    def shape(self) -> Tuple[int, int]:
        return int(np.sum(self.block_lengths)), int(np.sum(self.block_widths))

    def full_reduce(self, map_func, reduce_func, axis):
        """Perform a full reduce on the data.

        Note: This follows the 2-phase reduce paradigm, where each partition
            performs a local reduction (map_func), then partitions are brought
            together and the final reduction occurs.
        Args:
            map_func: The function that will be performed on all partitions.
                This is the local reduction on each partition.
            reduce_func: The final reduction function. This can differ from the
                `map_func`
            axis: The axis to perform this operation along
                (0 - index, 1 - columns)
        Returns:
            A Pandas Series
        """
        raise NotImplementedError("Blocked on Distributed Series")

    def map_across_blocks(self, map_func):
        """Applies `map_func` to every partition.

        Args:
            map_func: The function to apply.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        preprocessed_map_func = self.preprocess_func(map_func)
        new_partitions = np.array(
            [
                [part.apply(preprocessed_map_func) for part in row_of_parts]
                for row_of_parts in self.partitions
            ]
        )
        return self.__constructor__(new_partitions)

    def lazy_map_across_blocks(self, map_func, kwargs):
        preprocessed_map_func = self.preprocess_func(map_func)
        new_partitions = np.array(
            [
                [
                    part.add_to_apply_calls(preprocessed_map_func, kwargs)
                    for part in row_of_parts
                ]
                for row_of_parts in self.partitions
            ]
        )
        return self.__constructor__(new_partitions)

    def copartition_datasets(self, axis, other, left_func, right_func):
        """Copartition two BlockPartitions objects.

        Args:
            axis: The axis to copartition.
            other: The other BlockPartitions object to copartition with.
            left_func: The function to apply to left. If None, just use the dimension
                of self (based on axis).
            right_func: The function to apply to right. If None, check the dimensions of
                other and use the identity function if splitting needs to happen.

        Returns:
            A tuple of BlockPartitions objects, left and right.
        """
        if left_func is None:
            new_self = self
        else:
            new_self = self.map_across_full_axis(axis, left_func)

        # This block of code will only shuffle if absolutely necessary. If we do need to
        # shuffle, we use the identity function and then reshuffle.
        if right_func is None:
            if axis == 0 and not np.array_equal(
                other.block_lengths, new_self.block_lengths
            ):
                new_other = other.manual_shuffle(
                    axis, lambda x: x, new_self.block_lengths
                )
            elif axis == 1 and not np.array_equal(
                other.block_widths, new_self.block_widths
            ):
                new_other = other.manual_shuffle(
                    axis, lambda x: x, new_self.block_widths
                )
            else:
                new_other = other
        # Most of the time, we will be given an operation to do. We perform that with
        # manual_shuffle.
        else:
            new_other = other.manual_shuffle(
                axis,
                right_func,
                new_self.block_lengths if axis == 0 else new_self.block_widths,
            )
        return new_self, new_other

    def map_across_full_axis(self, axis, map_func):
        """Applies `map_func` to every partition.

        Note: This method should be used in the case that `map_func` relies on
            some global information about the axis.

        Args:
            axis: The axis to perform the map across (0 - index, 1 - columns).
            map_func: The function to apply.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        # Since we are already splitting the DataFrame back up after an
        # operation, we will just use this time to compute the number of
        # partitions as best we can right now.
        num_splits = self._compute_num_partitions()
        preprocessed_map_func = self.preprocess_func(map_func)
        partitions = self.column_partitions if not axis else self.row_partitions
        # For mapping across the entire axis, we don't maintain partitioning because we
        # may want to line to partitioning up with another BlockPartitions object. Since
        # we don't need to maintain the partitioning, this gives us the opportunity to
        # load-balance the data as well.
        result_blocks = np.array(
            [
                part.apply(preprocessed_map_func, num_splits=num_splits)
                for part in partitions
            ]
        )
        # If we are mapping over columns, they are returned to use the same as
        # rows, so we need to transpose the returned 2D numpy array to return
        # the structure to the correct order.
        return (
            self.__constructor__(result_blocks.T)
            if not axis
            else self.__constructor__(result_blocks)
        )

    def take(self, axis, n):
        """Take the first (or last) n rows or columns from the blocks

        Note: Axis = 0 will be equivalent to `head` or `tail`
              Axis = 1 will be equivalent to `front` or `back`

        Args:
            axis: The axis to extract (0 for extracting rows, 1 for extracting columns)
            n: The number of rows or columns to extract, negative denotes to extract
                from the bottom of the object

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        # These are the partitions that we will extract over
        if not axis:
            partitions = self.partitions
            bin_lengths = self.block_lengths
        else:
            partitions = self.partitions.T
            bin_lengths = self.block_widths
        if n < 0:
            length_bins = np.cumsum(bin_lengths[::-1])
            n *= -1
            idx = int(np.digitize(n, length_bins))
            if idx > 0:
                remaining = int(n - length_bins[idx - 1])
            else:
                remaining = n
            # In this case, we require no remote compute. This is much faster.
            if remaining == 0:
                result = partitions[-idx:]
            else:
                # Reverse for ease of iteration and then re-reverse at the end
                partitions = partitions[::-1]
                # We build this iloc to avoid creating a bunch of helper methods.
                # This code creates slice objects to be passed to `iloc` to grab
                # the last n rows or columns depending on axis.
                slice_obj = (
                    slice(-remaining, None)
                    if axis == 0
                    else (slice(None), slice(-remaining, None))
                )
                func = self.preprocess_func(lambda df: df.iloc[slice_obj])
                # We use idx + 1 here because the loop is not inclusive, and we
                # need to iterate through idx.
                result = np.array(
                    [
                        partitions[i]
                        if i != idx
                        else [obj.apply(func) for obj in partitions[i]]
                        for i in range(idx + 1)
                    ]
                )[::-1]
        else:
            length_bins = np.cumsum(bin_lengths)
            idx = int(np.digitize(n, length_bins))
            if idx > 0:
                remaining = int(n - length_bins[idx - 1])
            else:
                remaining = n
            # In this case, we require no remote compute. This is much faster.
            if remaining == 0:
                result = partitions[:idx]
            else:
                # We build this iloc to avoid creating a bunch of helper methods.
                # This code creates slice objects to be passed to `iloc` to grab
                # the first n rows or columns depending on axis.
                slice_obj = (
                    slice(remaining) if axis == 0 else (slice(None), slice(remaining))
                )
                func = self.preprocess_func(lambda df: df.iloc[slice_obj])
                # See note above about idx + 1
                result = np.array(
                    [
                        partitions[i]
                        if i != idx
                        else [obj.apply(func) for obj in partitions[i]]
                        for i in range(idx + 1)
                    ]
                )
        return self.__constructor__(result.T) if axis else self.__constructor__(result)

    def concat(self, axis, other_blocks):
        """Concatenate the blocks with another set of blocks.

        Note: Assumes that the blocks are already the same shape on the
            dimension being concatenated. A ValueError will be thrown if this
            condition is not met.

        Args:
            axis: The axis to concatenate to.
            other_blocks: the other blocks to be concatenated. This is a
                BaseBlockPartitions object.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        if type(other_blocks) is list:
            other_blocks = [blocks.partitions for blocks in other_blocks]
            return self.__constructor__(
                np.concatenate([self.partitions] + other_blocks, axis=axis)
            )
        else:
            return self.__constructor__(
                np.append(self.partitions, other_blocks.partitions, axis=axis)
            )

    def copy(self):
        """Create a copy of this object.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        return self.__constructor__(self.partitions.copy())

    def transpose(self, *args, **kwargs):
        """Transpose the blocks stored in this object.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        return self.__constructor__(self.partitions.T)

    def to_pandas(self, is_transposed=False):
        """Convert this object into a Pandas DataFrame from the partitions.

        Args:
            is_transposed: A flag for telling this object that the external
                representation is transposed, but not the internal.

        Returns:
            A Pandas DataFrame
        """
        # In the case this is transposed, it is easier to just temporarily
        # transpose back then transpose after the conversion. The performance
        # is the same as if we individually transposed the blocks and
        # concatenated them, but the code is much smaller.
        if is_transposed:
            return self.transpose().to_pandas(False).T
        else:
            retrieved_objects = [
                [obj.to_pandas() for obj in part] for part in self.partitions
            ]
            if all(
                isinstance(part, pandas.Series)
                for row in retrieved_objects
                for part in row
            ):
                axis = 0
            elif all(
                isinstance(part, pandas.DataFrame)
                for row in retrieved_objects
                for part in row
            ):
                axis = 1
            else:
                ErrorMessage.catch_bugs_and_request_email(True)
            df_rows = [
                pandas.concat([part for part in row], axis=axis)
                for row in retrieved_objects
                if not all(part.empty for part in row)
            ]
            if len(df_rows) == 0:
                return pandas.DataFrame()
            else:
                return pandas.concat(df_rows)

    @classmethod
    def from_pandas(cls, df):
        num_splits = cls._compute_num_partitions()
        put_func = cls._partition_class.put
        row_chunksize, col_chunksize = compute_chunksize(df, num_splits)

        # Each chunk must have a RangeIndex that spans its length and width
        # according to our invariant.
        def chunk_builder(i, j):
            chunk = df.iloc[i : i + row_chunksize, j : j + col_chunksize].copy()
            chunk.index = pandas.RangeIndex(len(chunk.index))
            chunk.columns = pandas.RangeIndex(len(chunk.columns))
            return put_func(chunk)

        parts = [
            [chunk_builder(i, j) for j in range(0, len(df.columns), col_chunksize)]
            for i in range(0, len(df), row_chunksize)
        ]
        return cls(np.array(parts))

    def get_indices(self, axis=0, index_func=None, old_blocks=None):
        """This gets the internal indices stored in the partitions.

        Note: These are the global indices of the object. This is mostly useful
            when you have deleted rows/columns internally, but do not know
            which ones were deleted.

        Args:
            axis: This axis to extract the labels. (0 - index, 1 - columns).
            index_func: The function to be used to extract the function.
            old_blocks: An optional previous object that this object was
                created from. This is used to compute the correct offsets.

        Returns:
            A Pandas Index object.
        """
        ErrorMessage.catch_bugs_and_request_email(not callable(index_func))
        func = self.preprocess_func(index_func)
        if axis == 0:
            # We grab the first column of blocks and extract the indices
            # Note: We use _partitions_cache in the context of this function to make
            # sure that none of the partitions are modified or filtered out before we
            # get the index information.
            # DO NOT CHANGE TO self.partitions under any circumstance.
            new_indices = (
                [idx.apply(func).get() for idx in self._partitions_cache.T[0]]
                if len(self._partitions_cache.T)
                else []
            )
            # This is important because sometimes we have resized the data. The new
            # sizes will not be valid if we are trying to compute the index on a
            # new object that has a different length.
            if old_blocks is not None:
                cumulative_block_lengths = np.array(old_blocks.block_lengths).cumsum()
            else:
                cumulative_block_lengths = np.array(self.block_lengths).cumsum()
        else:
            new_indices = (
                [idx.apply(func).get() for idx in self._partitions_cache[0]]
                if len(self._partitions_cache)
                else []
            )

            if old_blocks is not None:
                cumulative_block_lengths = np.array(old_blocks.block_widths).cumsum()
            else:
                cumulative_block_lengths = np.array(self.block_widths).cumsum()
        full_indices = new_indices[0] if len(new_indices) else new_indices
        if old_blocks is not None:
            for i in range(len(new_indices)):
                # If the length is 0 there is nothing to append.
                if i == 0 or len(new_indices[i]) == 0:
                    continue
                # The try-except here is intended to catch issues where we are
                # trying to get a string index out of the internal index.
                try:
                    append_val = new_indices[i] + cumulative_block_lengths[i - 1]
                except TypeError:
                    append_val = new_indices[i]

                full_indices = full_indices.append(append_val)
        else:
            full_indices = full_indices.append(new_indices[1:])
        return full_indices

    @classmethod
    def _compute_num_partitions(cls):
        """Currently, this method returns the default. In the future it will
            estimate the optimal number of partitions.

        :return:
        """
        from ...pandas import DEFAULT_NPARTITIONS

        return DEFAULT_NPARTITIONS

    # Extracting rows/columns
    def _get_blocks_containing_index(self, axis, index):
        """Convert a global index to a block index and local index.

        Note: This method is primarily used to convert a global index into a
            partition index (along the axis provided) and local index (useful
            for `iloc` or similar operations.

        Args:
            axis: The axis along which to get the indices
                (0 - columns, 1 - rows)
            index: The global index to convert.

        Returns:
            A tuple containing (block index and internal index).
        """
        if not axis:
            ErrorMessage.catch_bugs_and_request_email(index > sum(self.block_widths))
            cumulative_column_widths = np.array(self.block_widths).cumsum()
            block_idx = int(np.digitize(index, cumulative_column_widths))
            if block_idx == len(cumulative_column_widths):
                block_idx -= 1
            # Compute the internal index based on the previous lengths. This
            # is a global index, so we must subtract the lengths first.
            internal_idx = (
                index
                if not block_idx
                else index - cumulative_column_widths[block_idx - 1]
            )
        else:
            ErrorMessage.catch_bugs_and_request_email(index > sum(self.block_lengths))
            cumulative_row_lengths = np.array(self.block_lengths).cumsum()
            block_idx = int(np.digitize(index, cumulative_row_lengths))
            # See note above about internal index
            internal_idx = (
                index
                if not block_idx
                else index - cumulative_row_lengths[block_idx - 1]
            )
        return block_idx, internal_idx

    def _get_dict_of_block_index(self, axis, indices, ordered=False):
        """Convert indices to a dict of block index to internal index mapping.

        Note: See `_get_blocks_containing_index` for primary usage. This method
            accepts a list of indices rather than just a single value, and uses
            `_get_blocks_containing_index`.

        Args:
            axis: The axis along which to get the indices
                (0 - columns, 1 - rows)
            indices: A list of global indices to convert.

        Returns
            For unordered: a dictionary of {block index: list of local indices}.
            For ordered: a list of tuples mapping block index: list of local indices.
        """
        # Get the internal index and create a dictionary so we only have to
        # travel to each partition once.
        all_partitions_and_idx = [
            self._get_blocks_containing_index(axis, i) for i in indices
        ]

        # In ordered, we have to maintain the order of the list of indices provided.
        # This means that we need to return a list instead of a dictionary.
        if ordered:
            # In ordered, the partitions dict is a list of tuples
            partitions_dict = []
            # This variable is used to store the most recent partition that we added to
            # the partitions_dict. This allows us to only visit a partition once when we
            # have multiple values that will be operated on in that partition.
            last_part = -1
            for part_idx, internal_idx in all_partitions_and_idx:
                if part_idx == last_part:
                    # We append to the list, which is the value part of the tuple.
                    partitions_dict[-1][-1].append(internal_idx)
                else:
                    # This is where we add new values.
                    partitions_dict.append((part_idx, [internal_idx]))
                last_part = part_idx
        else:
            # For unordered, we can just return a dictionary mapping partition to the
            # list of indices being operated on.
            partitions_dict = {}
            for part_idx, internal_idx in all_partitions_and_idx:
                if part_idx not in partitions_dict:
                    partitions_dict[part_idx] = [internal_idx]
                else:
                    partitions_dict[part_idx].append(internal_idx)
        return partitions_dict

    def _apply_func_to_list_of_partitions(self, func, partitions, **kwargs):
        """Applies a function to a list of remote partitions.

        Note: The main use for this is to preprocess the func.

        Args:
            func: The func to apply
            partitions: The list of partitions

        Returns:
            A list of BaseRemotePartition objects.
        """
        preprocessed_func = self.preprocess_func(func)
        return [obj.apply(preprocessed_func, **kwargs) for obj in partitions]

    def apply_func_to_select_indices(self, axis, func, indices, keep_remaining=False):
        """Applies a function to select indices.

        Note: Your internal function must take a kwarg `internal_indices` for
            this to work correctly. This prevents information leakage of the
            internal index to the external representation.

        Args:
            axis: The axis to apply the func over.
            func: The function to apply to these indices.
            indices: The indices to apply the function to.
            keep_remaining: Whether or not to keep the other partitions.
                Some operations may want to drop the remaining partitions and
                keep only the results.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        if self.partitions.size == 0:
            return np.array([[]])
        # Handling dictionaries has to be done differently, but we still want
        # to figure out the partitions that need to be applied to, so we will
        # store the dictionary in a separate variable and assign `indices` to
        # the keys to handle it the same as we normally would.
        if isinstance(indices, dict):
            dict_indices = indices
            indices = list(indices.keys())
        else:
            dict_indices = None
        if not isinstance(indices, list):
            indices = [indices]
        partitions_dict = self._get_dict_of_block_index(
            axis, indices, ordered=not keep_remaining
        )
        if not axis:
            partitions_for_apply = self.partitions.T
        else:
            partitions_for_apply = self.partitions
        # We may have a command to perform different functions on different
        # columns at the same time. We attempt to handle this as efficiently as
        # possible here. Functions that use this in the dictionary format must
        # accept a keyword argument `func_dict`.
        if dict_indices is not None:

            def local_to_global_idx(partition_id, local_idx):
                if partition_id == 0:
                    return local_idx
                if axis == 0:
                    cumulative_axis = np.cumsum(self.block_widths)
                else:
                    cumulative_axis = np.cumsum(self.block_lengths)
                return cumulative_axis[partition_id - 1] + local_idx

            if not keep_remaining:
                result = np.array(
                    [
                        self._apply_func_to_list_of_partitions(
                            func,
                            partitions_for_apply[o_idx],
                            func_dict={
                                i_idx: dict_indices[local_to_global_idx(o_idx, i_idx)]
                                for i_idx in list_to_apply
                                if i_idx >= 0
                            },
                        )
                        for o_idx, list_to_apply in partitions_dict
                    ]
                )
            else:
                result = np.array(
                    [
                        partitions_for_apply[i]
                        if i not in partitions_dict
                        else self._apply_func_to_list_of_partitions(
                            func,
                            partitions_for_apply[i],
                            func_dict={
                                idx: dict_indices[local_to_global_idx(i, idx)]
                                for idx in partitions_dict[i]
                                if idx >= 0
                            },
                        )
                        for i in range(len(partitions_for_apply))
                    ]
                )
        else:
            if not keep_remaining:
                # We are passing internal indices in here. In order for func to
                # actually be able to use this information, it must be able to take in
                # the internal indices. This might mean an iloc in the case of Pandas
                # or some other way to index into the internal representation.
                result = np.array(
                    [
                        self._apply_func_to_list_of_partitions(
                            func,
                            partitions_for_apply[idx],
                            internal_indices=list_to_apply,
                        )
                        for idx, list_to_apply in partitions_dict
                    ]
                )
            else:
                # The difference here is that we modify a subset and return the
                # remaining (non-updated) blocks in their original position.
                result = np.array(
                    [
                        partitions_for_apply[i]
                        if i not in partitions_dict
                        else self._apply_func_to_list_of_partitions(
                            func,
                            partitions_for_apply[i],
                            internal_indices=partitions_dict[i],
                        )
                        for i in range(len(partitions_for_apply))
                    ]
                )
        return (
            self.__constructor__(result.T) if not axis else self.__constructor__(result)
        )

    def apply_func_to_select_indices_along_full_axis(
        self, axis, func, indices, keep_remaining=False
    ):
        """Applies a function to a select subset of full columns/rows.

        Note: This should be used when you need to apply a function that relies
            on some global information for the entire column/row, but only need
            to apply a function to a subset.

        Important: For your func to operate directly on the indices provided,
            it must use `internal_indices` as a keyword argument.

        Args:
            axis: The axis to apply the function over (0 - rows, 1 - columns)
            func: The function to apply.
            indices: The global indices to apply the func to.
            keep_remaining: Whether or not to keep the other partitions.
                Some operations may want to drop the remaining partitions and
                keep only the results.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        if self.partitions.size == 0:
            return self.__constructor__(np.array([[]]))
        if isinstance(indices, dict):
            dict_indices = indices
            indices = list(indices.keys())
        else:
            dict_indices = None
        if not isinstance(indices, list):
            indices = [indices]
        partitions_dict = self._get_dict_of_block_index(axis, indices)
        preprocessed_func = self.preprocess_func(func)
        # Since we might be keeping the remaining blocks that are not modified,
        # we have to also keep the block_partitions object in the correct
        # direction (transpose for columns).
        if not axis:
            partitions_for_apply = self.column_partitions
            partitions_for_remaining = self.partitions.T
        else:
            partitions_for_apply = self.row_partitions
            partitions_for_remaining = self.partitions
        # We may have a command to perform different functions on different
        # columns at the same time. We attempt to handle this as efficiently as
        # possible here. Functions that use this in the dictionary format must
        # accept a keyword argument `func_dict`.
        if dict_indices is not None:
            if not keep_remaining:
                result = np.array(
                    [
                        partitions_for_apply[i].apply(
                            preprocessed_func,
                            func_dict={
                                idx: dict_indices[idx] for idx in partitions_dict[i]
                            },
                        )
                        for i in partitions_dict
                    ]
                )
            else:
                result = np.array(
                    [
                        partitions_for_remaining[i]
                        if i not in partitions_dict
                        else self._apply_func_to_list_of_partitions(
                            preprocessed_func,
                            partitions_for_apply[i],
                            func_dict={
                                idx: dict_indices[idx] for idx in partitions_dict[i]
                            },
                        )
                        for i in range(len(partitions_for_apply))
                    ]
                )
        else:
            if not keep_remaining:
                # See notes in `apply_func_to_select_indices`
                result = np.array(
                    [
                        partitions_for_apply[i].apply(
                            preprocessed_func, internal_indices=partitions_dict[i]
                        )
                        for i in partitions_dict
                    ]
                )
            else:
                # See notes in `apply_func_to_select_indices`
                result = np.array(
                    [
                        partitions_for_remaining[i]
                        if i not in partitions_dict
                        else partitions_for_apply[i].apply(
                            preprocessed_func, internal_indices=partitions_dict[i]
                        )
                        for i in range(len(partitions_for_remaining))
                    ]
                )
        return (
            self.__constructor__(result.T) if not axis else self.__constructor__(result)
        )

    def apply_func_to_indices_both_axis(
        self,
        func,
        row_indices,
        col_indices,
        lazy=False,
        keep_remaining=True,
        mutate=False,
        item_to_distribute=None,
    ):
        """
        Apply a function to along both axis

        Important: For your func to operate directly on the indices provided,
            it must use `row_internal_indices, col_internal_indices` as keyword
            arguments.
        """
        if keep_remaining:
            row_partitions_list = self._get_dict_of_block_index(1, row_indices).items()
            col_partitions_list = self._get_dict_of_block_index(0, col_indices).items()
        else:
            row_partitions_list = self._get_dict_of_block_index(
                1, row_indices, ordered=True
            )
            col_partitions_list = self._get_dict_of_block_index(
                0, col_indices, ordered=True
            )
            result = np.empty(
                (len(row_partitions_list), len(col_partitions_list)), dtype=type(self)
            )

        if not mutate:
            partition_copy = self.partitions.copy()
        else:
            partition_copy = self.partitions

        row_position_counter = 0
        for row_idx, row_values in enumerate(row_partitions_list):
            row_blk_idx, row_internal_idx = row_values
            col_position_counter = 0
            for col_idx, col_values in enumerate(col_partitions_list):
                col_blk_idx, col_internal_idx = col_values
                remote_part = partition_copy[row_blk_idx, col_blk_idx]

                if item_to_distribute is not None:
                    item = item_to_distribute[
                        row_position_counter : row_position_counter
                        + len(row_internal_idx),
                        col_position_counter : col_position_counter
                        + len(col_internal_idx),
                    ]
                    item = {"item": item}
                else:
                    item = {}

                if lazy:
                    block_result = remote_part.add_to_apply_calls(
                        func,
                        row_internal_indices=row_internal_idx,
                        col_internal_indices=col_internal_idx,
                        **item
                    )
                else:
                    block_result = remote_part.apply(
                        func,
                        row_internal_indices=row_internal_idx,
                        col_internal_indices=col_internal_idx,
                        **item
                    )
                if keep_remaining:
                    partition_copy[row_blk_idx, col_blk_idx] = block_result
                else:
                    result[row_idx][col_idx] = block_result
                col_position_counter += len(col_internal_idx)

            row_position_counter += len(row_internal_idx)

        if keep_remaining:
            return self.__constructor__(partition_copy)
        else:
            return self.__constructor__(result)

    def inter_data_operation(self, axis, func, other):
        """Apply a function that requires two BaseBlockPartitions objects.

        Args:
            axis: The axis to apply the function over (0 - rows, 1 - columns)
            func: The function to apply
            other: The other BaseBlockPartitions object to apply func to.

        Returns:
            A new BaseBlockPartitions object, the type of object that called this.
        """
        if axis:
            partitions = self.row_partitions
            other_partitions = other.row_partitions
        else:
            partitions = self.column_partitions
            other_partitions = other.column_partitions
        func = self.preprocess_func(func)
        result = np.array(
            [
                partitions[i].apply(
                    func,
                    num_splits=self._compute_num_partitions(),
                    other_axis_partition=other_partitions[i],
                )
                for i in range(len(partitions))
            ]
        )
        return self.__constructor__(result) if axis else self.__constructor__(result.T)

    def manual_shuffle(self, axis, shuffle_func, lengths):
        """Shuffle the partitions based on the `shuffle_func`.

        Args:
            axis: The axis to shuffle across.
            shuffle_func: The function to apply before splitting the result.
            lengths: The length of each partition to split the result into.

        Returns:
             A new BaseBlockPartitions object, the type of object that called this.
        """
        if axis:
            partitions = self.row_partitions
        else:
            partitions = self.column_partitions
        func = self.preprocess_func(shuffle_func)
        result = np.array([part.shuffle(func, lengths) for part in partitions])
        return self.__constructor__(result) if axis else self.__constructor__(result.T)

    def __getitem__(self, key):
        return self.__constructor__(self.partitions[key])

    def __len__(self):
        return sum(self.block_lengths)

    def enlarge_partitions(self, n_rows=None, n_cols=None):
        data = self.partitions
        if n_rows:
            n_cols_lst = self.block_widths
            nan_oids_lst = [
                self._partition_class(
                    _get_nan_block_id(self._partition_class, n_rows, n_cols_)
                )
                for n_cols_ in n_cols_lst
            ]
            new_chunk = self.__constructor__(np.array([nan_oids_lst]))
            data = self.concat(axis=0, other_blocks=new_chunk)

        if n_cols:
            n_rows_lst = self.block_lengths
            nan_oids_lst = [
                self._partition_class(
                    _get_nan_block_id(self._partition_class, n_rows_, n_cols)
                )
                for n_rows_ in n_rows_lst
            ]
            new_chunk = self.__constructor__(np.array([nan_oids_lst]).T)
            data = self.concat(axis=1, other_blocks=new_chunk)
        return data

max_stars_repo_name: calzoom/modin

max_issues_repo_name: calzoom/modin

max_forks_repo_name: calzoom/modin


text:
import math
import torch
import torch.nn as nn
from torch.nn.parameter import Parameter as P
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
from hydroDL.model import waterNet
import importlib

importlib.reload(waterNet)
P = torch.tensor([1, 2, 3])[:, None]
E = torch.tensor([1, 2, 3])[:, None]
Q = torch.tensor([1, 2, 3])[:, None]/10
T = F.relu(torch.rand(3))[:, None]

x = F.relu(torch.rand(3, 5))
h = F.relu(torch.rand(3, 5))

we = torch.FloatTensor(5).uniform_(-1, 1)
wl = torch.FloatTensor(5).uniform_(-1, 1)
wk = torch.FloatTensor(5).uniform_(-1, 1)
ws = torch.FloatTensor(5).uniform_(-1, 1)

F.softmax(torch.FloatTensor(3,5),dim=0)

hn = h+x
h1 = torch.relu(hn-torch.exp(wl*2))
q1 = torch.relu(h1-E*torch.sigmoid(we))
h2 = hn-h1
q2 = h2 * torch.sigmoid(wk)
h = h2-q2
q2a = q2*torch.sigmoid(ws)
q2b = q2*(1-torch.sigmoid(ws))

ext: py

max_stars_repo_name: fkwai/geolearn

max_issues_repo_name: fkwai/geolearn

max_forks_repo_name: fkwai/geolearn

reason: import numpy


text:
import collections
import csv
import dataclasses
import datetime
import io
import itertools
import json
import logging
import os
import random
import typing
import uuid

import faker
import numpy as np
from google.cloud import storage

fake = faker.Faker()
# final datasets
orders = list()
users = list()
order_items = list()
events = list()
inventory_items = list()

# read from local csv and return products
def generate_products() -> typing.List[dict]:
    product_brand_dict = {}  # products partitioned by brand - unused
    product_category_dict = {}  # product partitioned by cateogry - unused
    gender_category_dict = {}  # products partitioned by gender and category - unused
    product_id_dict = {}  # products to generate events table - unused
    product_gender_dict = {}  # product partitioned by gender
    product_by_id_dict = {}  # products partitioned by product ID

    products = collections.defaultdict(list)
    with open(
        f"{os.environ['SOURCE_DIR']}/products.csv", encoding="utf-8"
    ) as productcsv:
        csv_reader = csv.DictReader(productcsv)
        for rows in csv_reader:
            for k, v in rows.items():
                products[k].append(v)

    product_id = products["id"]
    brands = products["brand"]
    name = products["name"]
    cost = products["cost"]
    category = products["category"]
    department = products["department"]
    sku = products["sku"]
    retail_price = products["retail_price"]
    distribution_center_id = products["distribution_center_id"]
    for _ in range(len(brands)):
        product_brand_dict[brands[_]] = []
        product_category_dict[category[_]] = []
        product_id_dict[product_id[_]] = []
        product_by_id_dict[product_id[_]] = []
        if department[_] == "Men":
            product_gender_dict["M"] = []
            gender_category_dict["M" + category[_]] = []
        if department[_] == "Women":
            product_gender_dict["F"] = []
            gender_category_dict["F" + category[_]] = []
    for col in list(
        zip(
            product_id,
            brands,
            name,
            cost,
            category,
            department,
            sku,
            retail_price,
            distribution_center_id,
        )
    ):
        product_id = col[0]
        brand = col[1]
        name = col[2]
        cost = col[3]
        category = col[4]
        department = col[5]
        sku = col[6]
        retail_price = col[7]
        distribution_center_id = col[8]

        product_by_id_dict[product_id] = {
            "brand": brand,
            "name": name,
            "cost": cost,
            "category": category,
            "department": department,
            "sku": sku,
            "retail_price": retail_price,
            "distribution_center_id": distribution_center_id,
        }
        product_brand_dict[brand].append(col)
        product_category_dict[category].append(col)
        if department == "Men":
            product_gender_dict["M"].append(col)
            gender_category_dict["M" + category].append(col)
        if department == "Women":
            product_gender_dict["F"].append(col)
            gender_category_dict["F" + category].append(col)

    # helper dict to generate events
    for col in list(zip(product_id, brands, category, department)):
        product_id_dict[col[0]] = {
            "brand": col[1],
            "category": col[2],
            "department": col[3],
        }
    return product_gender_dict, product_by_id_dict, products

# read from local csv and return locations
def generate_locations() -> typing.List[str]:
    location_data = []
    with open(
        f"{os.environ['SOURCE_DIR']}/world_pop.csv", encoding="utf-8"
    ) as worldcsv:
        csvReader = csv.DictReader(worldcsv)
        for rows in csvReader:
            location_data.append(rows)
    return location_data

SECONDS_IN_MINUTE = 60
MINUTES_IN_HOUR = 60
MINUTES_IN_DAY = 1440
MIN_AGE = 12
MAX_AGE = 71

products = generate_products()
logging.info("generating products helper dict")
logging.info("generating locations data")
LOCATION_DATA = generate_locations()
PRODUCT_GENDER_DICT = products[0]
PRODUCT_BY_ID_DICT = products[1]

def main(
    num_of_users: int,
    target_gcs_prefix: str,
    target_gcs_bucket: str,
    source_dir: str,
    extraneous_headers: typing.List[str],
) -> None:

    # read and generate location
    logging.info("generating data")
    for user_num in range(int(num_of_users)):
        logging.info(f"user transaction {user_num}")
        users.append(dataclasses.asdict(Users()))

    # remove extraneous keys in order_items
    logging.info("remove extraneous keys from order items")
    for oi in order_items:
        for key in extraneous_headers:
            del oi[key]

    # generate ghost events
    logging.info("generating ghost events")
    for user_num in range(int(num_of_users)):
        logging.info(f"ghost event {user_num}")
        GhostEvents()

    # write generated data to gcs
    table_dat = [users, orders, order_items, events, inventory_items]
    table_name = ["users", "orders", "order_items", "events", "inventory_items"]
    for name, table_dat in list(zip(table_name, table_dat)):
        logging.info(f"converting {name} dict to csv")
        csv_data = dict_to_csv(table_dat)
        logging.info(
            f"uploading output file to... gs://{target_gcs_bucket}/{target_gcs_prefix}/{name}.csv"
        )
        upload_to_bucket(
            target_bucket=target_gcs_bucket,
            target_prefix=target_gcs_prefix,
            target_object=f"{name}.csv",
            source_data=csv_data,
        )

    # upload static data to gcs
    file_names = ["products.csv", "distribution_centers.csv"]
    for file in file_names:
        logging.info(
            f"uploading output file to... gs://{target_gcs_bucket}/{target_gcs_prefix}/{file}"
        )
        upload_to_bucket(
            target_bucket=target_gcs_bucket,
            target_prefix=target_gcs_prefix,
            target_object=f"{file}",
            source_filepath=f"{source_dir}/{file}",
        )

# returns random address based off specified distribution
def get_address(
    *, country: str = "*", state: str = "*", postal_code: str = "*"
) -> dict:
    # country = '*' OR country = 'USA' OR country={'USA':.75,'UK':.25}
    # state = '*' OR state = 'California' OR state={'California':.75,'New York':.25}
    # postal_code = '*' OR postal_code = '95060' OR postal_code={'94117':.75,'95060':.25}
    # type checking is used to provide flexibility of inputs to function (ie. can be dict with proportions, or could be single string value)
    universe = []
    if postal_code != "*":
        if type(postal_code) == str:
            universe += list(
                filter(lambda row: row["postal_code"] == postal_code, LOCATION_DATA)
            )
        elif type(postal_code) == dict:
            universe += list(
                filter(
                    lambda row: row["postal_code"] in postal_code.keys(), LOCATION_DATA
                )
            )
    if state != "*":
        if type(state) == str:
            universe += list(filter(lambda row: row["state"] == state, LOCATION_DATA))
        elif type(state) == dict:
            universe += list(
                filter(lambda row: row["state"] in state.keys(), LOCATION_DATA)
            )
    if country != "*":
        if type(country) == str:
            universe += list(
                filter(lambda row: row["country"] == country, LOCATION_DATA)
            )
        elif type(country) == dict:
            universe += list(
                filter(lambda row: row["country"] in country.keys(), LOCATION_DATA)
            )
    if len(universe) == 0:
        universe = LOCATION_DATA

    total_pop = sum([int(loc["population"]) for loc in universe])

    for loc in universe:
        loc["population"] = int(loc["population"])
        if type(postal_code) == dict:
            if loc["postal_code"] in postal_code.keys():
                loc["population"] = postal_code[loc["postal_code"]] * total_pop
        if type(state) == dict:
            if loc["state"] in state.keys():
                loc["population"] = (
                    state[loc["state"]]
                    * (
                        loc["population"]
                        / sum(
                            [
                                loc2["population"]
                                for loc2 in universe
                                if loc["state"] == loc2["state"]
                            ]
                        )
                    )
                    * total_pop
                )
        if type(country) == dict:
            if loc["country"] in country.keys():
                loc["population"] = (
                    country[loc["country"]]
                    * (
                        loc["population"]
                        / sum(
                            [
                                loc2["population"]
                                for loc2 in universe
                                if loc["country"] == loc2["country"]
                            ]
                        )
                    )
                    * total_pop
                )

    loc = random.choices(
        universe, weights=[loc["population"] / total_pop for loc in universe]
    )[0]
    return {
        "street": fake.street_address(),
        "city": loc["city"],
        "state": loc["country"],
        "postal_code": loc["postal_code"],
        "country": loc["country"],
        "latitude": loc["latitude"],
        "longitude": loc["longitude"],
    }

# generates random date between now and specified date
def created_at(start_date: datetime.datetime) -> datetime.datetime:
    end_date = datetime.datetime.now()
    time_between_dates = end_date - start_date
    days_between_dates = time_between_dates.days
    if days_between_dates <= 1:
        days_between_dates = 2
    random_number_of_days = random.randrange(1, days_between_dates)
    created_at = (
        start_date
        + datetime.timedelta(days=random_number_of_days)
        + datetime.timedelta(minutes=random.randrange(MINUTES_IN_HOUR * 19))
    )
    return created_at

# generate URI for events table
def generate_uri(event: str, product: str) -> str:
    if event == "product":
        return f"/{event}/{product[0]}"
    elif event == "department":
        return f"""/{event}/{product[5].lower()}/category/{product[4].lower().replace(" ", "")}/brand/{product[1].lower().replace(" ", "")}"""
    else:
        return f"/{event}"

# converts list of dicts into csv format
def dict_to_csv(data: dict) -> str:
    output = io.StringIO()
    writer = csv.writer(output, quoting=csv.QUOTE_NONNUMERIC)
    header_writer = csv.DictWriter(output, fieldnames=data[0].keys())
    header_writer.writeheader()
    for dat in data:
        writer.writerow(dat.values())
    return output.getvalue()

# upload into GCS Bucket
def upload_to_bucket(
    target_bucket: str,
    target_prefix: str,
    target_object: str,
    source_data: list = None,
    source_filepath: str = None,
) -> str:
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(target_bucket)
    blob = bucket.blob(f"{target_prefix}/{target_object}")

    if source_data:
        blob.upload_from_string(str(source_data), content_type="text/csv")
    else:
        blob.upload_from_filename(source_filepath)
    return blob.public_url

# utility class
class DataUtil:
    def child_created_at(
        self, probability: str = "uniform"
    ) -> datetime.datetime:  # returns a random timestamp between now and parent date
        time_between_dates = datetime.datetime.now() - self.parent.created_at
        days_between_dates = time_between_dates.days
        if days_between_dates <= 1:
            days_between_dates = 2
        random_number_of_days = random.randrange(
            1, days_between_dates
        )  # generates random day between now and when user initially got created
        created_at = self.parent.created_at + datetime.timedelta(
            days=random_number_of_days
        )
        return created_at

    def random_item(
        self, population, **distribution
    ) -> str:  # returns single random item from a list based off distribution
        if distribution:
            return random.choices(
                population=population, weights=distribution["distribution"]
            )[0]
        else:
            return random.choices(population=population)[0]

@dataclasses.dataclass
class Address(object):
    def __init__(self, data: dict):
        self.street = data["street"]
        self.city = data["city"]
        self.state = data["state"]
        self.postal_code = data["postal_code"]
        self.country = data["country"]
        self.latitude = data["latitude"]
        self.longitude = data["longitude"]

    def __str__(self):
        return f"{self.street} \n{self.city}, {self.state} \n{self.postal_code} \n{self.country} \n{self.latitude} \n{self.longitude}"

@dataclasses.dataclass
class Users(DataUtil):
    logging.info("generating user")
    id: int = dataclasses.field(default_factory=itertools.count(start=1).__next__)
    first_name: str = dataclasses.field(init=False)
    last_name: str = dataclasses.field(init=False)
    email: str = dataclasses.field(init=False)
    age: int = dataclasses.field(init=False)
    gender: str = dataclasses.field(init=False)
    state: str = dataclasses.field(init=False)
    street_address: str = dataclasses.field(init=False)
    postal_code: str = dataclasses.field(init=False)
    city: str = dataclasses.field(init=False)
    country: str = dataclasses.field(init=False)
    latitude: float = dataclasses.field(init=False)
    longitude: float = dataclasses.field(init=False)
    traffic_source: str = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)

    def __post_init__(self):
        self.gender = self.random_item(population=["M", "F"])  # uniform distribution
        if self.gender == "M":
            self.first_name = fake.first_name_male()
            self.traffic_source = self.random_item(
                population=["Organic", "Facebook", "Search", "Email", "Display"],
                distribution=[0.15, 0.06, 0.7, 0.05, 0.04],
            )
        if self.gender == "F":
            self.first_name = fake.first_name_female()
            self.traffic_source = self.random_item(
                population=["Organic", "Facebook", "Search", "Email", "Display"],
                distribution=[0.15, 0.06, 0.7, 0.05, 0.04],
            )
        self.last_name = fake.last_name_nonbinary()
        address = Address(get_address())
        self.state = address.state
        self.street_address = address.street
        self.postal_code = address.postal_code
        self.city = address.city
        self.country = address.country
        self.latitude = address.latitude
        self.longitude = address.longitude
        self.email = f"{self.first_name.lower()}{self.last_name.lower()}@{fake.safe_domain_name()}"
        self.age = random.randrange(MIN_AGE, MAX_AGE)
        # weight newer users/orders
        choice = random.choices([0, 1], weights=[0.975, 0.025])[0]
        if choice == 0:
            self.created_at = created_at(datetime.datetime(2019, 1, 1))
        if choice == 1:
            self.created_at = created_at(
                datetime.datetime.now() - datetime.timedelta(days=7)
            )
        num_of_orders = random.choices(
            population=[0, 1, 2, 3, 4], weights=[0.2, 0.5, 0.2, 0.05, 0.05]
        )[0]
        if num_of_orders == 0:
            pass
        else:
            for _ in range(num_of_orders):
                orders.append(dataclasses.asdict(Order(user=self)))

    def __str__(self):
        return f"{self.id}, {self.first_name}, {self.last_name}, {self.email}, {self.age}, {self.gender}, {self.state}, {self.street_address}, {self.postal_code}, {self.city}, {self.traffic_source}, {self.created_at}"

@dataclasses.dataclass
class Product:
    logging.info("generating product")
    product_id: int = dataclasses.field(init=False)
    brand: str = dataclasses.field(init=False)
    name: str = dataclasses.field(init=False)
    cost: float = dataclasses.field(init=False)
    category: str = dataclasses.field(init=False)
    department: str = dataclasses.field(init=False)
    sku: str = dataclasses.field(init=False)
    retail_price: float = dataclasses.field(init=False)
    distribution_center_id: int = dataclasses.field(init=False)

    def __post_init__(self):
        person = Users()
        random_idx = np.random.choice(
            a=len(PRODUCT_GENDER_DICT[person.gender]), size=1
        )[0]
        product = PRODUCT_GENDER_DICT[person.gender][random_idx]
        self.brand = product[0]
        self.name = product[1]
        self.cost = product[2]
        self.category = product[3]
        self.department = product[4]
        self.sku = product[5]
        self.retail_price = product[6]
        self.distribution_center_id = product[7]

    def __str__(self):
        return f"{self.brand}, {self.name}, {self.cost}, {self.category}, {self.department}, {self.sku}, {self.retail_price}, {self.distribution_center_id}"

@dataclasses.dataclass
class Order(DataUtil):
    logging.info("generating order")
    order_id: int = dataclasses.field(default_factory=itertools.count(start=1).__next__)
    user_id: int = dataclasses.field(init=False)
    status: str = dataclasses.field(init=False)
    gender: str = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)
    returned_at: datetime.datetime = dataclasses.field(init=False)
    shipped_at: datetime.datetime = dataclasses.field(init=False)
    delivered_at: datetime.datetime = dataclasses.field(init=False)
    num_of_item: int = dataclasses.field(init=False)
    user: dataclasses.InitVar[typing.Any] = None

    def __post_init__(self, user=None):
        self.parent = user
        self.user_id = user.id
        self.gender = user.gender
        self.status = self.random_item(
            population=["Complete", "Cancelled", "Returned"],
            distribution=[0.85, 0.05, 0.1],
        )
        self.created_at = self.child_created_at()
        # add random generator for days it takes to ship, deliver, return etc.
        if self.status == "Returned":
            self.shipped_at = self.created_at + datetime.timedelta(
                minutes=random.randrange(MINUTES_IN_DAY * 3)
            )  # shipped between 0-3 days after order placed
            self.delivered_at = self.shipped_at + datetime.timedelta(
                minutes=random.randrange(MINUTES_IN_DAY * 5)
            )  # delivered between 0-5 days after ship date
            self.returned_at = self.delivered_at + datetime.timedelta(
                minutes=random.randrange(MINUTES_IN_DAY * 3)
            )  # returned 0-3 days after order is delivered
        elif self.status == "Complete":
            self.shipped_at = self.created_at + datetime.timedelta(
                minutes=random.randrange(MINUTES_IN_DAY * 3)
            )  # shipped between 0-3 days after order placed
            self.delivered_at = self.shipped_at + datetime.timedelta(
                minutes=random.randrange(MINUTES_IN_DAY * 5)
            )  # delivered between 0-5 days after ship date
            self.returned_at = None
        else:
            self.shipped_at = None
            self.delivered_at = None
            self.returned_at = None
        self.user = user  # pass person object to order_items
        # randomly generate number of items in an order
        num_of_items = self.random_item(
            population=[1, 2, 3, 4], distribution=[0.7, 0.2, 0.05, 0.05]
        )
        self.num_of_item = num_of_items
        for _ in range(num_of_items):
            order_items.append(dataclasses.asdict(OrderItem(order=self)))

    def __str__(self):
        return f"{self.order_id}, {self.user_id}, {self.status}, {self.created_at}, {self.shipped_at}, {self.delivered_at}, {self.returned_at}"

@dataclasses.dataclass
class Events:
    logging.info("generating event")
    id: int = dataclasses.field(default_factory=itertools.count(start=1).__next__)
    user_id: int = dataclasses.field(init=False)
    sequence_number: int = dataclasses.field(init=False)
    session_id: str = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)
    # inventory_item_id:int = field(init=False)
    ip_address: str = dataclasses.field(init=False)
    city: str = dataclasses.field(init=False)
    state: str = dataclasses.field(init=False)
    postal_code: str = dataclasses.field(init=False)
    browser: str = dataclasses.field(init=False)
    traffic_source: str = dataclasses.field(init=False)
    uri: str = dataclasses.field(init=False)
    event_type: str = dataclasses.field(init=False)
    order_item: dataclasses.InitVar[typing.Any] = None

    def __post_init__(self, order_item=None):
        self.sequence_number = order_item.sequence_number
        self.user_id = order_item.user_id
        self.created_at = order_item.created_at
        self.session_id = order_item.session_id
        self.ip_address = order_item.ip_address
        self.city = order_item.person.city
        self.state = order_item.person.state
        self.postal_code = order_item.person.postal_code
        self.event_type = order_item.event_type
        self.browser = order_item.browser
        self.uri = order_item.uri
        self.traffic_source = order_item.traffic_source

    def __str__(self):
        return f"{self.created_at}, {self.ip_address}, {self.city}, {self.state}, {self.postal_code}"

inv_item_id = 0

@dataclasses.dataclass
class OrderItem(DataUtil):
    logging.info("generating order item")
    id: int = dataclasses.field(default_factory=itertools.count(start=1).__next__)
    order_id: int = dataclasses.field(init=False)
    user_id: int = dataclasses.field(init=False)
    product_id: int = dataclasses.field(init=False)
    inventory_item_id: int = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)

    shipped_at: datetime.datetime = dataclasses.field(init=False)
    delivered_at: datetime.datetime = dataclasses.field(init=False)
    returned_at: datetime.datetime = dataclasses.field(init=False)

    sale_price: float = dataclasses.field(init=False)

    # extras
    event_type: str = dataclasses.field(init=False)
    ip_address: str = dataclasses.field(init=False)
    browser: str = dataclasses.field(init=False)
    traffic_source: str = dataclasses.field(init=False)
    session_id: str = dataclasses.field(init=False)
    sequence_number: int = dataclasses.field(init=False)
    uri: str = dataclasses.field(init=False)
    is_sold: bool = dataclasses.field(init=False)
    order: dataclasses.InitVar[typing.Any] = None

    def __post_init__(self, order=None):
        global inv_item_id

        self.order_id = order.order_id
        self.user_id = order.user_id
        inv_item_id = inv_item_id + 1
        self.inventory_item_id = inv_item_id
        self.created_at = order.created_at - datetime.timedelta(
            seconds=random.randrange(SECONDS_IN_MINUTE * 240)
        )  # order purchased within 4 hours

        self.shipped_at = order.shipped_at
        self.delivered_at = order.delivered_at
        self.returned_at = order.returned_at

        random_idx = np.random.choice(a=len(PRODUCT_GENDER_DICT[order.gender]), size=1)[
            0
        ]
        product = PRODUCT_GENDER_DICT[order.gender][random_idx]
        self.product_id = product[0]
        self.sale_price = product[3]
        self.ip_address = fake.ipv4()
        self.browser = self.random_item(
            population=["IE", "Chrome", "Safari", "Firefox", "Other"],
            distribution=[0.05, 0.5, 0.2, 0.2, 0.05],
        )
        self.traffic_source = self.random_item(
            population=["Email", "Adwords", "Organic", "YouTube", "Facebook"],
            distribution=[0.45, 0.3, 0.05, 0.1, 0.1],
        )
        self.session_id = str(uuid.uuid4())

        self.person = order.user  # pass person object to events
        self.is_sold = True
        previous_created_at = None

        # Generate Events Table
        if order.num_of_item == 1:  # if only 1 item in order go through flow
            for idx, val in enumerate(
                ["home", "department", "product", "cart", "purchase"]
            ):
                self.sequence_number = idx + 1
                self.event_type = val
                self.uri = generate_uri(val, product)
                events.append(dataclasses.asdict(Events(order_item=self)))
                previous_created_at = self.created_at
                self.created_at = previous_created_at + datetime.timedelta(
                    seconds=random.randrange(SECONDS_IN_MINUTE * 3)
                )
        else:  # if multiple items
            sequence_num = 0  # track sequence num of purchase event
            for _ in range(order.num_of_item):
                for event in ["department", "product", "cart"]:
                    sequence_num += 1
                    self.sequence_number = sequence_num
                    self.event_type = event
                    self.uri = generate_uri(event, product)
                    events.append(dataclasses.asdict(Events(order_item=self)))
                    sequence_num = self.sequence_number
                    previous_created_at = self.created_at
                    self.created_at = previous_created_at + datetime.timedelta(
                        seconds=random.randrange(180)
                    )
            self.sequence_number = sequence_num + 1
            self.created_at += datetime.timedelta(random.randrange(5))
            self.event_type = "purchase"
            self.uri = generate_uri("purchase", product)
            events.append(dataclasses.asdict(Events(order_item=self)))

        # sold inventory item
        inventory_items.append(dataclasses.asdict(InventoryItem(order_item=self)))

        # unsold inventory items
        num_of_items = self.random_item(
            population=[1, 2, 3], distribution=[0.5, 0.3, 0.2]
        )
        for _ in range(num_of_items):
            self.is_sold = False
            inv_item_id += 1
            self.inventory_item_id = inv_item_id
            inventory_items.append(dataclasses.asdict(InventoryItem(order_item=self)))

@dataclasses.dataclass
class InventoryItem:
    id: int = dataclasses.field(init=False)
    product_id: int = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)
    sold_at: datetime.datetime = dataclasses.field(init=False)
    cost: float = dataclasses.field(init=False)
    product_category: str = dataclasses.field(init=False)
    product_name: str = dataclasses.field(init=False)
    product_brand: str = dataclasses.field(init=False)
    product_retail_price: float = dataclasses.field(init=False)
    product_department: str = dataclasses.field(init=False)
    product_sku: str = dataclasses.field(init=False)
    product_distribution_center_id: int = dataclasses.field(init=False)
    order_item: dataclasses.InitVar[typing.Any] = None

    def __post_init__(self, order_item=None):
        self.id = order_item.inventory_item_id
        self.product_id = order_item.product_id
        if order_item.is_sold is True:
            self.created_at = order_item.created_at - datetime.timedelta(
                minutes=random.randrange(86400)
            )  # in inventory between 0 and 60 days
            self.sold_at = (
                order_item.created_at
            )  # sold on the date/time the order_items was logged
        if order_item.is_sold is False:
            self.created_at = created_at(datetime.datetime(2020, 1, 1))
            self.sold_at = None
        self.cost = PRODUCT_BY_ID_DICT[self.product_id]["cost"]
        self.product_category = PRODUCT_BY_ID_DICT[self.product_id]["category"]
        self.product_name = PRODUCT_BY_ID_DICT[self.product_id]["name"]
        self.product_brand = PRODUCT_BY_ID_DICT[self.product_id]["brand"]
        self.product_retail_price = PRODUCT_BY_ID_DICT[self.product_id]["retail_price"]
        self.product_department = PRODUCT_BY_ID_DICT[self.product_id]["department"]
        self.product_sku = PRODUCT_BY_ID_DICT[self.product_id]["sku"]
        self.product_distribution_center_id = PRODUCT_BY_ID_DICT[self.product_id][
            "distribution_center_id"
        ]

    def __str__(self):
        return f"{self.id}, {self.product_id}, {self.created_at}, {self.cost}, {self.product_category}, {self.product_name}, {self.product_brand}, {self.product_retail_price}, {self.product_department}, {self.product_sku}, {self.product_distribution_center_id}"

@dataclasses.dataclass
class GhostEvents(DataUtil):
    id: int = dataclasses.field(init=False)
    user_id: int = dataclasses.field(init=False)
    sequence_number: int = dataclasses.field(init=False)
    session_id: str = dataclasses.field(init=False)
    created_at: datetime.datetime = dataclasses.field(init=False)
    ip_address: str = dataclasses.field(init=False)
    city: str = dataclasses.field(init=False)
    state: str = dataclasses.field(init=False)
    postal_code: str = dataclasses.field(init=False)
    browser: str = dataclasses.field(init=False)
    traffic_source: str = dataclasses.field(init=False)
    uri: str = dataclasses.field(init=False)
    event_type: str = dataclasses.field(init=False)

    def __post_init__(self):
        address = get_address()
        self.sequence_number = 0
        self.user_id = None
        self.created_at = created_at(datetime.datetime(2019, 1, 1))
        self.session_id = str(uuid.uuid4())
        self.ip_address = fake.ipv4()
        self.city = address["city"]
        self.state = address["state"]
        self.postal_code = address["postal_code"]
        self.browser = self.random_item(
            population=["IE", "Chrome", "Safari", "Firefox", "Other"],
            distribution=[0.05, 0.5, 0.2, 0.2, 0.05],
        )
        self.traffic_source = self.random_item(
            population=["Email", "Adwords", "Organic", "YouTube", "Facebook"],
            distribution=[0.45, 0.3, 0.05, 0.1, 0.1],
        )

        products = PRODUCT_GENDER_DICT[
            self.random_item(population=["M", "F"], distribution=[0.5, 0.5])
        ]
        product = self.random_item(products)

        # different event type combinations
        cancelled_browsing = ["product", "cart", "cancel"]
        abandoned_cart = ["department", "product", "cart"]
        viewed_product = ["product"]
        viewed_department = ["department", "product"]

        random_events = self.random_item(
            population=[
                cancelled_browsing,
                abandoned_cart,
                viewed_product,
                viewed_department,
            ]
        )

        for event in random_events:
            # set ghost events ID to max of original
            event_id = len(events)
            self.id = event_id + 1
            event_id = self.id

            self.event_type = event
            self.uri = generate_uri(event, product)
            self.sequence_number += 1
            self.created_at = self.created_at + datetime.timedelta(
                minutes=random.randrange(int(MINUTES_IN_HOUR * 0.5))
            )
            events.append(dataclasses.asdict(self))

    def __str__(self):
        return f"{self.created_at}, {self.ip_address}, {self.city}, {self.state}, {self.postal_code}"

if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)
    main(
        num_of_users=int(os.environ["NUM_OF_USERS"]),
        target_gcs_prefix=os.environ["TARGET_GCS_PREFIX"],
        target_gcs_bucket=os.environ["TARGET_GCS_BUCKET"],
        source_dir=os.environ["SOURCE_DIR"],
        extraneous_headers=json.loads(os.environ["EXTRANEOUS_HEADERS"]),
    )

max_stars_repo_name: arjunsgill/public-datasets-pipelines

max_issues_repo_name: arjunsgill/public-datasets-pipelines

max_forks_repo_name: arjunsgill/public-datasets-pipelines


text:
#import mahotas as mh
import numpy as np
import cv2
import PIL
import math
import argparse
import imutils

from picamera.array import PiRGBArray
from picamera import PiCamera
import time

"""
    | eixoY
    |
    |
    |
    |
    |               eixoX
  ----------------------
    |

"""

def busca_cruzamento(eixoY):
    eixoX=1
    saida=0
    eixoX2=1
    img=0
    while(eixoX<comprimento):
        if saida==0:
            (b)=thresholding[eixoY,eixoX]
            if b==255:
                eixoX2=eixoX
                while(eixoX2<comprimento):
                    (b1)=thresholding[eixoY,eixoX2]
                    if b1==255:
                        eixoX2=eixoX2+1
                        img=img+1
                        saida=1
                    if b1==0:
                        saida=0
                        return 0
                        break
                if eixoX2==comprimento:
                    break
            eixoX=eixoX+1
    if saida==1 and img>comprimento:
        return 1
    if saida==0:
        return 0

def busca_direita(eixoY):
    eixoX=1
    saida=0
    eixoX2=1
    while(eixoX<comprimento):
        if saida==0:
            (b)=thresholding[eixoY,eixoX]
            if b==255:
                eixoX2=eixoX
                while(eixoX2<comprimento):
                    (b1)=thresholding[eixoY,eixoX2]
                    if b1==255:
                        saida=1
                        #cv2.line(frame,(a0,eixoY),(eixoX2,eixoY),(0,0,255),5)
                        break
                    eixoX2=eixoX2+1
            eixoX=eixoX+1
        if saida==1:
            return (eixoX2,eixoY,1)
            break
    if saida==0:
        return(1000,1000,0)

def busca_esquerda(eixoY):
    eixoX=comprimento-1
    saida=0
    eixoX2=comprimento-1
    while(eixoX>0):
        if saida==0:
            (b)=thresholding[eixoY,eixoX]
            if b==255 :
                eixoX2=eixoX
                while(eixoX2>0):
                    (b1)=thresholding[eixoY,eixoX2]
                    if b1==255 :
                        saida=1
                        #cv2.line(frame,(a0,eixoY),(img1,eixoY),(0,0,255),5)
                        break
                    eixoX2=eixoX2-1
            eixoX=eixoX-1
        if saida==1:
            return (eixoX2,eixoY,1)
            break
    if saida==0:
        return (1000,1000,0)

#############################
def busca_direita_esquerda_teste(eixoY):
    eixoX=comprimento
    saida=0
    eixoX2=comprimento
    contador=1
    estadoArmazenado=0
    while(eixoX>0):
        (b)=thresholding[eixoY,eixoX]
        if b==255:
            if estadoArmazenado==0:
                eixoX2=eixoX
            contador=contador+1
            eixoX=eixoX-1
            estadoArmazenado=1
        else:
            eixoX=eixoX-1
            contador=0
            estadoArmazenado=0
        if(contador>65):
            saida=1
            cv2.line(frame,(eixoX2,eixoY),(eixoX2+1,eixoY),(0,0,255),5)
            return (eixoX2,eixoY,1)
            break

    if saida==1:
        return (eixoX2,eixoY,1)
    if saida==0:
        return (1000,1000,0)

#############################
#############################
def busca_esquerda_direita_teste(eixoY):
    eixoX=1
    saida=0
    eixoX2=1
    contador=1
    estadoArmazenado=0
    while(eixoX<comprimento):
        (b)=thresholding[eixoY,eixoX]
        if b==255:
            if estadoArmazenado==0:
                eixoX2=eixoX
            contador=contador+1
            eixoX=eixoX+1
            estadoArmazenado=1
        else:
            eixoX=eixoX+1
            contador=0
            estadoArmazenado=0
        if(contador>65):
            saida=1
            cv2.line(frame,(eixoX2,eixoY),(eixoX2+1,eixoY),(255,0,0),5)
            return (eixoX2,eixoY,1)
            break

    if saida==1:
        return (eixoX2,eixoY,1)
    if saida==0:
        return (1000,1000,0)

#############################

def busca_centro(eixoY):
    bola0=busca_esquerda_direita_teste(eixoY)
    bola1=busca_direita_esquerda_teste(eixoY)
    if bola0[2]==1 or bola1[2]==1:
        #cv2.line(frame,(eixoX+2,eixoY),(eixoX2-2,eixoY),(255,0,0),2)
        #cordenadas das duas extremidades da linha preta
        dx2 = (bola0[0]-bola1[0])**2
        dy2 = (bola0[1]-bola1[1])**2
        distanceFloat=math.sqrt(dx2 + dy2)
        distanceInt=int((distanceFloat)/2)
        #a cordenada da esquerda acrescida a metade da distancia entre as duas
        #extremidades da linha preta, logo, resulta no centro da linha
        centro=bola0[0]+distanceInt
        cv2.circle(frame,(centro,eixoY),5,(120,50,120),-1)
        return (centro,eixoY,bola0[1],eixoY)
    if bola0[2]==0 or bola1[2]==0:
        return (1000,1000,1000,1000)

def distance_roxo_centro(bola,centro):
    retornoLado=0
    #condição inibidora
    if(bola[0]!=1000 and bola[1]!=1000 and centro[0]!=1000 and centro[1]!=1000):
        #distancia do centro a bola
        dx2 = (bola[0]-centro[0])**2
        dy2 = (bola[1]-centro[1])**2
        #direita
        if meio<bola[0]:
            zi=int((math.sqrt(dx2 + dy2))*-1)
            cv2.putText(frame,'>>',(meio,centro[1]+40),cv2.FONT_HERSHEY_SIMPLEX,0.7,(0,100,255),2,cv2.LINE_AA)
            cv2.putText(frame,str(zi),(meio,centro[1]+23),cv2.FONT_HERSHEY_SIMPLEX,0.7,(255,255,255),2,cv2.LINE_AA)
        #esquerda
        else:
            zi=int(math.sqrt(dx2 + dy2))
            cv2.putText(frame,'<<',(meio,centro[1]+40),cv2.FONT_HERSHEY_SIMPLEX,0.7,(100,0,255),2,cv2.LINE_AA)
            cv2.putText(frame,str(zi),(meio,centro[1]+23),cv2.FONT_HERSHEY_SIMPLEX,0.7,(255,255,255),2,cv2.LINE_AA)
        return (zi)
    else:
        cv2.putText(frame,'0',(meio,centro[1]+40),cv2.FONT_HERSHEY_SIMPLEX,0.7,(55,100,55),2,cv2.LINE_AA)
        return (1000)

def busca_branco_direita(bola):
    eixoX=bola[0]+3
    saida=0
    img1=bola[0]+3
    eixoY=bola[1]
    while(eixoX<comprimento):
        if saida==0:
            eixoX2=eixoX
            while(eixoX2<comprimento):
                (b1,g1,r1)=frame[eixoY,eixoX2]
                if b1==255 and g1==255 and r1==255:
                    saida=1
                    #print('Branco',eixoY,'yes')
                    break
                eixoX2=eixoX2+1
        eixoX=eixoX+1
        if saida==1:
            return 1
            break
        else:
            return 0
            break

def busca_branco_esquerda(bola):
    eixoX=bola[0]-3
    saida=0
    eixoX2=bola[0]-3
    eixoY=bola[1]
    while(eixoX<comprimento):
        if saida==0:
            eixoX2=eixoX
            while(eixoX2>0):
                (b1,g1,r1)=frame[eixoY,eixoX2]
                if b1==255 and g1==255 and r1==255:
                    saida=1
                    #print('Branco',eixoY,'yes')
                    break
                eixoX2=eixoX2-1
        eixoX=eixoX-1
        if saida==1:
            return 1
            break
        else:
            return 0
            break

#biblioteca da lei dos cossenos, calcula o angulo do triangulo, utilizando as
#cordenadas já definidas
def leiCos(bola3,bola4,bola5):

    #calculo das 3 distancias, 3 lados do triangulo
    dx2 = (bola4[0]-bola5[0])**2
    dy2 = (bola4[1]-bola5[1])**2
    zi=int(math.sqrt(dx2 + dy2))
    distance45 = str(zi)

    dx2 = (bola4[0]-bola3[0])**2
    dy2 = (bola4[1]-bola3[1])**2
    xi=int((math.sqrt(dx2 + dy2)))
    distancever = str(xi)

    dx2 = (bola5[0]-bola3[0])**2
    dy2 = (bola5[1]-bola3[1])**2
    yi=int(math.sqrt(dx2 + dy2))
    distanceponta= str(yi)

    #calculo dos 3 angulos do triangulo
    if zi!=0 and xi!=0 and yi!=0:
        cosi=(((yi**2)-(zi**2)-(xi**2))/(2*xi*zi))*-1
        cof = np.cos(cosi)
        coffe = np.arccos(cof)
        coffe=cosi*coffe
        a=int(180+(coffe*180))

    if zi!=0 and xi!=0 and yi!=0:
        cosi=(((zi**2)-(yi**2)-(xi**2))/(2*xi*yi))*-1
        cof = np.cos(cosi)
        coffe = np.arccos(cof)
        coffe=cosi*coffe
        b=int(180-(coffe*180))

    if zi!=0 and xi!=0 and yi!=0:
        cosi=(((xi**2)-(yi**2)-(zi**2))/(2*zi*yi))*-1
        cof = np.cos(cosi)
        coffe = np.arccos(cof)
        coffe=cosi*coffe
        c=int(180-(coffe*180))

    #lugar onde é exibido o texto
    local=(parte2,30)
    if a>90 and b>30 and busca_branco_direita(bola4)==1:
        cv2.putText(frame,'curva de 90 direita',local,cv2.FONT_HERSHEY_SIMPLEX,0.8,(0,100,255),3,cv2.LINE_AA)

    elif a>90 and b>30 and busca_branco_esquerda(bola4)==1:
        cv2.putText(frame,'curva de 90 esquerda',local,cv2.FONT_HERSHEY_SIMPLEX,0.8,(100,0,255),3,cv2.LINE_AA)

    elif a>10 and busca_branco_esquerda(bola4)==1:
        cv2.putText(frame,'leve curva a esquerda',local,cv2.FONT_HERSHEY_SIMPLEX,0.8,(100,0,255),2,cv2.LINE_AA)
    elif a>4 and busca_branco_direita(bola4)==1:
        cv2.putText(frame,'leve curva a direita',local,cv2.FONT_HERSHEY_SIMPLEX,0.8,(0,100,255),2,cv2.LINE_AA)
    else:
        cv2.putText(frame,'reta',local,cv2.FONT_HERSHEY_SIMPLEX,0.8,(255,100,0),2,cv2.LINE_AA)

#variáveis pré definidas
altura=479
comprimento=639
bola5=bola4=bola3=bola2=bola1=(0,90)
roxo=[0,0,0,0,0]
partex=int(altura/6)
x=35
parte0=int(partex*1)+x
parte1=int(partex*2)+x
parte2=int(partex*3)+x
parte3=int(partex*4)+x
parte4=int(partex*5)+x
meio=320
isto=(0,0)
somatorioErro=0

#configurações da camera
camera = PiCamera()
camera.resolution = (640, 480)
camera.framerate = 30
camera.sharpness=0
camera.contrast=40
camera.brightness=60
camera.saturation=0
camera.ISO=100
camera.exposure_compensation=0
camera.exposure_mode='auto'
camera.meter_mode='average'
camera.awb_mode='tungsten'
camera.image_effect='none'
camera.color_effects= None
camera.rotation= 10
camera.hflip=False
camera.vflip=False
camera.crop=(0.0,0.0,1.0,1.0)

rawCapture = PiRGBArray(camera, size = (640, 480))
time.sleep(0.1)

for frame0 in camera.capture_continuous(rawCapture, format = 'bgr',
                                       use_video_port = True):

    # Captura frame por frame
    frame2 = frame0.array
    key = cv2.waitKey(1) & 0xFF
    rawCapture.truncate(0)

    #filtros e tratamento da imagem
    frame=cv2.GaussianBlur(frame2,(5,5),0)
    frame2=cv2.cvtColor(frame2, cv2.COLOR_BGR2HSV)
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    #ret,thresholding=cv2.threshold(gray,127,255,cv2.THRESH_BINARY_INV)
    #thresholding=cv2.adaptiveThreshold(gray,255,cv2.ADAPTIVE_THRESH_GAUSSIAN_C,\
    #                                   cv2.THRESH_BINARY_INV,3,1)
    ret,thresholding = cv2.threshold(gray,0,255,cv2.THRESH_BINARY_INV+cv2.THRESH_OTSU)
    kernel = np.ones((5,5),np.uint8)
    thresholding = cv2.erode( thresholding, kernel, iterations=2)
    thresholding = cv2.dilate( thresholding, kernel, iterations=1)
    im2,contours,hierarchy = cv2.findContours(thresholding,cv2.RETR_TREE,cv2.CHAIN_APPROX_NONE )

                ##############################################
                ###############FUNÇÃO DAS BOXES###############
                ##############################################

    #Cria lista/vetor conforme o num de boxes
    #Acha a maior boxe e elimina todas as outras boxes menores
    area=[None]*len(contours)
    vetor=[None]*len(contours)
    num=0
    while (num<len(contours)):
        cnt = contours[num]
        a = cv2.contourArea(cnt)
        if a>10:
            rect = cv2.minAreaRect(cnt)
            box = cv2.boxPoints(rect)
            box = np.int0(box)
            #cv2.drawContours(frame,[box],0,(255,0,0),1)
            area[num]=(a)
            vetor[num]=([box])
        num=num+1

    #print(area.count(None))
    while int(area.count(None))!=0 :
        area.remove(None)
        vetor.remove(None)

    if(len(contours)>1):
        lek=area.index(max(area))
        area.remove(max(area))
        del(vetor[lek])
        #print("0",area,"0")
        #print([box])
        num=0
        while (num<len(area)):
            cv2.drawContours(thresholding,vetor[num],0,(0,0,0),-1)
            #cv2.drawContours(frame,vetor[num],0,(255,255,255),-1)
            num=num+1

                    #################FIM#################

    #thresholding = cv2.erode( thresholding, kernel, iterations=1)

    #busca centro da imagem e traça linha horizontal
    bola5=busca_centro(parte0)
    bolav1=(bola5[2],bola5[3])
    bola4=busca_centro(parte1)
    bolav2=(bola4[2],bola4[3])
    bola3=busca_centro(parte2)
    bola2=busca_centro(parte3)
    bola1=busca_centro(parte4)

    #testa as cordenadas válidas, e vai mudando o triangulo de posição
    if (bola3[0]!=1000 and bola4[0]!=1000 and bola5[0]!=1000):
        cv2.line(frame,(bola3[0],bola3[1]),(bola5[0],bola5[1]),(255,255,255),2)
        leiCos((bola3[0],bola3[1]),(bola4[0],bola4[1]),(bola5[0],bola5[1]))
        cv2.line(frame,(bola4[0],bola4[1]),(bola5[0],bola5[1]),(255,255,255),2)
        cv2.line(frame,(bola3[0],bola3[1]),(bola4[0],bola4[1]),(255,255,255),2)
        isto=bola5

    elif (bola2[0]!=1000 and bola3[0]!=1000 and bola4[0]!=1000):
        cv2.line(frame,(bola2[0],bola2[1]),(bola4[0],bola4[1]),(255,255,255),2)
        leiCos((bola2[0],bola2[1]),(bola3[0],bola3[1]),(bola4[0],bola4[1]))
        cv2.line(frame,(bola2[0],bola2[1]),(bola3[0],bola3[1]),(255,255,255),2)
        cv2.line(frame,(bola3[0],bola3[1]),(bola4[0],bola4[1]),(255,255,255),2)
        isto=bola4

    elif (bola1[0]!=1000 and bola2[0]!=1000 and bola3[0]!=1000):
        cv2.line(frame,(bola1[0],bola1[1]),(bola3[0],bola3[1]),(255,255,255),2)
        leiCos((bola1[0],bola1[1]),(bola2[0],bola2[1]),(bola3[0],bola3[1]))
        cv2.line(frame,(bola1[0],bola1[1]),(bola2[0],bola2[1]),(255,255,255),2)
        cv2.line(frame,(bola2[0],bola2[1]),(bola3[0],bola3[1]),(255,255,255),2)
        isto=bola3

    #quando todas as cordenadas são invalidas
    elif (bola1[0]==1000 and bola2[0]==1000 and bola3[0]==1000 and bola4[0]==1000 and bola5[0]==1000):
        cv2.putText(frame,"gap",(100,100),cv2.FONT_HERSHEY_SIMPLEX,0.7,(0,255,0),1,cv2.LINE_AA)

    elif bola2[0]!=1000:
        isto=bola2
    elif bola1[0]!=1000:
        isto=bola1

            #######################################################
            ##################CALCULO DO CONTROLE##################
            #######################################################

    #distancia do roxo da linha e do roxo central
    roxo[0]=distance_roxo_centro(bola5,(meio,parte0))
    roxo[1]=distance_roxo_centro(bola4,(meio,parte1))
    roxo[2]=distance_roxo_centro(bola3,(meio,parte2))
    roxo[3]=distance_roxo_centro(bola2,(meio,parte3))
    roxo[4]=distance_roxo_centro(bola1,(meio,parte4))

    #conta as bolas que estão para a direita e para esquerda do centro da img
    """i=0
    esquerda=0
    direita=0
    while i<5:
        if roxo[i]==2:
           direita=direita+1
        elif roxo[i]==1:
            esquerda=esquerda+1
        i=i+1
    if direita>esquerda:
        cv2.putText(frame,'direita',(400,parte4),cv2.FONT_HERSHEY_SIMPLEX,0.7,(0,100,255),2,cv2.LINE_AA)
    else:
        cv2.putText(frame,'esquerda',(400,parte4),cv2.FONT_HERSHEY_SIMPLEX,0.7,(100,0,255),2,cv2.LINE_AA)
    """
    passa=1
    if(roxo[3]!=0 and roxo[3]!=1000):
        erro=roxo[3]
    elif(roxo[2]!=0 and roxo[2]!=1000):
        erro=roxo[2]
    elif(roxo[4]!=0 and roxo[4]!=1000):
        erro=roxo[4]
    elif(roxo[1]!=0 and roxo[1]!=1000):
        erro=roxo[1]
    elif(roxo[0]!=0 and roxo[0]!=1000):
        erro=roxo[0]
    else:
        erro=1

    kp=2.15
    proporcional=kp*erro
    pwm=int((255*(proporcional+(320*kp)))/(640*kp))
    #cv2.putText(frame,str(pwm),(50,100),cv2.FONT_HERSHEY_SIMPLEX,\
     #           0.7,(100,0,255),2,cv2.LINE_AA)

    ti=1/(5)
    erroAnterior=erro
    somatorioErro=somatorioErro+erro
    if somatorioErro>2000:
        somatorioErro=2000
    if somatorioErro<-2000:
        somatorioErro=-2000
    integral=int((kp*erro)+(kp*(1/ti)*somatorioErro)+127)
    pwm=int((255*(integral+6627)/(13254)))
    if(pwm>255):
        pwm=255
    if(pwm<-255):
        pwm=-255

    cv2.putText(frame,str(pwm),(50,100),cv2.FONT_HERSHEY_SIMPLEX,\
                0.7,(100,0,255),2,cv2.LINE_AA)

                    #################FIM#################

    '''
    #circulos centrais
    cv2.circle(frame,(meio,parte0),2,(120,50,120),-1)
    cv2.circle(frame,(meio,parte1),2,(120,50,120),-1)
    cv2.circle(frame,(meio,parte2),2,(120,50,120),-1)
    cv2.circle(frame,(meio,parte3),2,(120,50,120),-1)
    cv2.circle(frame,(meio,parte4),2,(120,50,120),-1)
    '''
                #############################################
                ###############FUNÇÃO DO VERDE###############
                #############################################

    #define a mascara a ser considerada para o verde
    lower=np.array([40,60,50])
    upper=np.array([80,255,255])
    mask=cv2.inRange(frame2,lower,upper)
    mask2 = cv2.erode( mask, kernel, iterations=1)
    _,contours,_=cv2.findContours(mask2,cv2.RETR_TREE,cv2.CHAIN_APPROX_SIMPLE)
    NumeroContornos = str(len(contours))
    #cv2.drawContours(frame,contours,-1,(0,0,255),2)
    if int(NumeroContornos)>0:
        #mascara verde1
        mascara=np.zeros(frame.shape[:2],dtype="uint8")
        (cX,cY)=(frame.shape[1] // 2, frame.shape[0] // 2)
        cv2.circle(mascara,(isto[0]-55,isto[1]-5),20,255,-1)
        frame_mascara= cv2.bitwise_and(frame2,frame2,mask=mascara)

        cv2.circle(frame,(isto[0]-55,isto[1]-5),20,255,1)
        mask=cv2.inRange(frame_mascara,lower,upper)
        _,contours,_=cv2.findContours(mask,cv2.RETR_TREE,cv2.CHAIN_APPROX_SIMPLE)
        NumeroPXSBrancos = int(cv2.countNonZero(mask))
        #print(NumeroPXSBrancos)
        if int(NumeroPXSBrancos)>500:
            cv2.putText(frame,"verde lado esquerdo",(100,100),cv2.FONT_HERSHEY_SIMPLEX,0.9,(0,0,255),3,cv2.LINE_AA)
            esquerda=1
            direita=0
        #mascara verde2
        mascara=np.zeros(frame.shape[:2],dtype="uint8")
        (cX,cY)=(frame.shape[1] // 2, frame.shape[0] // 2)
        cv2.circle(mascara,(isto[0]+55,isto[1]-5),20,255,-1)
        frame_mascara= cv2.bitwise_and(frame2,frame2,mask=mascara)
        cv2.circle(frame,(isto[0]+55,isto[1]-5),20,255,1)

        mask=cv2.inRange(frame_mascara,lower,upper)
        _,contours,_=cv2.findContours(mask,cv2.RETR_TREE,cv2.CHAIN_APPROX_SIMPLE)
        NumeroPXSBrancos = int(cv2.countNonZero(mask))
        #print(NumeroPXSBrancos)
        if int(NumeroPXSBrancos)>500:
            cv2.putText(frame,"verde lado direito",(100,100),cv2.FONT_HERSHEY_SIMPLEX,0.9,(0,0,255),3,cv2.LINE_AA)
            direita=1
            esquerda=0

                    #################FIM#################

    #exibe o resultado
    cv2.imshow('frame',frame)
    cv2.imshow('thresholding',thresholding)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cv2.destroyAllWindows()

max_stars_repo_name: lorenzoppx/Line-Follower-OpenCV

max_issues_repo_name: lorenzoppx/Line-Follower-OpenCV

max_forks_repo_name: lorenzoppx/Line-Follower-OpenCV