text:
from tqdm import tqdm
import os
import numpy as np
import matplotlib.pyplot as plt # noqa
import logging

logger = logging.getLogger(__name__)

class Model:
    """Model abstract class"""
    def __init__(self):
        self.layers = []
        self.train_loss_idx = []
        self.test_loss_idx = []
        self.train_loss = []
        self.test_loss = []

    def __len__(self):
        pass

    def __str__(self):
        out_string = ""
        for i, _ in list(enumerate(self.layers)):
            if self.layers[i].contains_weights:
                out_string += f"Layer : {i} {self.layers[i]}"
                out_string += f" Weights, {self.layers[i].weights.shape}"
                out_string += f" Bias {self.layers[i].bias.shape}"
            else:
                out_string += f"Layer : {i} {self.layers[i]}"
            out_string += "\n"
        return out_string

    def split_train_test(self, X, Y, train_split_ratio=0.8, random_seed=42):
        train_end_index = int(train_split_ratio * X.shape[0])
        logging.info(f"Data splitting into {train_end_index} Train and {X.shape[0]-train_end_index} test data points") # noqa
        idx = np.random.RandomState(seed=random_seed).permutation(X.shape[0])
        train_idx, test_idx = idx[:train_end_index], idx[train_end_index:]
        return X[train_idx, :], Y[train_idx], X[test_idx, :], Y[test_idx]

    def get_loss_function(self, loss_function):
        if loss_function == 'BinaryCrossEntropy':
            from deepnumpy.loss.classification_loss import BinaryCrossEntropy  # noqa
        elif loss_function == 'MeanSquaredError':
            from deepnumpy.loss.regression_loss import MeanSquaredError  # noqa
        else:
            logger.error("Selected Loss function is not implemented."
                         "Taking default loss function 'BinaryCrossEntropy'")
            loss_function == 'BinaryCrossEntropy'

        return eval(loss_function)

    def train(self, X, Y, learning_rate=0.01,
              epochs=100, loss_function='BinaryCrossEntropy',
              metrics="ClassificationMetrics",
              train_split_ratio=0.8, random_seed=42,
              early_stopping={},
              output_dir='./output', verbose=False):

        logger.info(f"Training Specs: \n"
                    f"Learning Rate: {learning_rate} \n"
                    f"Maximum Epochs Limit: {epochs} \n"
                    f"Loss Function: {loss_function} \n"
                    f"Metrics Function: {metrics} \n"
                    f"Train Test Split Ratio Function: {train_split_ratio} \n"
                    f"Early Stopping: {early_stopping} \n")

        if early_stopping:
            self.initialize_monitoring(early_stopping)

        x_train, y_train, x_test, y_test = self.split_train_test(X, Y, train_split_ratio, random_seed) # noqa
        self.loss_func_instance = self.get_loss_function(loss_function)
        for this_epoch in tqdm(range(epochs)):
            this_loss = self.optimize(x_train, y_train, learning_rate, self.loss_func_instance) # noqa
            self.train_loss.append(this_loss)
            self.train_loss_idx.append(this_epoch)
            if verbose:
                if this_epoch % 10 == 0:
                    logger.info(f"Epoch: {this_epoch}. Training Loss: {this_loss}") # noqa
                    this_loss = self.evaluate(x_test, y_test, metrics=metrics, output_dir=output_dir) # noqa
                    self.test_loss.append(this_loss)
                    self.test_loss_idx.append(this_epoch)
                    self.plot_loss_curve(output_dir)
                    if early_stopping:
                        if self.monitor_for_early_stopping({'val_loss': this_loss}): # noqa
                            logging.info(f"Training stopped due to satisfied early stopping condition at {this_epoch} epoch") # noqa
                            return None

    def predict(self, input_x):
        '''
        Input --> Layer1 --> Act1 --> Layer2 --> Act2 --> Output
        '''
        forward_value = input_x.copy()
        for i, _ in enumerate(self.layers):
            forward_value = self.layers[i].forward(forward_value)
        return forward_value

    def initialize_monitoring(self, early_stopping):
        self.early_stopping_dict = early_stopping
        self.prev_value_dict = {'val_loss': np.inf, 'val_acc': 0}
        self.monitoring_dict = {}
        for this_key in early_stopping.keys():
            self.monitoring_dict[this_key] = 0

    def monitor_for_early_stopping(self, current_values_dict: dict):
        stop_flag = False
        for this_key in self.early_stopping_dict.keys():
            if this_key not in current_values_dict.keys():
                raise "Monitoring Value not calculated"
            if (((self.early_stopping_dict.get(this_key)[0] == 'inc') and (current_values_dict[this_key] >= self.prev_value_dict.get(this_key)))  # noqa
                or ((self.early_stopping_dict.get(this_key)[0] == 'dec') and (current_values_dict[this_key] <= self.prev_value_dict.get(this_key)))):  # noqa
                self.monitoring_dict[this_key] += 1
                logging.info("Identified Early Stopping condition. Counter incremented")  # noqa
                if self.monitoring_dict[this_key] > self.early_stopping_dict.get(this_key)[1]:  # noqa
                    return True
            else:
                self.monitoring_dict[this_key] = 0
        self.prev_value_dict = current_values_dict
        return stop_flag

    def optimize(self):
        pass

    def get_metric_function(self, metrics):
        if metrics == "ClassificationMetrics":
            from deepnumpy.metrics.classification_metrics import ClassificationMetrics  # noqa
        else:
            logger.error("Selected Loss function is not implemented."
                         "Taking default loss function 'ClassificationMetrics'")  # noqa
            metrics == 'ClassificationMetrics'
            from deepnumpy.metrics.classification_metrics import ClassificationMetrics  # noqa
        return eval(metrics)

    def evaluate(self, x_test, y_actual,
                 metrics="ClassificationMetrics", output_dir='./output'):
        y_pred = self.predict(x_test)
        test_loss = self.loss_func_instance(y_pred, y_actual)
        test_loss_value = test_loss.forward()
        logger.info(f"Validation Loss: {test_loss_value}")
        metric_instance = self.get_metric_function(metrics)(output_dir)
        metric_instance.run(y_pred, y_actual)
        return test_loss_value

    def plot_loss_curve(self, output_path):
        plt.figure(figsize=(17, 10))
        plt.title('Loss Curve : Loss vs Epochs', fontsize=20)
        plt.xlabel('Epoch', fontsize=16)
        plt.ylabel('Loss Value', fontsize=16)
        plt.plot(self.train_loss_idx, self.train_loss, "-g", label="Train Loss")  # noqa
        plt.plot(self.test_loss_idx, self.test_loss, "-r", label="Test Loss")
        plt.legend(loc="upper right")
        plt.grid()
        plt.savefig(os.path.join(output_path, 'TrainTestLossCurve.png'))
        plt.close()

max_stars_repo_name: RajuSaladi/DeepLearningWithNumpy

max_issues_repo_name: RajuSaladi/DeepLearningWithNumpy

max_forks_repo_name: RajuSaladi/DeepLearningWithNumpy


text:
"""Functions to calculate and plot standard EEG electrode position systems."""
# Copyright (c) 2018-2021, Stefan Appelhoff
# BSD-3-Clause

import numpy as np
import pandas as pd

def find_point_at_fraction(p1, p2, p3, frac):
    """Find a point on an arc spanned by three points.

    Given three points `p1`, `p2` and `p3` on a sphere with origin (0, 0, 0),
    find the coordinates of a `point` at a fraction `frac` of the overall
    distance on an arc spanning from `p1` over `p2` to `p3`. Given this
    assumption, for fractions of zero, `point` will equal `p1`; for fractions
    of one, `point` will equal `p3`; and for fractions of one half, `point`
    will equal `p2` [1]_.

    Parameters
    ----------
    p1, p2, p3 : tuple
        Each tuple containing x, y, z cartesian coordinates.
    frac : float
        Fraction of distance from `p1` to `p3` over p2` at which
        to find coordinates of `point`.

    Returns
    -------
    point : tuple
        The x, y, z cartesian coordinates of the point at fraction.

    Notes
    -----
    The assumptions of this function require `p1`, `p2` and `p3` to be
    equidistant from the origin. They must not be collinear (i.e., all be
    along one line) and none of the points may be equal.

    References
    ----------
    .. [1] Nominal Animal
       (https://math.stackexchange.com/users/318422/nominal-animal),
       find intermediate points on small circle of a sphere,
       URL (version: 2018-06-02): https://math.stackexchange.com/q/2805204

    Examples
    --------
    >>> p1 = (1., 0., 0.)
    >>> p2 = (0., 0., 1.)
    >>> p3 = (-1., 0., 0.)
    >>> find_point_at_fraction(p1, p2, p3, frac=0.)
    (1.0, 0.0, 0.0)
    >>> find_point_at_fraction(p1, p2, p3, frac=.5)
    (0.0, 0.0, 1.0)
    >>> find_point_at_fraction(p1, p2, p3, frac=1.)
    (-1.0, 0.0, 0.0)
    >>> find_point_at_fraction(p1, p2, p3, frac=.3)
    (0.5878, 0.0, 0.809)

    """
    # Unpack the point tuples
    x1, y1, z1 = p1
    x2, y2, z2 = p2
    x3, y3, z3 = p3

    # Using the cross product, find unit normal vector (xn, yn, zn)
    # of the plane spanning the three points:
    x12 = x2 - x1
    y12 = y2 - y1
    z12 = z2 - z1
    x13 = x3 - x1
    y13 = y3 - y1
    z13 = z3 - z1

    xn = y12 * z13 - z12 * y13
    yn = z12 * x13 - x12 * z13
    zn = x12 * y13 - y12 * x13

    n = np.sqrt(xn ** 2 + yn ** 2 + zn ** 2)

    if n <= 0.0:
        raise ValueError("Points are either collinear or share the same coordinates.")

    xn = xn / n
    yn = yn / n
    zn = zn / n

    # Use dot product to get signed distance from plane to origin
    # interchangeably use p1, p2, or p3 with normal vector
    d = xn * x1 + yn * y1 + zn * z1

    # At intersection of sphere and plane, we have a circle
    # with the following center:
    xc = d * xn
    yc = d * yn
    zc = d * zn

    # Construct a 2D coordinate system with the unit circle corresponding
    # to the above circle with first axis towards p1 and second towards p2:
    # 2D U axis unit vector
    xu = x1 - xc
    yu = y1 - yc
    zu = z1 - zc

    # 2D V axis unit vector
    xv = yn * zu - zn * yu
    yv = zn * xu - xn * zu
    zv = xn * yu - yn * xu

    # Choose V axis towards (x2, y2, z2)
    v2 = (x2 - xc) * xv + (y2 - yc) * yv + (z2 - zc) * zv
    if v2 < 0.0:
        xv = -xv
        yv = -yv
        zv = -zv

    # Find theta, the positive plane angle towards p3 (x3, y3, z3).
    xt = x3 - xc
    yt = y3 - yc
    zt = z3 - zc

    thetau = xt * xu + yt * yu + zt * zu
    thetav = xt * xv + yt * yv + zt * zv
    theta = np.arctan2(thetav, thetau)

    if theta < 0.0:
        # Add 360 degrees, or 2*Pi in radians, to make it positive
        theta = theta + 2 * np.pi

    # Now calculate coordinates at fraction
    x = xc + xu * np.cos(frac * theta) + xv * np.sin(frac * theta)
    y = yc + yu * np.cos(frac * theta) + yv * np.sin(frac * theta)
    z = zc + zu * np.cos(frac * theta) + zv * np.sin(frac * theta)

    # Round to 4 decimals and collect the points in tuple
    point = np.asarray((x, y, z))
    point = tuple(point.round(decimals=4))
    return point

def _add_points_along_contour(df, contour):
    """Compute points along `contour` and add them to `df`.

    Parameters
    ----------
    df : pandas.DataFrame
        The data with columns ["label", "x", "y", "z"].
    contour : list of str
        Each entry in `contour` is the label of a point, and
        all points in `contour` are ordered. Must be of length
        17 or 21.

    Returns
    -------
    df : pandas.DataFrame
        The data with columns ["label", "x", "y", "z"] and
        new points (rows in `df`) along `contour` added.

    """
    contour_len = len(contour)
    if contour_len == 21:
        midpoint_idx = 10
    elif contour_len == 17:
        midpoint_idx = 8
    else:
        raise ValueError(f"contour must be of len 17 or 21 but is {contour_len}")

    # Get the reference points from data frame
    p1 = _get_xyz(df, contour[0])
    p2 = _get_xyz(df, contour[midpoint_idx])
    p3 = _get_xyz(df, contour[-1])

    # Calculate all other points at fractions of distance
    other_ps = {}
    for i, label in enumerate(contour):
        other_ps[label] = find_point_at_fraction(
            p1, p2, p3, frac=i / (len(contour) - 1)
        )

    # Append to data frame
    df = _append_ps_to_df(df, other_ps)
    return df

def _append_ps_to_df(df, ps):
    """Add points `ps` to data frame `df`.

    This function removes duplicates from `df`, always keeping
    the first entry it finds.

    Parameters
    ----------
    df : pandas.DataFrame
        The data with columns ["label", "x", "y", "z"].
    ps : dict
        Each key is a label, with a tuple value of (x, y, z)
        coordinates.

    Returns
    -------
    df : pandas.DataFrame
        The data with columns ["label", "x", "y", "z"],
        and `ps` added.

    """
    tmp = pd.DataFrame.from_dict(ps, orient="index")
    tmp.columns = ["x", "y", "z"]
    tmp["label"] = tmp.index
    df = df.append(tmp, ignore_index=True, sort=True)

    # Remove duplicates, keeping the first computations
    df = df.drop_duplicates(subset="label", keep="first")
    return df

# Convenient helper function to access xyz coordinates from df
def _get_xyz(df, label):
    """Get xyz coordinates from a pandas data frame.

    Parameters
    ----------
    df : pandas.DataFrame
        Data frame with (at least) columns x, y, z, label.
    label : str
        Electrode label for which to get x, y, z from `df`.

    Returns
    -------
    x, y, z : float
        Positions of electrodes on a unit sphere.

    """
    # Check that all labels are present
    for var in ["label", "x", "y", "z"]:
        if var not in df.columns:
            raise ValueError(f"df must contain a column '{var}'")

    # Check we get exactly one row of data
    subdf = df[df["label"] == label]
    nrows = subdf.shape[0]
    if nrows == 0 or nrows > 1:
        raise ValueError(f"Expected one row of data but got: {nrows}")

    # Get the data
    x = float(df[df["label"] == label].x)
    y = float(df[df["label"] == label].y)
    z = float(df[df["label"] == label].z)
    return x, y, z

def _get_coords_on_circle(cx=0, cy=0, r=1, steps=180 / 20):
    """Get the cartesian coordinates [x,y] for a number of points on a circle.

    Assume that top of circle is 0 degrees.

    Parameters
    ----------
    cx, cy : int
        Coordinates (x and y) of origin of circle. Defaults
        to 0 for x and y.
    r : int
        Radius of circle, defaults to 1.
    steps : int
        Spacing between evenly spaced points on the
        circle in degrees. Defaults to 9 (360 degrees
        divided into 40 parts, i.e., 5 percent parts)

    Returns
    -------
    coords : list of list
        Nested list of points: ``[[x1, y1], [x2, y2], ...]``.

    """
    # Make sure we are dealing with integer steps
    assert steps / int(steps) == 1.0
    steps = int(steps)

    coords = []
    for a in np.arange(0, 360, steps):
        x = cx + r * np.cos(np.deg2rad(a))
        y = cy + r * np.sin(np.deg2rad(a))
        coords.append([x, y])

    # Exchange x and y so that 0 degree is top of circle
    # also round off
    coords = [[round(y, 5), round(x, 5)] for x, y in coords]  # noqa

    return coords

def _stereographic_projection(x, y, z, scale=1.0):
    """Calculate the stereographic projection.

    Given a unit sphere with radius ``r = 1`` and center at
    The origin. Project the point ``p = (x, y, z)`` from the
    sphere's South pole ``(0, 0, -1)`` onto a tangent plane
    on the sphere's North pole ``(0, 0, 1)``. The resulting
    point is ``p' = (x', y')``.

    ``x', y' = (1 / (x + z)), (1 / (y + z))``

    Parameters
    ----------
    x, y, z : float
        Positions of electrodes on a unit sphere
    scale : float
        Determines the distance of the projection point
        from the origin of the sphere in terms of the radius.
        Defaults to 1.0, which is a point on the sphere.

    Returns
    -------
    x, y : float
        Positions of electrodes as projected onto a unit circle.

    """
    mu = 1.0 / (scale + z)
    x = x * mu
    y = y * mu
    return np.asarray(x), np.asarray(y)

ext: py

reason: import numpy


text:
import json
import warnings
from itertools import combinations, groupby
from typing import Dict, Iterable, List, Tuple

import numpy as np
import spacy
from spacy.language import Language
from spacy.tokens import Doc
from spacy.training import Example

Annotations = List[Tuple[int, int, str]]
AnnotationsDict = Dict[str, Iterable[int]]

class TaggedCorpus:
    def __init__(
        self,
        text: str,
        annotations: Annotations,
        tokenizer: spacy.tokenizer.Tokenizer = None,
    ) -> None:
        self.text = text
        self.annotations = annotations
        if tokenizer is None:
            nlp = spacy.blank("en")
            # add more rules for tokenizing
            infixes = nlp.Defaults.infixes + [r"[-/=)(><]"]
            infix_regex = spacy.util.compile_infix_regex(infixes)
            nlp.tokenizer.infix_finditer = infix_regex.finditer
            self.tokenizer = nlp.tokenizer
        else:
            self.tokenizer = tokenizer
        # additional properties
        self.arr_char_spans, self.tokens = self._get_char_spans_and_tokens(
            text=text, tokenizer=self.tokenizer
        )
        self.aligned_annotations_dict = self._align_span_dict(
            annotations=annotations, text=text, tokenizer=self.tokenizer
        )

    @staticmethod
    def _annotations2dict(annotations: Annotations) -> AnnotationsDict:
        """
        Go from annotations list to a dictionary
        :param annotations: Entity annotations [(start, stop, entity), ...]
        :return: Dictionary of annotations {entity: [(start, stop), ...]}
        """
        s = sorted(annotations, key=lambda a: a[-1])
        grp = groupby(s, key=lambda a: a[-1])
        return {key: [val[:-1] for val in vals] for key, vals in grp}

    @staticmethod
    def _dict2annotations(annotations_dict: AnnotationsDict) -> Annotations:
        """
        Go from annotations dictionary to a list
        :param annotations_dict: Dictionary of annotations {entity: [(start, stop), ...]}
        :return: Dictionary of annotations {entity: [(start, stop), ...]}
        """
        return [
            (span[0], span[1], tag)
            for tag, spans in annotations_dict.items()
            for span in spans
        ]

    def _get_char_spans_and_tokens(
        self, text: str, tokenizer: spacy.tokenizer.Tokenizer
    ) -> (np.ndarray, List[str]):
        """
        Get the character spans and the text for each token in the input text.
        :param text: Input text
        :return: 2 components - numpy 2D array - rows are tokens, columns are start, stop character span; list of tokens
        """
        doc = tokenizer(text)
        arr = doc.to_array(["IDX", "LENGTH"])
        tokens = [token.text for token in doc]
        return np.array([arr[:, 0], arr[:, 0] + arr[:, -1]]).T, tokens

    def _align_span_dict(
        self, annotations: Annotations, text: str, tokenizer: spacy.tokenizer.Tokenizer
    ) -> Dict[str, Dict[str, List[str]]]:
        """
        Get the aligned character spans - spans start at first character index and end on last characted index + 1.
        For example, in "test-token.", `token` spans from (5,10) - as index starts at zero.
        :param annotations: Annotation list
        :param text: Input text
        :return: Dictionary of dictionaries - for each entity get the list of character spans and token sequences
        (example: {entity: {char_spans: [(0, 6), ...], token_seq: [ENT_1, ...]})
        """

        def _consecutive(arr: np.ndarray) -> List[Tuple[int, ...]]:
            """Get tuples of the consecutive integers in 1D array"""
            idx_splits = (np.diff(arr) != 1).nonzero()[0] + 1
            arr_consec = np.split(ary=arr, indices_or_sections=idx_splits)
            return [tuple(a) for a in arr_consec]

        def _stich_spans(
            arr_spans: np.ndarray, tk_spans: np.ndarray
        ) -> List[Tuple[int, int]]:
            """Stich back together the spans of consecutive tokens."""
            first_and_last_idx = [
                (grp[0], grp[-1]) for grp in _consecutive(arr_spans.nonzero()[0])
            ]
            return [
                (token_spans[start, 0], token_spans[end, -1])
                for start, end in first_and_last_idx
            ]

        d_arr = {}
        annotations_dict = self._annotations2dict(annotations=annotations)
        token_spans, _ = self._get_char_spans_and_tokens(text=text, tokenizer=tokenizer)
        for tag, spans in annotations_dict.items():
            # span of entity must include start and end characters of token to be tagged
            is_span = tuple(
                ((token_spans[:, 0] >= span[0]) & (token_spans[:, 1] <= span[1]))
                for span in spans
            )
            # gather all the tokens that meet criteria
            is_tag = np.any(is_span, axis=0)
            arr_tagged_tokens = np.where(is_tag, tag, "O")
            # for subsequent tokens with the same entity, we want to join the spans
            aligned_char_span = (
                _stich_spans(arr_spans=is_tag, tk_spans=token_spans)
                if any(is_tag)
                else []
            )
            d_arr[tag] = {
                "char_spans": aligned_char_span,
                "token_seq": arr_tagged_tokens.tolist(),
            }
        return d_arr

    def to_tokens(self) -> List[str]:
        """Get the token list."""
        return self.tokens

    def to_token_char_spans(self) -> List[Tuple[int, int]]:
        """Get the character spans for each token."""
        return [(e[0], e[1]) for e in self.arr_char_spans.tolist()]

    def to_dict(
        self,
    ) -> Dict[str, List[str]]:
        """
        Get the entities in a dictionary.
        :return: Dictionary with token sequences (ex: {entity_1: [ENT1, ...], entity_2: [ENT2, ...], ...})
        """
        d = {k: v["token_seq"] for k, v in self.aligned_annotations_dict.items()}
        return d

    def to_doc(self, entity: str = None) -> spacy.tokens.Doc:
        """
        Get the entities in an annotated spaCy document.
        :param entity: Optional parameter - if set, return document annotated with only selected entity
        :return: SpaCy document annotated with entities
        """
        annotations_dict = {
            k: [(e[0], e[1]) for e in v["char_spans"]]
            for k, v in self.aligned_annotations_dict.items()
        }
        # get the character span entity annotations
        if entity is None:
            aligned_annotations = self._dict2annotations(
                annotations_dict=annotations_dict
            )
            annotations = rm_colliding(aligned_annotations)
            if sorted(annotations) != sorted(aligned_annotations):
                warnings.warn("Found overlapping spans. Keep only longest span.")
        else:
            annotations = [
                (span[0], span[1], entity) for span in annotations_dict[entity]
            ]

        # generate a blank spacy document and spans for the entity annotations
        doc = self.tokenizer(self.text)
        entity_spans = [
            doc.char_span(start, end, label=entity)
            for start, end, entity in annotations
        ]
        assert (
            None not in entity_spans
        ), f"Token with annotation {annotations[entity_spans.index(None)]} don't match token spans."
        # annotate document with entity spans
        doc.set_ents(entity_spans)
        return doc

    def to_iob_list(self, entity: str = None) -> List[str]:
        """
        Get the entities in a list of entities (IOB format).
        :param entity: Optional parameter - if set, return document annotated with only selected entity
        :return: List of strings where each string represents the entity associated with the token
        """
        doc = self.to_doc(entity=entity)
        return doc2ents(doc)

    def to_multi_iob_list(self) -> (List[Tuple[str, ...]], List[str]):
        """
        Get all the entities in a list of entities (IOB format).
        :return: :return: List of tuples where each tuple contains one element for each entity present in the document
        """
        all_entities = [
            self.to_iob_list(entity=entity)
            for entity in self.aligned_annotations_dict.keys()
        ]
        return list(zip(*all_entities)), list(self.aligned_annotations_dict.keys())

def get_entities(s: str) -> Annotations:
    """Process annotations to get in the format of (start, stop, label)"""
    d = json.loads(s)

    # check if dictionary is not empty
    if not d:
        return None

    assert isinstance(d["objects"], list), "Couldn't find `objects` list."
    return [
        (el["data"]["location"]["start"], el["data"]["location"]["end"], el["value"])
        for el in d["objects"]
    ]

def rm_groups(a: Annotations) -> Annotations:
    """Remove the group number info - ie: `g2_response_rate` becomes `response_rate`"""

    def _rm_grp_info(s: str) -> str:
        if s.startswith("g1_") or s.startswith("g2_"):
            return s[3:]
        elif s in ["group1", "group2"]:
            return "group"
        else:
            return s

    return list({(start, stop, _rm_grp_info(ent)) for start, stop, ent in a})

def rm_colliding(x: Annotations) -> Annotations:
    """Remove any duplicated annotations (one token cannot have two entities)"""

    def _is_overlap(a: Tuple[int, int], b: Tuple[int, int]) -> bool:
        return a[0] < b[1] and b[0] < a[1]

    def _span_to_drop(a: Tuple[int, int], b: Tuple[int, int]) -> Tuple[int, int]:
        return a if (a[1] - a[0]) < (b[1] - b[0]) else b

    def _rm_overlap(ranges: Iterable[Tuple[int, int]]):
        not_valid = [
            _span_to_drop(a, b) for a, b in combinations(ranges, 2) if _is_overlap(a, b)
        ]
        valid = [el for el in ranges if el not in not_valid]

        # check for other overlaps (in case of 3-way collisions)
        if not_valid:
            return _rm_overlap(valid)
        else:
            return valid

    # if duplicated spans then take first
    d = {(start, stop): ent for start, stop, ent in x[::-1]}
    valid_spans = _rm_overlap(d.keys())
    return [
        (start, stop, ent)
        for (start, stop), ent in d.items()
        if (start, stop) in valid_spans
    ]

def doc2ents(doc: Doc) -> List[str]:
    """Take a spaCy `Doc` and return a list of IOB token-based entities"""
    return [
        f"{token.ent_iob_}-{token.ent_type_}"
        if token.ent_type_
        else token.ent_iob_
        if token.ent_iob_
        else "O"
        for token in doc
    ]

def generate_examples(
    texts: Iterable[str], entity_offsets: Iterable[Annotations], nlp: Language
) -> List[Example]:
    if len(texts) != len(entity_offsets):
        raise ValueError("Lengths don't match.")
    # get docs (with predictions)
    docs = [nlp.make_doc(text) for text in texts]
    examples = [
        Example.from_dict(doc, {"entities": entity_offsets})
        for doc, entity_offsets in zip(docs, entity_offsets)
    ]
    return examples

max_stars_repo_name: akashphaniteja/Bio-BERT

max_issues_repo_name: akashphaniteja/Bio-BERT

max_forks_repo_name: akashphaniteja/Bio-BERT


text:
"""
    本代码用于: 观察实验结果的损失值、重构损失、KLD 值曲线图
    创建时间: 2021 年 9 月 2 日
    创建人: MorningStar
    最后一次修改时间: 2021 年 9 月 2 日
    具体步骤:
        step1: 绘制损失值图像
        step2: 绘制重构损失值图像
        step3: 绘制 KLD 值图像
        step4: 绘制三条曲线在一张图的图像
"""
# -------------------- 导入必要的包 -------------------- #
# ----- 系统操作相关 ----- #
import time
import os

# ----- 文件读取相关 ----- #
import csv

# ----- 数据处理相关 ----- #
import numpy as np
import scipy.io as scio

# ----- 绘制图片相关 ----- #
import matplotlib
import matplotlib.pyplot as plt

# ----- 创建字体对象 ----- #
from matplotlib.font_manager import FontProperties
font = FontProperties(fname='SimHei.ttf', size=16)

# -------------------- 设置常量参数 -------------------- #
# ----- 读取文件相关 ----- #
PATH_LOSS = "logs/VanillaVAE/" + "version_11/" + "metrics.csv"

# ----- 保存文件相关 ----- #
PATH_SAVE_PIC = "output_qxy/VanillaVAE/" + "version_11/"

# -------------------- 函数实现 -------------------- #

# -------------------- 主函数运行 -------------------- #
if __name__ == '__main__':
    # ----- 开始计时 ----- #
    T_Start = time.time()
    print("程序开始运行 !")

    # ---------- step0: 绘图前的准备 ---------- #
    # ----- 读取文件 ----- #
    csv_reader = csv.reader(open(PATH_LOSS))

    # ----- 保存各个 loss ----- #
    list_loss = []
    list_recon = []
    list_kld = []
    list_iter = []
    i = 0
    j = 0
    for line in csv_reader:
        # ----- 获取各个值 ----- #
        temp_epoch = line[4]

        i += 1

        # ----- 记录数据 ----- #
        if i > 4 and temp_epoch == '':

            # 获取他们的值
            temp_loss = float(line[0])
            temp_recon_loss = float(line[1])
            temp_kld = float(line[2])

            # print(temp_loss)
            list_loss.append(temp_loss)
            list_recon.append(temp_recon_loss)
            list_kld.append(temp_kld)
            list_iter.append(j)

            j += 1

    # print(list_loss)

    # ----- 创建保存图片的路径 ----- #
    if not os.path.exists(PATH_SAVE_PIC):
        os.makedirs(PATH_SAVE_PIC)

    # ---------- step1: 绘制损失值图像 ---------- #
    # 暂不画

    # ---------- step2: 绘制重构损失值图像 ---------- #
    # 暂不画

    # ---------- step3: 绘制 KLD 值图像 ---------- #
    plt.figure()
    plt.plot(list_kld[0:3000], color='skyblue', label="KLD")

    plt.xlabel("iter", fontproperties=font)
    plt.ylabel("kld 值", fontproperties=font)
    plt.title("KLD Analysis", fontproperties=font)
    plt.legend() # 显示图例
    plt.savefig(os.path.join(PATH_SAVE_PIC, "KLD Analysis.png"))

    # ---------- step4: 绘制三条曲线在一张图的图像 ---------- #
    plt.figure()
    plt.plot(list_loss[0:3000], color='red', label="loss")
    plt.plot(list_recon[0:3000], color='green', label="recon loss")
    # plt.plot(list_iter, list_kld, color='skyblue', label="KLD/10")

    plt.xlabel("iter", fontproperties=font)
    plt.ylabel("loss", fontproperties=font)
    plt.title("Result Analysis", fontproperties=font)
    plt.legend() # 显示图例
    plt.savefig(os.path.join(PATH_SAVE_PIC, "Result Analysis.png"))
    # plt.show()

    # ----- 结束计时 ----- #
    T_End = time.time()
    T_Sum = T_End  - T_Start
    T_Hour = int(T_Sum/3600)
    T_Minute = int((T_Sum%3600)/60)
    T_Second = round((T_Sum%3600)%60, 2)
    print("程序运行时间: {}时{}分{}秒".format(T_Hour, T_Minute, T_Second))

    print("程序已结束 ！")

max_stars_repo_name: MorningStarOvO/VAE-Watch-Experiment

max_issues_repo_name: MorningStarOvO/VAE-Watch-Experiment

max_forks_repo_name: MorningStarOvO/VAE-Watch-Experiment

reason: import numpy,import scipy


text:
import sys
import time

import matplotlib.pyplot as plt
import numpy as np
import numpy.random as rnd
import pandas as pd

import GPflow

sys.path.append('..')
import opt_tools as ot

rnd.seed(4)

X = np.linspace(0, 5, 100)[:, None]
Y = 0.3 * np.sin(2 * X) + 0.05 * rnd.randn(*X.shape)

# model = GPflow.sgpr.SGPR(X, Y, GPflow.kernels.RBF(1), X[rnd.permutation(len(X))[:3], :])
model = GPflow.sgpr.SGPR(X, Y, GPflow.kernels.RBF(1), X[:3, :].copy())
model._compile()

optlog = ot.GPflowOptimisationHelper(
    model,
    [
        ot.tasks.DisplayOptimisation(ot.seq_exp_lin(1.0, 1.0)),
        ot.tasks.GPflowLogOptimisation(ot.seq_exp_lin(1.0, 1.0), store_fullg=True, store_x=True),
        ot.tasks.StoreOptimisationHistory('./opthist.pkl', ot.seq_exp_lin(1.0, np.inf, 5.0, 5.0), verbose=True)
    ]
)
optlog.callback(model.get_free_state())

# Start optimisation
model.optimize(callback=optlog.callback, disp=False, maxiter=10)
optlog.finish(model.get_free_state())
print("Finished first optimisation run. %i iterations." % optlog.hist.i.max())
print("")

# Resume optimisation
hist = pd.read_pickle('./opthist.pkl')
time.sleep(3.0)
model._wrapped_objective = model._objective

def obj(x):
    time.sleep(0.2)
    return model._wrapped_objective(x)

model._objective = obj
optlog = ot.GPflowOptimisationHelper(
    model,
    [
        ot.tasks.DisplayOptimisation(ot.seq_exp_lin(1.0, 1.0)),
        ot.tasks.GPflowLogOptimisation(ot.seq_exp_lin(1.0, 1.0), store_fullg=True, store_x=True, hist=hist),
        ot.tasks.StoreOptimisationHistory('./opthist.pkl', ot.seq_exp_lin(1.0, np.inf, 5.0, 5.0), verbose=True),
        ot.tasks.Timeout(5.0)
    ]
)
try:
    model.optimize(callback=optlog.callback, disp=False)
except ot.OptimisationTimeout:
    print("Optimisation timeout...")
except KeyboardInterrupt:
    print("Cancelled by user...")
    optlog.finish(model.get_free_state())

print("Plotting...")
hist = pd.read_pickle('./opthist.pkl')

plt.figure(figsize=(14, 5))
plt.subplot(131)
plt.plot(hist.i, hist.f, '-x')
plt.xscale('log')

plt.subplot(132)
plt.plot(hist.i, hist.gnorm, '-x')
plt.xscale('log')
plt.yscale('log')

plt.subplot(133)
plt.plot(X, Y, 'x')
pY, pYv = model.predict_y(X)
plt.plot(X, pY)
plt.plot(X, pY + pYv ** 0.5 * 2, 'r')
plt.plot(X, pY - pYv ** 0.5 * 2, 'r')

plt.show()

max_stars_repo_name: markvdw/convgp

max_issues_repo_name: markvdw/convgp

max_forks_repo_name: markvdw/convgp


text:
import numpy as np
import pytest

from structure_factor.hyperuniformity import Hyperuniformity
from structure_factor.point_processes import (
    GinibrePointProcess,
    HomogeneousPoissonPointProcess,
)

@pytest.mark.parametrize(
    "sf, expected",
    [
        (HomogeneousPoissonPointProcess.structure_factor, False),
        (GinibrePointProcess.structure_factor, True),
    ],
)
def test_effective_hyperuniformity(sf, expected):
    # verify that the hyperuniformity index for the ginibre ensemble is less than 1e-3
    k = np.linspace(0, 10, 100)
    sf_k = sf(k)
    hyperuniformity_test = Hyperuniformity(k, sf_k)
    index_H, _ = hyperuniformity_test.effective_hyperuniformity(k_norm_stop=4)
    result = index_H < 1e-3
    assert result == expected

def f(c, alpha, x):
    return c * x ** alpha

x_1 = np.linspace(0, 3, 100)
x_2 = np.linspace(0.5, 2, 50)

@pytest.mark.parametrize(
    "x, fx, c, alpha",
    [
        (x_1, f(8, 2, x_1), 8, 2),
        (x_2, f(6, 0.5, x_2), 6, 0.5),
    ],
)
def test_hyperuniformity_class_on_polynomial(x, fx, c, alpha):
    test = Hyperuniformity(x, fx)
    assert alpha, c == test.hyperuniformity_class()

@pytest.mark.parametrize(
    "sf, expected_alpha",
    [
        (GinibrePointProcess.structure_factor, 2),
    ],
)
def test_hyperuniformity_class_ginibre(sf, expected_alpha):
    # verify that the hyperuniformity index for the ginibre ensemble is less than 1e-3
    k = np.linspace(0, 1, 3000)
    sf_k = sf(k)
    hyperuniformity_test = Hyperuniformity(k, sf_k)
    alpha, _ = hyperuniformity_test.hyperuniformity_class(k_norm_stop=0.001)
    diff_alpha = alpha - expected_alpha
    np.testing.assert_almost_equal(diff_alpha, 0, decimal=3)

max_stars_repo_name: For-a-few-DPPs-more/structure-factor

max_issues_repo_name: For-a-few-DPPs-more/structure-factor

max_forks_repo_name: For-a-few-DPPs-more/structure-factor


text:
from sklearn import datasets
iris = datasets.load_iris()  # loading dataset

import pandas as pd
X = pd.DataFrame(iris.data) # initializing rows and columns
X.columns = ['Sepal_Length','Sepal_Width','Petal_Length','Petal_Width']
y = pd.DataFrame(iris.target)
y.columns = ['Targets']

from sklearn.cluster import KMeans
model = KMeans(n_clusters=3) # clustering data
model.fit(X)
model.labels_

import numpy as np
colormap = np.array(['red', 'lime', 'black'])

import matplotlib.pyplot as plt
plt.figure(figsize=(14,7))   # ploting dataset and comparing Kmeans and EM
plt.subplot(1, 2, 1)
plt.scatter(X.Petal_Length, X.Petal_Width, c=colormap[y.Targets], s=40)
plt.title('EM Clustering')
plt.subplot(1, 2, 2)
plt.scatter(X.Petal_Length, X.Petal_Width, c=colormap[model.labels_], s=40)
plt.title('K-Means clustering')

import sklearn.metrics as sm
acc = sm.accuracy_score( y ,model.labels_) # calculating accuracy
print(acc * 100)


text:
import numpy as np
from .hp import IMAGE_SIZE, BYTE
from PIL import Image
from typing import List

def image_router(*images, ratio: int, key: int=None):
    """Just a simple encapsulation, logic is described in the core file."""
    if ratio <= 0 or ratio >= BYTE:
        raise Exception("Error: Excepted values for ratio are in the range [1,8)")
    if key:
        if len(images) == 1:
            return _decode_image_with_key(images[0], key=key, ratio=ratio)
        return _encode_images_with_key(images, key=key, ratio=ratio)
    # temp
    raise Exception("Error: No key methods are not useful and will be removed.")
    if len(images) == 1:
        return _decode_image_no_key(images[0])
    return _encode_images_no_key(images)

def _encode_images_with_key(org_images: List[np.ndarray],
                            key: int,
                            ratio: int) -> Image.Image:
    """Takes a list of images and combines them. Every image
    except the first one is hid into the first image.

    The hiding is done by storing the 4 most significant bits of
    the hidden images in the randomly chosen places of first
    image's least significant bits.

    Parameters
    ----------
    images : list
        A list of images to be combined. First is the "cover", the
        rest is hidden.
    key : int
        The key is needed for decoding the image later.
    ratio : int
        Tells how many bits are from the cover image.

    Returns
    -------
    PIL.Image.Image
        A cover image with hidden images in it.
    """
    # the function works with the flattened data
    images = [im.reshape(np.prod(im.shape)).copy() for im in org_images]
    # the cover image
    data = images[0]
    # generating the indexes using key
    np.random.seed(key)
    final_idxs = []
    full_idxs = []
    for _ in range(len(images)-1):
        availables = np.setdiff1d(np.arange(data.shape[0]), full_idxs)
        idxs = np.random.choice(availables, np.prod((*IMAGE_SIZE, 3)), replace=False)
        full_idxs = [*full_idxs, *idxs]
        final_idxs.append(idxs)
    # making the magic happen using bit shifting operators
    for i in range(len(images)-1):
        data[final_idxs[i]] = (data[final_idxs[i]] >> (BYTE-ratio) << (BYTE-ratio)) + \
                              (images[i+1] >> ratio)
    return Image.fromarray(data.reshape(org_images[0].shape))

def _decode_image_with_key(image: np.ndarray,
                           key: int,
                           ratio: int) -> List[Image.Image]:
    """Takes an image with hidden images in it and returns
    a list of hidden images.

    Parameters
    ----------
    images : list
        A list of images to be combined. First is the "cover", the
        rest is hidden.
    key : int
        The key used when encoding the data.
    ratio : int
        Tells how many bits are from the cover image.

    Returns
    -------
    list[PIL.Image.Image]
        List of extracted images.
    """
    # determining the max number of hiddens
    num_hidden = np.prod(image.shape) // np.prod((*IMAGE_SIZE, 3))
    # the function works with the flattened data
    data = image.reshape(np.prod(image.shape)).copy()
    # generating the indexes using key
    np.random.seed(key)
    final_idxs = []
    full_idxs = []
    for _ in range(num_hidden):
        availables = np.setdiff1d(np.arange(data.shape[0]), full_idxs)
        idxs = np.random.choice(availables, np.prod((*IMAGE_SIZE, 3)), replace=False)
        full_idxs = [*full_idxs, *idxs]
        final_idxs.append(idxs)
    # getting hidden images from the right indexes using bitshifting
    images = []
    for i in range(num_hidden):
        images.append(Image.fromarray((data[final_idxs[i]] << ratio).reshape((*IMAGE_SIZE, 3))))
    return images

def _encode_images_no_key(images: List[np.ndarray]) -> np.ndarray:
    """Takes a list of images and combines them. Every image
    is combined into an image that looks like the first image.

    The combination is done by combining as much significant
        bits as possible per image.

    If two images -> combines their 4 most significant digits.
    If three images -> combines 3 most from the first two images
        and 2 most from the last image.
    If four images -> combines their 2 most significant digits.

    Parameters
    ----------
    images : list
        A list of images to be combined.

    Returns
    -------
    list[numpy.ndarray]
        a list of images in the form of numpy arrays.
    """
    if len(images) == 4:
        return sum([image >> 6 << i for (image, i) in zip(images, range(6, -1, -2))])
    if len(images) == 3:
        return (images[0] >> 5 << 5) + (images[1] >> 5 << 2) + (images[2] >> 6)
    if len(images) == 2:
        return (images[0] >> 4 << 4) + (images[1] >> 4)
    raise Exception("Error: Number of images more than 4 is not supported.")

def _decode_image_no_key(combined_image: np.ndarray,
                         num_hidden: int) -> List[Image.Image]:
    """Extracts and returns the `num_hidden` images from the given image.

    Parameters
    ----------
    combined_image : numpy.ndarray
        An image represented by a numpy array.
    num_hidden : int
        Total number of images to be extracted.

    Returns
    -------
    list[PIL.Image.Image]
        a list of extracted images.
    """
    if num_hidden == 4:
        return [Image.fromarray(combined_image >> i << 6) for i in range(6, -1, -2)]
    if num_hidden == 3:
        return [Image.fromarray(im) for im in [(combined_image >> 5 << 5),
                                               (combined_image >> 2 << 5),
                                               (combined_image << 6)]]
    if num_hidden == 2:
        return [Image.fromarray(combined_image >> 4 << 4), Image.fromarray(combined_image << 4)]
    raise Exception("Range of the number of images is [2, 4].")

max_stars_repo_name: cihatceliker/ImageSteganography

max_issues_repo_name: cihatceliker/hider

max_forks_repo_name: cihatceliker/hider


text:
from easyocr.recognition import get_recognizer, AlignCollate
from easyocr.utils import get_image_list
import torch
from PIL import Image
import numpy as np
import torch.nn.functional as F
from torch import nn
import cv2
import onnx
import onnxruntime

def to_numpy(tensor):
    if tensor.requires_grad:
        return tensor.detach().cpu().numpy()
    return tensor.cpu().numpy()

def custom_mean(x):
    return x.prod()**(2.0/np.sqrt(len(x)))

network_params = {
    'input_channel': 1,
    'output_channel': 256,
    'hidden_size': 256
}

recognizer, converter = get_recognizer(
    "generation2",
    network_params,
    "0123456789!\"#$%&'()*+,-./:;<=>?@[\]^_`{|}~ €ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz",
    {},
    {'en': 'easyocr/dict/en.txt'},
    "/Users/nitram/.EasyOCR//model/english_g2.pth",
)

imgH = 64
img_cv_grey = cv2.imread("examples/screen2.png", cv2.IMREAD_GRAYSCALE)
print(img_cv_grey.shape)  # (26, 293)

image_list, max_width = get_image_list(
    [[6, 254, 2, 26]], [], img_cv_grey, model_height=imgH)
image = image_list[0]

print(f"max_width: {max_width}")

process = AlignCollate(imgH=imgH, imgW=int(
    max_width), keep_ratio_with_pad=True)
image = process([Image.fromarray(image[1], 'L')])

recognizer.eval()
with torch.no_grad():
    print(image.shape)  # torch.Size([1, 1, 64, 704])
    preds = recognizer(image)
    print(preds.shape)  # torch.Size([1, 175, 97])

    # torch.onnx.export(
    #     model=recognizer,
    #     args=(image,),
    #     f="model.onnx",
    #     opset_version=12,
    #     input_names=['input'],
    #     output_names=['output'],
    #     dynamic_axes={
    #         'input': {2: "height", 3: "width"},
    #         'output': {1: "dim1", 2: "dim2"}
    #     })

    # verify exported onnx model
    recognizer_onnx = onnx.load("recognizer.onnx")
    onnx.checker.check_model(recognizer_onnx)
    # print(f"Model Inputs:\n {recognizer_onnx.graph.input}\n{'*'*80}")
    # print(f"Model Outputs:\n {recognizer_onnx.graph.output}\n{'*'*80}")

    ort_session = onnxruntime.InferenceSession("recognizer.onnx")

    # compute ONNX Runtime output prediction
    ort_inputs = {ort_session.get_inputs()[0].name: to_numpy(image)}
    preds_onnx = ort_session.run(None, ort_inputs)[0]
    preds = torch.tensor(preds_onnx)

    # Select max probabilty (greedy decoding) then decode index to character
    preds_size = torch.IntTensor([preds.size(1)] * 1)

    ######## filter ignore_char, rebalance
    preds_prob = F.softmax(preds, dim=2)
    preds_prob = preds_prob.cpu().detach().numpy()
    pred_norm = preds_prob.sum(axis=2)
    preds_prob = preds_prob/np.expand_dims(pred_norm, axis=-1)
    preds_prob = torch.from_numpy(preds_prob).float().to("cpu")

    _, preds_index = preds_prob.max(2)
    preds_index = preds_index.view(-1)
    preds_str = converter.decode_greedy(
        preds_index.data.cpu().detach().numpy(), preds_size.data)
    print(preds_str)

    preds_prob = preds_prob.cpu().detach().numpy()
    values = preds_prob.max(axis=2)
    indices = preds_prob.argmax(axis=2)
    preds_max_prob = []
    for v, i in zip(values, indices):
        max_probs = v[i != 0]
        if len(max_probs) > 0:
            preds_max_prob.append(max_probs)
        else:
            preds_max_prob.append(np.array([0]))

    for pred, pred_max_prob in zip(preds_str, preds_max_prob):
        confidence_score = custom_mean(pred_max_prob)
        print([pred, confidence_score])

max_stars_repo_name: pYtoner/EasyOCR

max_issues_repo_name: pYtoner/EasyOCR

max_forks_repo_name: pYtoner/EasyOCR


text:
"""
Test for file IO
"""
from pathlib import Path

import numpy as np
import pytest

import biorbd

def test_load_model():
    biorbd.Model('../../models/pyomecaman.bioMod')

def test_dof_ranges():
    m = biorbd.Model('../../models/pyomecaman.bioMod')
    pi = 3.14159265358979323846

    # Pelvis
    ranges = m.segment(0).ranges()
    assert(ranges[0].min() == -10)
    assert(ranges[0].max() == 10)
    assert(ranges[1].min() == -10)
    assert(ranges[1].max() == 10)
    assert(ranges[2].min() == -pi)
    assert(ranges[2].max() == pi)

    # BrasD
    ranges = m.segment(3).ranges()
    assert(ranges[0].min() == -pi)
    assert(ranges[0].max() == pi)
    assert(ranges[1].min() == 0)
    assert(ranges[1].max() == pi)

    # BrasG
    ranges = m.segment(4).ranges()
    assert(ranges[0].min() == -pi)
    assert(ranges[0].max() == pi)
    assert(ranges[1].min() == 0)
    assert(ranges[1].max() == pi)

    # CuisseD
    ranges = m.segment(5).ranges()
    assert(ranges[0].min() == -pi / 12)
    assert(ranges[0].max() == pi / 2 + pi / 3)

    # JambeD
    ranges = m.segment(6).ranges()
    assert(ranges[0].min() == -pi / 2 - pi / 6)
    assert(ranges[0].max() == 0)

    # PiedD
    ranges = m.segment(7).ranges()
    assert(ranges[0].min() == -pi / 2)
    assert(ranges[0].max() == pi / 2)

    # CuisseG
    ranges = m.segment(8).ranges()
    assert(ranges[0].min() == -pi / 12)
    assert(ranges[0].max() == pi / 2 + pi / 3)

    # JambeG
    ranges = m.segment(9).ranges()
    assert(ranges[0].min() == -pi / 2 - pi / 6)
    assert(ranges[0].max() == 0)

    # PiedG
    ranges = m.segment(10).ranges()
    assert(ranges[0].min() == -pi / 2)
    assert(ranges[0].max() == pi / 2)

max_stars_repo_name: jdowlingmedley/biorbd

max_issues_repo_name: jdowlingmedley/biorbd

max_forks_repo_name: jdowlingmedley/biorbd


text:
"""Implements a pipeline of Transformations

- Windows contain Tranformations
- Transformations have Params

This has become a re-implementation of the work done by Bob Kerner here:
- https://github.gatech.edu/bkerner3/trackbar

So kudos to him for the original work and inspiration!
"""
import logging

from . import (
    sequence_utils,
    parallel_utils
)

import numpy as np
import cv2

log = logging.getLogger(__name__)

def nothing(*a,**k):
    pass

class Window(object):
    counter = 1

    def __init__(self, transforms, name='', track_src=''):
        """A Window that transforms will be applied in named `name`

        Args:
            transforms ([Transform]): list of transforms in execution order
            name (str): default 'Step N' if not supplied
        """
        self.last_output = None
        self.transforms = transforms
        if not name:
            self.name = f'Step {self.counter}'
            Window.counter += 1
        else:
            self.name = name
        self.track_src = self.name if not track_src else track_src

    @property
    def dirty(self):
        """Returns index of first tranform that's dirty or -1 if None dirty"""
        for i, transform in enumerate(self.transforms):
            transform.update_params(self.track_src)
            if transform.dirty:
                return i
        return -1

    def draw(self, img):
        """Call _draw on each child transform in sequence and show it in window"""
        result = img
        for transform in self.transforms:
            result = transform._draw(result)
        self.last_output = result
        cv2.imshow(self.name, result.astype(np.uint8))
        return result

class Transform(object):
    def __init__(self):
        """A transformation that will be applied to an image"""
        self.last_output = None
        self.params = self._get_params()

    @property
    def dirty(self):
        """Returns True if any child parameter has been modified"""
        for p in self.params.values():
            if p.dirty:
                return True
        return False

    def _get_params(self):
        """Return dict of Param objects defined in order on class"""
        params = {}
        for attr in self.__class__.__dict__.keys():
            # TODO: Think there's a better Meta programming way to convert
            # class var's into instance vars (think django Fields)
            if isinstance(getattr(self, attr), Param):
                inst = getattr(self, attr)
                # Create instance var from class var so we don't ref class var
                new_inst = Param(
                    label=inst.label,
                    _max=inst.max,
                    _min=inst.min,
                    default=inst.default,
                    adjust=inst.adjust
                )
                setattr(self, attr, new_inst)
                label = new_inst.label if new_inst.label else attr
                new_inst.label = label
                params[label] = new_inst
        return params

    def update_params(self, win_name):
        """Updates the value from trackbar for each child parameter

        Args:
            win_name (str): Name of the window to get param from
        """
        if not self.params:
            return False
        for p in self.params.values():
            p.update_value(win_name)

    def compute_values(self):
        """Can modify self.your_param.value being passed to draw function"""
        pass

    def _draw(self, img):
        """Perform the transform and save the output

        Args:
            img (np.array): Image to operate on

        Returns:
            img (np.array): Output image
        """
        self.compute_values()
        result = img
        try:
            result = self.draw(img)
        except Exception as e:
            print(e)
        self.last_output = result

        # Nothing is dirty since we've just drawn image
        for p in self.params.values():
            p.dirty = False
        return result

    def draw(self, img):
        """Perform your transformation

        Args:
            img (np.array): Image to operate on

        Returns:
            img (np.array): Output image
        """
        raise NotImplementedError

class Param(object):
    def __init__(self, label='', _max=100, _min=0, default=1, adjust=None):
        """Represents a parameter to be used in your Transform.draw method

        Args:
            label (str): Name of param as shown in trackbard window. If not
                supplied, attribute name is used
            _max (int): max position of trackbar
            default (int): starting value of trackbar
            adjust (callable): adjustment that can be made to your value.
                ex: lambda x: x if x % 2 == 0 else x + 1
        """
        self.label = label
        self.max = _max
        self.min = _min
        self.adjust = adjust
        self.default = default
        self.value = max(_min, default)
        self.dirty = True
        self._pos = self.value
        if adjust:
            self.value = adjust(self.value)

    def update_value(self, win_name):
        """Fetches parm position from trackbar and sets dirty

        Args:
            win_name (str): Name of window to fetch param from
        """
        pos = cv2.getTrackbarPos(self.label, win_name)
        self.dirty = False if self._pos == pos else True
        self._pos = pos
        if callable(self.adjust):
            pos = self.adjust(pos)
        self.value = max(pos, self.min)

def run_pipe(transforms, img=None):
    """Run the pipeline

    Args:
        transforms ([Transform/Group]): list of all transforms or windows of
            transforms
        img (np.array): Input image to first transform. Can leave as None and
            allow your first Transform to perform any image loading (for things
            like videos, etc). Default: None
        verbose (bool): If True, sets logging.LogLevel(logging.DEBUG) (Default: False)
    """
    windows = sequence_utils.collect_windows(transforms)
    sequence_utils.check_dup_win_labels(windows)

    # Build the windows and trackbars
    for window in windows:
        cv2.namedWindow(
            window.name, cv2.WINDOW_NORMAL|cv2.WINDOW_KEEPRATIO|cv2.WINDOW_GUI_EXPANDED)
        for t in window.transforms:
            for k, v in t.params.items():
                cv2.createTrackbar(k, window.name, v._pos, v.max, nothing)

    # Draw an intitial image
    orig = None if img is None else np.copy(img)
    for window in windows:
        orig = window.draw(orig)

    # Loop performing transforms in each window
    while True:
        result = np.copy(img)
        # Break on escape key
        k = cv2.waitKey(1) & 0xFF
        if k==27:
            break

        # Break if all windows closed
        vals = [cv2.getWindowProperty(win.name, cv2.WND_PROP_VISIBLE) for win in windows]
        if not any(vals):
            break

        # Only update starting at first dirty window
        dirty = np.array([win.dirty for win in windows])
        clean = dirty == -1
        if all(clean):
            continue
        offset = np.argmax(dirty >= 0)
        result = result if offset == 0 else windows[offset - 1].last_output
        for window in windows[offset:]:
            result = window.draw(result)

    cv2.destroyAllWindows()

def run_parallel_pipe(transforms, img_paths):
    """Run a single list of transforms on multiple images

    Args:
        transforms [Transform]: list of transforms in order of execution
        img_paths ([str]): List of paths to images. First image is master
    """

    # Can only have a single window
    windows = sequence_utils.collect_windows(transforms)
    if len(windows) > 1:
        raise ValueError('Must only be a single window for parallel processing')
    sequence_utils.check_dup_win_labels(windows)

    # Get all images with their loaded img content
    images = parallel_utils.load_images(img_paths)

    # Setup the master window with the trackbars
    master_win = windows[0]
    master_win.name = images[0]['path']
    master_win.track_src = images[0]['path']
    parallel_utils.create_master_trackbars(master_win)

    # Create windows that won't have trackbars
    slave_wins = parallel_utils.create_slave_windows(transforms, master_win.name, images[1:])
    all_windows = [master_win] + slave_wins

    # Update once so we have an image in each window
    for img in images:
        img['img'] = np.copy(img['img'])
    for win, img in zip(all_windows, images):
        win.draw(img['img'])

    # Loop performing transforms in each window
    while True:
        # Copy all images
        for img in images:
            img['img'] = np.copy(img['img'])

        # Break on escape key
        k = cv2.waitKey(1) & 0xFF
        if k==27:
            break

        # Break if all windows closed
        vals = [cv2.getWindowProperty(win.name, cv2.WND_PROP_VISIBLE) for win in all_windows]
        if not any(vals):
            break

        if all_windows[0].dirty == -1:
            continue
        for win, img in zip(all_windows, images):
            win.draw(img['img'])

    cv2.destroyAllWindows()

max_stars_repo_name: geoff-va/trackpipe

max_issues_repo_name: geoff-va/trackpipe

max_forks_repo_name: geoff-va/trackpipe


text:
import numpy as np
from scipy import signal
import logging
logger = logging.getLogger(__name__)

def filter_win(rec, cutoff, sample_rate, numtaps, axis=-1):
    """
    low pass filter, returns filtered signal using FIR window
    filter

    Args:
        rec: record to filter
        cutoff: cutoff frequency
        sample_rate: sampling rate
        numtaps: number of frequency comppnents to use in the filer
        axis: axis of record to apply filter along
    """
    nyq_rate = sample_rate / 2.
    fir_coef = signal.firwin(numtaps, cutoff / nyq_rate)
    filtered_rec = signal.lfilter(fir_coef, [1.0], rec, axis=axis)
    return filtered_rec

def filter_ls(rec, cutoff, sample_rate, numtaps, axis=-1):
    """
    low pass filter, returns filtered signal using FIR
    least squared filter

    Args:
        rec: record to filter
        cufoff: cutoff frequency
        sample_rate: sampling rate
        numtaps: number of frequency comppnents to use in the filer
        axis: axis of record to apply filter along
    """
    raise NotImplementedError

def filter_ham(rec, cutoff, sample_rate, numtaps):
    raise NotImplementedError
    # sample_rate = self.sample_rate
    # nyq_rate = sample_rate / 2.
    # fir_coef = signal.firwin(numtaps,
    #                          cutoff / nyq_rate,
    #                          window="hamming")
    # filtered_rec = 2 * signal.lfilter(fir_coef, 1.0, rec)
    # return filtered_rec

class Demodulator:

    def __init__(self,
                 buffers_per_acquisition: int,
                 records_per_buffer: int,
                 samples_per_record: int,
                 sample_rate: float,
                 filter_settings,
                 demod_freqs,
                 average_buffers: bool=True,
                 average_records: bool=True,
                 integrate_samples: bool=True):

        self.filter_settings = filter_settings
        self.sample_rate = sample_rate
        if average_buffers:
            len_buffers = 1
        else:
            len_buffers = buffers_per_acquisition

        if average_records:
            len_records = 1
        else:
            len_records = records_per_buffer

        num_demods = len(demod_freqs)
        self.demod_freqs = np.array(demod_freqs)
        mat_shape = (num_demods, len_buffers,
                     len_records, samples_per_record)
        self.mat_shape = mat_shape
        integer_list = np.arange(samples_per_record)
        integer_mat = (np.outer(np.ones(len_buffers),
                                np.outer(np.ones(len_records), integer_list)))
        angle_mat = 2 * np.pi * \
                    np.outer(demod_freqs, integer_mat).reshape(mat_shape) / sample_rate
        self.cos_mat = np.cos(angle_mat)
        self.sin_mat = np.sin(angle_mat)
        self.integrate_samples = integrate_samples

    def demodulate(self, volt_rec, int_delay, int_time):
        """
        Applies low bandpass filter and demodulation fit,
        and integration limits to samples array

        Args:
            record (numpy array): record from alazar to be multiplied
                                  with the software signal, filtered and limited
                                  to ifantegration limits shape = (samples_taken, )

        Returns:
            re_limited (numpy array): shape = (demod_length, samples_after_limiting)
            im_limited (numpy array): shape = (demod_length, samples_after_limiting)
        """

        # volt_rec to matrix and multiply with demodulation signal matrices
        demod_length = len(self.demod_freqs)
        volt_rec_mat = np.outer(np.ones(demod_length), volt_rec).reshape(self.mat_shape)
        re_mat = np.multiply(volt_rec_mat, self.cos_mat)
        im_mat = np.multiply(volt_rec_mat, self.sin_mat)

        # filter out higher freq component
        cutoff = max(self.demod_freqs)/10
        if self.filter_settings['filter'] == 0:
            re_filtered = filter_win(re_mat, cutoff,
                                     self.sample_rate,
                                     self.filter_settings['numtaps'],
                                     axis=-1)
            im_filtered = filter_win(im_mat, cutoff,
                                     self.sample_rate,
                                     self.filter_settings['numtaps'],
                                     axis=-1)
        elif self.filter_settings['filter'] == 1:
            re_filtered = filter_ls(re_mat, cutoff,
                                    self.sample_rate,
                                    self.filter_settings['numtaps'],
                                    axis=-1)
            im_filtered = filter_ls(im_mat, cutoff,
                                    self.sample_rate,
                                    self.filter_settings['numtaps'],
                                    axis=-1)
        elif self.filter_settings['filter'] == 2:
            re_filtered = re_mat
            im_filtered = im_mat
        else:
            raise RuntimeError("Filter setting: {} not implemented".format(self.filter_settings['filter']))

        if self.integrate_samples:
            # apply integration limits
            beginning = int(int_delay * self.sample_rate)
            end = beginning + int(int_time * self.sample_rate)

            re_limited = re_filtered[..., beginning:end]
            im_limited = im_filtered[..., beginning:end]
        else:
            re_limited = re_filtered
            im_limited = im_filtered

        return re_limited, im_limited

    @staticmethod
    def verify_demod_freq(value, sample_rate, int_time):
        """
        Function to validate a demodulation frequency

        Checks:
            - 1e6 <= value <= 500e6
            - number of oscillation measured using current 'int_time' param value
              at this demodulation frequency value
            - oversampling rate for this demodulation frequency

        Args:
            value: proposed demodulation frequency
            sample_rate: Rate with witch data is sampled
            int_time: total time that data is sampled over used for demodulation
        Returns:
            bool: Returns True if suitable number of oscillations are measured and
            oversampling is > 1, False otherwise.
        Raises:
            ValueError: If value is not a valid demodulation frequency.
        """
        if (value is None) or not (1e6 <= value <= 500e6):
            raise ValueError('demod_freqs must be 1e6 <= value <= 500e6')
        isValid = True
        min_oscillations_measured = int_time * value
        oversampling = sample_rate / (2 * value)
        if min_oscillations_measured < 10:
            isValid = False
            logger.warning('{} oscillation measured for largest '
                            'demod freq, recommend at least 10: '
                            'decrease sampling rate, take '
                            'more samples or increase demodulation '
                            'freq'.format(min_oscillations_measured))
        elif oversampling < 1:
            isValid = False
            logger.warning('oversampling rate is {}, recommend > 1: '
                            'increase sampling rate or decrease '
                            'demodulation frequency'.format(oversampling))

        return isValid

max_stars_repo_name: FarBo/qdev-wrappers

max_issues_repo_name: FarBo/qdev-wrappers

max_forks_repo_name: FarBo/qdev-wrappers

reason: import numpy,from scipy


text:
#Refer to Bao and Meskas tutorial on D2Q9 LBM
import numpy as np
import matplotlib.pyplot as plt

# Global variables
nx = 128
ny = 128

dx = 1.0
dy = 1.0

nt = 1000
dt = 1.0

cval = dx/dt

lid_vel = 0.5
Re_n = 100.0

nu_val = float(nx)*lid_vel/Re_n
tau_val = 0.5*(6.0*nu_val/(cval*dx)+1.0)

print('The tau value is: ',tau_val)
print('The nu is: ', nu_val)

# Weights for equilibrium distribution
w = np.asarray([4.0/9.0, 1.0/9.0, 1.0/9.0, 1.0/9.0, 1.0/9.0, 1.0/36.0, 1.0/36.0, 1.0/36.0, 1.0/36.0])

def velocity_decompose(u,v):
	u_decompose = np.zeros(shape=(nx,ny,9),dtype='double')
	v_decompose = np.zeros(shape=(nx,ny,9),dtype='double')

	u_decompose[:,:,0] = 0.0
	u_decompose[:,:,1] = u[:,:]
	u_decompose[:,:,2] = 0.0
	u_decompose[:,:,3] = -u[:,:]
	u_decompose[:,:,4] = 0.0
	u_decompose[:,:,5] = u[:,:]
	u_decompose[:,:,6] = -u[:,:]
	u_decompose[:,:,7] = -u[:,:]
	u_decompose[:,:,8] = u[:,:]

	v_decompose[:,:,0] = 0.0
	v_decompose[:,:,1] = 0.0
	v_decompose[:,:,2] = v[:,:]
	v_decompose[:,:,3] = 0.0
	v_decompose[:,:,4] = -v[:,:]
	v_decompose[:,:,5] = v[:,:]
	v_decompose[:,:,6] = v[:,:]
	v_decompose[:,:,7] = -v[:,:]
	v_decompose[:,:,8] = -v[:,:]

	decompose = u_decompose + v_decompose

	return decompose

def calculate_s_i(u,v):
	# Calculate decomposed velocities
	u_dec = velocity_decompose(u,v)

	# Constant norm of velocity
	const_val = np.zeros(shape=(nx,ny),dtype='double')
	const_val[:,:] = -1.5/(cval*cval)*(u[:,:]*u[:,:] + v[:,:]*v[:,:])

	# Calculate s values for equilibrium calculation
	s_val = np.zeros(shape=(nx,ny,9),dtype='double')
	# e_0
	s_val[:,:,0] = w[0]*(const_val[:,:])
	# e_1
	s_val[:,:,1] = w[1]*(3.0/cval*(u_dec[:,:,1]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,1]**2)+const_val[:,:])
	# e_2
	s_val[:,:,2] = w[2]*(3.0/cval*(u_dec[:,:,2]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,2]**2)+const_val[:,:])
	# e_3
	s_val[:,:,3] = w[3]*(3.0/cval*(u_dec[:,:,3]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,3]**2)+const_val[:,:])
	# e_4
	s_val[:,:,4] = w[4]*(3.0/cval*(u_dec[:,:,4]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,4]**2)+const_val[:,:])
	# e_5
	s_val[:,:,5] = w[5]*(3.0/cval*(u_dec[:,:,5]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,5]**2)+const_val[:,:])
	# e_6
	s_val[:,:,6] = w[6]*(3.0/cval*(u_dec[:,:,6]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,6]**2)+const_val[:,:])
	# e_7
	s_val[:,:,7] = w[7]*(3.0/cval*(u_dec[:,:,7]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,7]**2)+const_val[:,:])
	# e_8
	s_val[:,:,8] = w[8]*(3.0/cval*(u_dec[:,:,8]) + 9.0/(2.0*cval*cval)*(u_dec[:,:,8]**2)+const_val[:,:])

	return s_val

def calculate_eq(u,v,rho):
	s_val = calculate_s_i(u,v)
	f_eq = np.zeros(shape=(nx,ny,9),dtype='double')
	range_val = np.arange(0,9)
	f_eq[:,:,range_val] = w[None,None,range_val]*rho[:,:,None] + rho[:,:,None]*s_val[:,:,range_val]

	return f_eq

def initialize_problem():

	# Initialize equilibrium distribution
	u = np.zeros(shape=(nx,ny),dtype='double')# U velocity
	v = np.zeros(shape=(nx,ny),dtype='double')# V velocity
	rho = np.ones(shape=(nx,ny),dtype='double')
	#Assign lid velocity
	u[1:nx-1,ny-1] = lid_vel

	# Distribution calculations
	f_eq = calculate_eq(u,v,rho)
	f = np.copy(f_eq)

	return f, f_eq, rho, u, v

def calculate_density(f):
	rho = np.sum(f,axis=2)
	return rho

def calculate_velocity(f,rho):
	u = f[:,:,1] - f[:,:,3] + (f[:,:,5]+f[:,:,8]) - (f[:,:,6]+f[:,:,7])
	v = f[:,:,2] - f[:,:,4] + (f[:,:,5]+f[:,:,6]) - (f[:,:,7]+f[:,:,8])

	u[:,:] = 1.0/rho[:,:]*(cval*u[:,:])
	v[:,:] = 1.0/rho[:,:]*(cval*v[:,:])

	return u, v

def streaming_boundaries(f,f_temp,f_eq,rho,u,v):
	#----------------------------------------------------------------------------#
	#----------------------------------------------------------------------------#
	# Left wall updated except top left corner - where velocity boundaries are used
	f[0,0:ny-1,1] = f_temp[0,0:ny-1,3] #Validated bounce back

	f[0,1:ny-1,2] = f_temp[0,0:ny-2,2]
	f[0,0,2] = f_temp[0,0,4]#Bounce back lower left corner - Validated

	f[0,0:ny-1,3] = f_temp[1,0:ny-1,3]#Validated

	f[0,0:ny-1,4] = f_temp[0,1:ny,4]# Validated

	f[0,0:ny-1,5] = f_temp[0,0:ny-1,7]# Validated - bounce back

	f[0,1:ny-1,6] = f_temp[1,0:ny-2,6]
	f[0,0,6] = f_temp[0,0,8]# Validated

	f[0,0:ny-1,7] = f_temp[1,1:ny,7]# Validated

	f[0,0:ny-1,8] = f_temp[0,0:ny-1,6]	#Validated bounce back

	#----------------------------------------------------------------------------#
	#----------------------------------------------------------------------------#
	# Right wall updated except top right corner - where velocity boundaries are used
	f[nx-1,0:ny-1,1] = f_temp[nx-2,0:ny-1,1] # Validated

	f[nx-1,1:ny-1,2] = f_temp[nx-1,0:ny-2,2]
	f[nx-1,0,2] = f_temp[nx-1,0,4]#Bounce back lower right corner - Validated

	f[nx-1,0:ny-1,3] = f_temp[nx-1,0:ny-1,1]#Validated bounce back

	f[nx-1,0:ny-1,4] = f_temp[nx-1,1:ny,4]# Validated

	f[nx-1,1:ny-1,5] = f_temp[nx-2,0:ny-2,5]# Validated
	f[nx-1,0,5] = f_temp[nx-1,0,7]# Validated

	f[nx-1,0:ny-1,6] = f_temp[nx-1,0:ny-1,8] #Validated bounce back

	f[nx-1,0:ny-1,7] = f_temp[nx-1,0:ny-1,5]# Validated bounce back

	f[nx-1,0:ny-1,8] = f_temp[nx-2,1:ny,8]	#Validated

	#----------------------------------------------------------------------------#
	#----------------------------------------------------------------------------#
	# Lower wall except bottom corners (taken care of previously)
	f[1:nx-1,0,1] = f_temp[0:nx-2,0,1] # Validated

	f[1:nx-1,0,2] = f_temp[1:nx-1,0,4] # Validated bounce back

	f[1:nx-1,0,3] = f_temp[2:nx,0,3] # Validated

	f[1:nx-1,0,4] = f_temp[1:nx-1,1,4] # Validated

	f[1:nx-1,0,5] = f_temp[1:nx-1,0,7] # Validated bounce back

	f[1:nx-1,0,6] = f_temp[1:nx-1,0,8] # Validated bounce back

	f[1:nx-1,0,7] = f_temp[2:nx,1,7] # Validated

	f[1:nx-1,0,8] = f_temp[0:nx-2,1,8] # Validated

	# ----------------------------------------------------------------------------------
	# ----------------------------------------------------------------------------------

	# Lid - velocity BC - needs to be fixed

	f[1:nx,ny-1,1] = f_temp[0:nx-1,ny-1,1]
	f[0,ny-1,1] = f_temp[0,ny-1,3] # Validated - bounce back

	f[0:nx,ny-1,2] = f_temp[0:nx,ny-2,2] # Validated

	f[0:nx-1,ny-1,3] = f_temp[1:nx,ny-1,3] # Validated
	f[nx-1,ny-1,3] = f_temp[nx-1,ny-1,1] # Validated bounce back

	f[0:nx,ny-1,4] = f_temp[0:nx,ny-1,2]

	f[1:nx,ny-1,5] = f_temp[0:nx-1,ny-2,5]
	f[0,ny-1,5] = f_temp[0,ny-1,7] # Validated - corner bounce back only

	f[0:nx-1,ny-1,6] = f_temp[1:nx,ny-2,6]
	f[nx-1,ny-1,6] = f_temp[nx-1,ny-1,8] # Validated - corner bounce back only

	rhon = f_temp[1:nx-1,ny-1,0] + f_temp[1:nx-1,ny-1,1] + f_temp[1:nx-1,ny-1,3] + 2.0*(f_temp[1:nx-1,ny-1,2]+f_temp[1:nx-1,ny-1,6]+f_temp[1:nx-1,ny-1,5])

	f[1:nx-1,ny-1,7] = f_temp[1:nx-1,ny-1,5] + 0.5*(f_temp[1:nx-1,ny-1,1]-f_temp[1:nx-1,ny-1,3])-0.5*rhon*lid_vel
	f[0,ny-1,7] = f_temp[0,ny-1,5]
	f[nx-1,ny-1,7] = f_temp[nx-1,ny-1,5]

	f[1:nx-1,ny-1,8] = f_temp[1:nx-1,ny-1,6] + 0.5*(f_temp[1:nx-1,ny-1,3]-f_temp[1:nx-1,ny-1,1])+0.5*rhon*lid_vel
	f[0,ny-1,8] = f_temp[0,ny-1,6]
	f[nx-1,ny-1,8] = f_temp[nx-1,ny-1,6]

	return f

def streaming(f,f_eq,rho,u,v):
	# Temporary copy of f
	f_temp = np.copy(f)
	# Boundary conditions update
	f = streaming_boundaries(f,f_temp,f_eq,rho,u,v)

	#Define ranges for within the domain - f0
	nx_internal, ny_internal = np.meshgrid(np.arange(1,nx-1), np.arange(1,ny-1))
	f[:,:,0] = f_temp[:,:,0]

	#Define ranges for within the domain - f1
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(0,nx-2), np.arange(1,ny-1))
	f[nx_internal,ny_internal,1] = f_temp[nx_neighbor,ny_neighbor,1]

	#Define ranges for within the domain - f2
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(1,nx-1), np.arange(0,ny-2))
	f[nx_internal,ny_internal,2] = f_temp[nx_neighbor,ny_neighbor,2]

	#Define ranges for within the domain - f3
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(2,nx), np.arange(1,ny-1))
	f[nx_internal,ny_internal,3] = f_temp[nx_neighbor,ny_neighbor,3]

	#Define ranges for within the domain - f4
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(1,nx-1), np.arange(2,ny))
	f[nx_internal,ny_internal,4] = f_temp[nx_neighbor,ny_neighbor,4]

	#Define ranges for within the domain - f5
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(0,nx-2), np.arange(0,ny-2))
	f[nx_internal,ny_internal,5] = f_temp[nx_neighbor,ny_neighbor,5]

	#Define ranges for within the domain - f6
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(2,nx), np.arange(0,ny-2))
	f[nx_internal,ny_internal,6] = f_temp[nx_neighbor,ny_neighbor,6]

	#Define ranges for within the domain - f7
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(2,nx), np.arange(2,ny))
	f[nx_internal,ny_internal,7] = f_temp[nx_neighbor,ny_neighbor,7]

	#Define ranges for within the domain - f8
	nx_neighbor, ny_neighbor = np.meshgrid(np.arange(0,nx-2), np.arange(2,ny))
	f[nx_internal,ny_internal,8] = f_temp[nx_neighbor,ny_neighbor,8]

	del f_temp

	return f

def collision_step(f,f_eq):
	f[:,:,:] = f[:,:,:] - 1.0/tau_val*(f[:,:,:]-f_eq[:,:,:])

def time_loop(f,f_eq,rho,u,v):

	t = 0

	print('Macroscopic density sum: ',np.sum(rho))
	while t<nt:
		t = t+1

		# Calculate new equilibrium distribution
		f_eq = calculate_eq(u,v,rho)

		# Collision step
		collision_step(f,f_eq)

		# Streaming
		streaming(f,f_eq,rho,u,v)

		# Calculate physical quantities from updated distribution
		rho = calculate_density(f)
		u, v = calculate_velocity(f,rho)

		# Setting BCs
		u[1:nx-1,ny-1] = lid_vel
		v[1:nx-1,ny-1] = 0.0

		print('Macroscopic density sum: ',np.sum(rho), 'Time step: ',t)

	plt.figure()
	plt.title('U')
	p1 = plt.contourf(np.transpose(u))
	plt.colorbar(p1)
	plt.show()

	plt.figure()
	plt.title('V')
	p1 = plt.contourf(np.transpose(v))
	plt.colorbar(p1)
	plt.show()

	plt.figure()
	plt.title('Macroscopic density')
	p1 = plt.contourf(np.transpose(rho))
	plt.colorbar(p1)
	plt.show()

f, f_eq, rho, u, v = initialize_problem()
time_loop(f,f_eq,rho,u,v)

max_stars_repo_name: Romit-Maulik/Tutorials-Demos-Practice

max_issues_repo_name: omersan/Practice

max_forks_repo_name: omersan/Practice


text:
# -*- coding: utf-8 -*-

import websocket
import msgpack
import gym
import io
from PIL import Image
from PIL import ImageOps
from gym import spaces
import numpy as np
import time

aglib = []

class port_number():
        def __init__(self):
            self.count = 4649

        def incliment(self):
            self.count += 1
            print self.count

class GymUnityEnv(gym.Env):

    def __init__(self):
        if 'pnum' in globals():
            self.pnum.incliment()
        else:
            self.pnum = new port_number()

        websocket.enableTrace(True)
        print "ws://localhost:"+ str(pnum.count) +"/CommunicationGym"
    	self.ws = websocket.create_connection("ws://localhost:"+ str(self.pnum.count) +"/CommunicationGym")

        print self.pnum.count
        self.action_space = spaces.Discrete(3)
        self.depth_image_dim = 32 * 32
        self.depth_image_count = 1
        self.observation, _, _ = self.receive()

    def reset(self):
        return self.observation

    def step(self, action):

        actiondata = msgpack.packb({"command": str(action)})
        self.ws.send(actiondata)

        # Unity Process

        observation, reward, end_episode = self.receive()

        return observation, reward, end_episode, {}

    def receive(self):

        while True:

            statedata = self.ws.recv()

            if not statedata:
                continue

            state = msgpack.unpackb(statedata)

            image = []
            for i in xrange(self.depth_image_count):
                image.append(Image.open(io.BytesIO(bytearray(state['image'][i]))))
            depth = []
            for i in xrange(self.depth_image_count):
                d = (Image.open(io.BytesIO(bytearray(state['depth'][i]))))
                depth.append(np.array(ImageOps.grayscale(d)).reshape(self.depth_image_dim))

            observation = {"image": image, "depth": depth}
            reward = state['reward']
            end_episode = state['endEpisode']

            return observation, reward, end_episode
            break

    def close(self):
        self.ws.close()u


text:
from __future__ import division, print_function
import h5py
import pandas as pd
import numpy as np

from dnn_reco.modules.data.labels.default_labels import simple_label_loader
"""
All label functions must have the following parameters and return values:

    Parameters
    ----------
    input_data : str
            Path to input data file.
    config : dict
        Dictionary containing all settings as read in from config file.
        Label function specific settings can be passed via the config file.
    label_names : None, optional
        The names of the labels. This defines which labels to include as well
        as the ordering.
        If label_names is None (e.g. first call to initate name list), then
        a list of label names needs to be created and returned.
    *args
        Variable length argument list.
    **kwargs
        Arbitrary keyword arguments.

    Returns
    -------
    np.ndarray
        The numpy array containing the labels.
        Shape: [batch_size] + label_shape
    list of str
        The names of the labels
"""

def biased_muongun(input_data, config, label_names=None, *args, **kwargs):
    """Default muon scattering labels in addtion to relaxed thresholds for
    classification.

    Parameters
    ----------
    input_data : str
            Path to input data file.
    config : dict
        Dictionary containing all settings as read in from config file.
    label_names : None, optional
        The names of the labels. This defines which labels to include as well
        as the ordering.
        If label_names is None, then all keys except event specificers will
        be used.

    *args
        Variable length argument list.
    **kwargs
        Arbitrary keyword arguments.

    Returns
    -------
    list of str
        The names of the labels
    """

    # define keys to load
    if 'labels_biased_selection_filter_keys' in config:
        filter_mask_keys = config['labels_biased_selection_filter_keys']
    else:
        filter_mask_keys = ['CascadeFilter', 'MuonFilter']

    if 'labels_biased_selection_dnn_label_names' in config:
        dnn_labels_names = config['labels_biased_selection_dnn_label_names']
    else:
        base = 'DeepLearningReco_'
        dnn_labels_names = {
            base+'event_selection_cscdl3_300m_01': [
                ('p_starting_300m', 0.8),
            ],
            base+'event_selection_dnn_cscd_l3a_starting_events_03': [
                ('p_starting_300m', 0.5),
            ],
        }

    # create empty label dict
    label_dict = {}

    try:
        # special handling for FilterMask keys
        # FilterMask columns often have the year appended, e.g. _12
        # Here, we map a filter_name_12 to filter_name
        with h5py.File(input_data, 'r') as f:
            _mask = f['FilterMask'][:]
        for filter_mask_key in filter_mask_keys:
            for filter_name in _mask.dtype.fields.keys():
                if filter_mask_key in filter_name:
                    label_dict['p_passed_'+filter_mask_key] = \
                        _mask[filter_name][:, 1]

        # load CscdL3 and CscdL3_Cont_Tag and all defined dnn_labels
        with pd.HDFStore(input_data,  mode='r') as f:

            _cscd_l3 = f['CscdL3']['value']
            _cscd_l3_cont_tag = f['CscdL3_Cont_Tag']['value']
            label_dict['p_passed_cscdl3'] = _cscd_l3
            label_dict['p_passed_cscdl3_cont'] = \
                np.logical_and(_cscd_l3_cont_tag > 0, _cscd_l3)

            for model_name, values in dnn_labels_names.items():
                _dnn = f[model_name]
                for name, cut in values:
                    label_name = '{}_{}'.format(
                        model_name.replace('DeepLearningReco_', ''), name)
                    score = _dnn[name]
                    label_dict['p_passed_' + label_name] = score > cut
                    label_dict[label_name] = score

    except KeyError as k:
        print(k)
        print('Skipping file')

    if label_names is None:
        label_names = sorted(label_dict.keys())

    labels = []
    for name in label_names:
        labels.append(label_dict[name])

    labels = np.array(labels, dtype=config['np_float_precision']).T

    return labels, label_names


text:
import numpy as np
import tensorflow as tf
import scipy.io

vgg_rawnet = scipy.io.loadmat('./imagenet-vgg-verydeep-19.mat')
print("Loaded vgg19 pretrained imagenet")

def build_net(ntype,nin,nwb=None,name=None):
    if ntype=='conv':
        return tf.nn.relu(tf.nn.conv2d(nin,nwb[0],strides=[1,1,1,1],padding='SAME',name=name)+nwb[1])
    elif ntype=='pool':
        return tf.nn.avg_pool(nin,ksize=[1,2,2,1],strides=[1,2,2,1],padding='SAME')

def get_weight_bias(vgg_layers,i):
    weights=vgg_layers[i][0][0][2][0][0]
    weights=tf.constant(weights)
    bias=vgg_layers[i][0][0][2][0][1]
    bias=tf.constant(np.reshape(bias,(bias.size)))
    return weights,bias

def build_vgg19(input,features='conv1_2',reuse=False):
    if reuse:
        tf.get_variable_scope().reuse_variables()
    net={}
    vgg_layers=vgg_rawnet['layers'][0]
    net['input']=input-np.array([123.6800, 116.7790, 103.9390]).reshape((1,1,1,3))
    net['conv1_1']=build_net('conv',net['input'],get_weight_bias(vgg_layers,0),name='vgg_conv1_1')
    net['conv1_2']=build_net('conv',net['conv1_1'],get_weight_bias(vgg_layers,2),name='vgg_conv1_2')
    net['pool1']=build_net('pool',net['conv1_2'])
    net['conv2_1']=build_net('conv',net['pool1'],get_weight_bias(vgg_layers,5),name='vgg_conv2_1')
    net['conv2_2']=build_net('conv',net['conv2_1'],get_weight_bias(vgg_layers,7),name='vgg_conv2_2')
    net['pool2']=build_net('pool',net['conv2_2'])
    net['conv3_1']=build_net('conv',net['pool2'],get_weight_bias(vgg_layers,10),name='vgg_conv3_1')
    net['conv3_2']=build_net('conv',net['conv3_1'],get_weight_bias(vgg_layers,12),name='vgg_conv3_2')
    net['conv3_3']=build_net('conv',net['conv3_2'],get_weight_bias(vgg_layers,14),name='vgg_conv3_3')
    net['conv3_4']=build_net('conv',net['conv3_3'],get_weight_bias(vgg_layers,16),name='vgg_conv3_4')
    net['pool3']=build_net('pool',net['conv3_4'])
    net['conv4_1']=build_net('conv',net['pool3'],get_weight_bias(vgg_layers,19),name='vgg_conv4_1')
    net['conv4_2']=build_net('conv',net['conv4_1'],get_weight_bias(vgg_layers,21),name='vgg_conv4_2')
    net['conv4_3']=build_net('conv',net['conv4_2'],get_weight_bias(vgg_layers,23),name='vgg_conv4_3')
    net['conv4_4']=build_net('conv',net['conv4_3'],get_weight_bias(vgg_layers,25),name='vgg_conv4_4')
    net['pool4']=build_net('pool',net['conv4_4'])
    net['conv5_1']=build_net('conv',net['pool4'],get_weight_bias(vgg_layers,28),name='vgg_conv5_1')
    net['conv5_2']=build_net('conv',net['conv5_1'],get_weight_bias(vgg_layers,30),name='vgg_conv5_2')
    return net

def compute_percep_loss(input, output, reuse=False):
    vgg_real=build_vgg19(output*255.0,reuse=reuse)
    vgg_fake=build_vgg19(input*255.0,reuse=True)
    p0=compute_l1_loss(vgg_real['input'],vgg_fake['input'])
    p1=compute_l1_loss(vgg_real['conv1_2'],vgg_fake['conv1_2'])/2.6
    p2=compute_l1_loss(vgg_real['conv2_2'],vgg_fake['conv2_2'])/4.8
    p3=compute_l1_loss(vgg_real['conv3_2'],vgg_fake['conv3_2'])/3.7
    p4=compute_l1_loss(vgg_real['conv4_2'],vgg_fake['conv4_2'])/5.6
    p5=compute_l1_loss(vgg_real['conv5_2'],vgg_fake['conv5_2'])*10/1.5
    return p0+p1+p2+p3+p4+p5

def compute_l1_loss(input, output):
    loss=tf.reduce_mean(tf.abs(input-output), [1,2,3], keepdims=True)
    return loss

max_issues_repo_name: banananabear/Deep-Deghosting-HDR

max_forks_repo_name: banananabear/Deep-Deghosting-HDR


text:
# --------------------------------------------------------
# ImageShop
# Licensed under The MIT License [see LICENSE for details]
# Copyright 2019 smarsu. All Rights Reserved.
# --------------------------------------------------------

import cv2
import numpy as np

def blur(image, factor=(0.25, 1)):
    """
    Args:
        image: ndarray.
        factor: (float, float) (low, high)
    """
    low, high = factor
    factor = np.random.uniform(low, high)
    h, w = image.shape[:2]
    tmph, tmpw = round(h * factor), round(w * factor)
    tmpimage = cv2.resize(image, (tmpw, tmph))
    image = cv2.resize(tmpimage, (w, h))
    return image

def flip(image):
    """0.5 perceptor for horizontal flip.

    Args:
        image: ndarray.

    Returns:
        image:
        target: If true, the ret image are flipped.
    """
    target = np.random.choice(2)
    if target:
        image = cv2.flip(image, 1)
    return image, target

def noise(image, alpha=(0.8, 1.25), beta=(-10, 10)):
    """Apply y = alpha * x + beta to image.

    Args:
        image: ndarray.
        alpha:
        beta:
    """
    image = image.astype(np.float32)
    alpha = np.random.uniform(alpha[0], alpha[1])
    beta = np.random.uniform(beta[0], beta[1])
    image = alpha * image + beta
    image = np.maximum(image, 0)
    image = np.minimum(image, 255)
    return image

max_stars_repo_name: smarsu/facenet

max_issues_repo_name: smarsu/facenet

max_forks_repo_name: smarsu/facenet


text:
import pickle
import numpy as np
import math
import cv2
import os

import torch

from datasets import BboxEncoder

def create2Dbbox_poly(bbox2D):
    u_min = bbox2D[0] # (left)
    u_max = bbox2D[1] # (rigth)
    v_min = bbox2D[2] # (top)
    v_max = bbox2D[3] # (bottom)

    poly = {}
    poly['poly'] = np.array([[u_min, v_min], [u_max, v_min],
                             [u_max, v_max], [u_min, v_max]], dtype='int32')

    return poly

def draw_2d_polys(img, polys):
    img = np.copy(img)
    for poly in polys:
        if 'color' in poly:
            bg = poly['color']
        else:
            bg = np.array([0, 255, 0], dtype='float64')

        cv2.polylines(img, np.int32([poly['poly']]), True, bg, lineType=cv2.LINE_AA, thickness=2)

        text = "%.2f" % poly["prob"]
        fontFace = cv2.FONT_HERSHEY_SIMPLEX
        fontScale = 0.4
        thickness = 1
        t_size, _ = cv2.getTextSize(text, fontFace, fontScale+0.1, thickness)
        p_org = tuple(np.amax(poly['poly'], 0))
        p_top = (p_org[0] - t_size[0], p_org[1] - t_size[1])
        p = (p_top[0] + 1, p_org[1] - 1)
        cv2.rectangle(img, p_top, p_org, bg, cv2.FILLED)
        cv2.putText(img, text, p, fontFace, fontScale, [255, 255, 255], 1, cv2.LINE_AA)

    return img

def draw_2d_polys_no_text(img, polys):
    img = np.copy(img)
    for poly in polys:
        if 'color' in poly:
            bg = poly['color']
        else:
            bg = np.array([100, 100, 100], dtype='float64')

        cv2.polylines(img, np.int32([poly['poly']]), True, bg, lineType=cv2.LINE_AA, thickness=2)

    return img

for sequence in ["0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010", "0011", "0012", "0013", "0014", "0015", "0016", "0017", "0018", "0019", "0020"]:
    print (sequence)

    # NOTE! here you can choose what model's output you want to visualize
    with open("/root/retinanet/training_logs/model_eval_val_seq/eval_dict_val_seq_%s.pkl" % sequence, "rb") as file:
        eval_dict = pickle.load(file)

    img_height = 375
    img_width = 1242

    project_dir = "/root/3DOD_thesis/"
    data_dir = project_dir + "data/kitti/tracking/training/"
    img_dir = data_dir + "image_02/" + sequence + "/"

    bbox_encoder = BboxEncoder(img_h=img_height, img_w=img_width)

    img_data_dict = {}
    for img_id in eval_dict:
        data_dict = {}

        img_dict = eval_dict[img_id]

        if img_dict["pred_bboxes"] is not None:
            pred_bboxes_xywh = torch.from_numpy(img_dict["pred_bboxes"])
            pred_probs = img_dict["pred_max_scores"]
            pred_class_labels = img_dict["pred_class_labels"]
            if pred_bboxes_xywh.size() == torch.Size([4]):
                pred_bboxes_xywh = pred_bboxes_xywh.unsqueeze(0)
                pred_probs = np.array([pred_probs])
                pred_class_labels = np.array([pred_class_labels])
            pred_bboxes_xxyy = bbox_encoder._xywh_2_xxyy(pred_bboxes_xywh).numpy()

            pred_bbox_polys = []
            for i in range(pred_bboxes_xxyy.shape[0]):
                pred_bbox_xxyy = pred_bboxes_xxyy[i]
                pred_prob = pred_probs[i]
                pred_class_label = pred_class_labels[i]

                pred_bbox_poly = create2Dbbox_poly(pred_bbox_xxyy)

                if pred_class_label == 1: # (Car)
                    pred_bbox_poly["color"] = np.array([255, 0, 0], dtype='float64')
                elif pred_class_label == 2: # (Pedestrian)
                    pred_bbox_poly["color"] = np.array([0, 200, 0], dtype='float64')
                elif pred_class_label == 3: # (Cyclist)
                    pred_bbox_poly["color"] = np.array([0, 0, 255], dtype='float64')

                pred_bbox_poly["prob"] = pred_prob

                pred_bbox_polys.append(pred_bbox_poly)
        else:
            pred_bbox_polys = []

        gt_bboxes_xywh = torch.from_numpy(img_dict["gt_bboxes"])
        if gt_bboxes_xywh.size(0) > 0:
            gt_class_labels = img_dict["gt_class_labels"]
            if gt_bboxes_xywh.size() == torch.Size([4]):
                gt_bboxes_xywh = gt_bboxes_xywh.unsqueeze(0)
                gt_class_labels = np.array([gt_class_labels])
            gt_bboxes_xxyy = bbox_encoder._xywh_2_xxyy(gt_bboxes_xywh).numpy()

            gt_bbox_polys = []
            for i in range(gt_bboxes_xxyy.shape[0]):
                gt_bbox_xxyy = gt_bboxes_xxyy[i]
                gt_class_label = gt_class_labels[i]

                gt_bbox_poly = create2Dbbox_poly(gt_bbox_xxyy)

                if gt_class_label == 1: # (Car)
                    gt_bbox_poly["color"] = np.array([255, 0, 0], dtype='float64')
                elif gt_class_label == 2: # (Pedestrian)
                    gt_bbox_poly["color"] = np.array([0, 200, 0], dtype='float64')
                elif gt_class_label == 3: # (Cyclist)
                    gt_bbox_poly["color"] = np.array([0, 0, 255], dtype='float64')

                gt_bbox_polys.append(gt_bbox_poly)
        else:
            gt_bbox_polys = []

        data_dict["gt_bbox_polys"] = gt_bbox_polys
        data_dict["pred_bbox_polys"] = pred_bbox_polys

        img_data_dict[img_id] = data_dict

    sorted_img_ids = []
    img_names = sorted(os.listdir(img_dir))
    for img_name in img_names:
        img_id = img_name.split(".png")[0]
        sorted_img_ids.append(img_id)

    # ################################################################################
    # # create a video of images (no bboxes):
    # ################################################################################
    # out = cv2.VideoWriter("/root/retinanet/training_logs/model_eval_val_seq/eval_val_seq_%s_img.avi" % sequence, cv2.VideoWriter_fourcc(*"MJPG"), 12, (img_width, img_height), True)
    #
    # for img_id in sorted_img_ids:
    #     print (img_id)
    #
    #     img = cv2.imread(img_dir + img_id + ".png", -1)
    #
    #     img = cv2.resize(img, (img_width, img_height)) # (the image MUST have the size specified in VideoWriter)
    #
    #     out.write(img)

    # ################################################################################
    # # create a video of images with GT:
    # ################################################################################
    # out = cv2.VideoWriter("/root/retinanet/training_logs/model_eval_val_seq/eval_val_seq_%s_img_GT.avi" % sequence, cv2.VideoWriter_fourcc(*"MJPG"), 12, (img_width, img_height), True)
    #
    # for img_id in sorted_img_ids:
    #     print (img_id)
    #
    #     img = cv2.imread(img_dir + img_id + ".png", -1)
    #
    #     img_with_gt_bboxes = img
    #
    #     if img_id in img_data_dict:
    #         data_dict = img_data_dict[img_id]
    #         gt_bbox_polys = data_dict["gt_bbox_polys"]
    #
    #         img_with_gt_bboxes = draw_2d_polys_no_text(img_with_gt_bboxes, gt_bbox_polys)
    #
    #     img_with_gt_bboxes = cv2.resize(img_with_gt_bboxes, (img_width, img_height)) # (the image MUST have the size specified in VideoWriter)
    #
    #     out.write(img_with_gt_bboxes)

    # ################################################################################
    # # create a video of images with pred:
    # ################################################################################
    # out = cv2.VideoWriter("/root/retinanet/training_logs/model_eval_val_seq/eval_val_seq_%s_img_pred.avi" % sequence, cv2.VideoWriter_fourcc(*"MJPG"), 12, (img_width, img_height), True)
    #
    # for img_id in sorted_img_ids:
    #     print (img_id)
    #
    #     img = cv2.imread(img_dir + img_id + ".png", -1)
    #
    #     img_with_pred_bboxes = img
    #
    #     if img_id in img_data_dict:
    #         data_dict = img_data_dict[img_id]
    #         pred_bbox_polys = data_dict["pred_bbox_polys"]
    #
    #         img_with_pred_bboxes = draw_2d_polys(img_with_pred_bboxes, pred_bbox_polys)
    #
    #     img_with_pred_bboxes = cv2.resize(img_with_pred_bboxes, (img_width, img_height)) # (the image MUST have the size specified in VideoWriter)
    #
    #     out.write(img_with_pred_bboxes)

    ################################################################################
    # create a video of images with GT on top of pred:
    ################################################################################
    out = cv2.VideoWriter("/root/retinanet/training_logs/model_eval_val_seq/eval_val_seq_%s_img_GT_pred.avi" % sequence, cv2.VideoWriter_fourcc(*"MJPG"), 12, (img_width, 2*img_height), True)

    for img_id in sorted_img_ids:
        print (img_id)

        img = cv2.imread(img_dir + img_id + ".png", -1)

        img_with_pred_bboxes = img
        img_with_gt_bboxes = img

        if img_id in img_data_dict:
            data_dict = img_data_dict[img_id]
            pred_bbox_polys = data_dict["pred_bbox_polys"]
            gt_bbox_polys = data_dict["gt_bbox_polys"]

            img_with_pred_bboxes = draw_2d_polys(img_with_pred_bboxes, pred_bbox_polys)
            img_with_gt_bboxes = draw_2d_polys_no_text(img_with_gt_bboxes, gt_bbox_polys)

        img_with_pred_bboxes = cv2.resize(img_with_pred_bboxes, (img_width, img_height))
        img_with_gt_bboxes = cv2.resize(img_with_gt_bboxes, (img_width, img_height))

        combined_img = np.zeros((2*img_height, img_width, 3), dtype=np.uint8)
        combined_img[0:img_height] = img_with_gt_bboxes
        combined_img[img_height:] = img_with_pred_bboxes

        out.write(combined_img)