text:
# coding=utf-8
import logging
from os.path import join
from typing import List
from typing import Optional
from typing import Tuple

import numpy as np
import pandas as pd
from matplotlib import pyplot as plt
from seaborn import boxplot
from seaborn import heatmap
from sklearn.preprocessing import normalize
from statsmodels.stats.weightstats import DescrStatsW

from .tests import vtests
from .tests import ttests
from ..plot.base import make_annotated_pie
from ..utils.base import remove_missing
from ..utils.base import to_percent
from ..utils.base import weighted_value_counts

logger = logging.getLogger(__name__)

def check_clustering(
    assignments: np.ndarray, names: Optional[List[str]] = None
) -> None:
    """Check which clusters are singular and report them."""
    _, counts = np.unique(assignments, return_counts=True)
    singular = counts < 2
    if np.any(singular):
        indx = np.argwhere(singular)
        malformed = np.asarray(names)[indx] if names is not None else indx
        logger.warning("Malformed clusters `{}`".format(malformed))

class ClusterDescriptiveStats(object):
    """Container for population and cluster descriptive stats and
    statistical tests."""

    def __init__(
        self,
        data: pd.DataFrame,
        assignments: np.ndarray,
        weights: Optional[pd.Series] = None,
        names: Optional[List[str]] = None,
        categoricals: Optional[List[str]] = None,
        numericals: Optional[List[str]] = None,
    ):
        """Initialize analysis metadata, validate weights and initialize
        the descriptive stats object."""
        check_clustering(assignments, names)
        self._data = data
        self._labels = assignments
        self._names = names
        self._columns = sorted(data.columns.tolist())
        self._df = data.loc[:, self._columns]
        self._cat_cols = sorted(categoricals) if categoricals else self._columns
        self._num_cols = sorted(numericals) if numericals else self._columns
        self._masks = []

        labels = np.unique(assignments)
        self.n_clusters = len(labels)
        for label in labels:
            mask = np.equal(assignments, label)
            self._masks.append(mask)

        if weights is None:
            weights = pd.Series(
                np.ones_like(assignments, dtype=np.float32), index=data.index
            )
        self.sample_weight = weights
        self.cluster_weight_series = pd.Series(
            data=np.zeros(self.n_clusters), index=self._names, name="Cluster weight"
        )
        self.pop_stats = DescrStatsW(data.loc[:, self._num_cols], weights, ddof=1)
        # TODO: Percentage of missing mass on each attribute?
        self.cluster_stats = []
        for i, mask in enumerate(self._masks):
            # Compute cluster means on attributes of interest
            x = data.loc[mask, self._num_cols]
            s = DescrStatsW(x, weights=weights[mask], ddof=1)
            self.cluster_stats.append(s)
            self.cluster_weight_series[i] = s.nobs

    def run(self, output_dir=None, topk=None, transparent=False):
        """Compute stats, plot, write relevant files to disk"""
        num_stats_df = self.numerical_stats()
        categorical_stats_dict = self.categorical_stats(topk=topk)
        ttests_df = self.ttests()
        vtests_dict = self.vtests(topk=topk)
        if output_dir:
            logger.info("[Clustering stats] In {}".format(output_dir))
            self.visualize(output_dir, num_stats_df, transparent)
            self.write_to_disk(
                output_dir, ttests_df, num_stats_df, categorical_stats_dict, vtests_dict
            )
        return num_stats_df, categorical_stats_dict, ttests_df, vtests_dict

    def write_to_disk(
        self,
        output_dir,
        ttests_df=None,
        num_stats_df=None,
        categorical_stats_dict=None,
        vtests_dict=None,
    ):
        # Distribution of weight per cluster
        clst_weight = self.cluster_weight_series.name
        weight_pct_name = clst_weight + " (%)"
        weight_pct_df = pd.DataFrame(
            np.append(to_percent(self.cluster_weight_series.values), 100),
            index=self.cluster_weight_series.index.tolist() + ["Total"],
            columns=[weight_pct_name],
        )
        weight_df = pd.DataFrame(
            np.append(self.cluster_weight_series, self.cluster_weight_series.sum()),
            index=self.cluster_weight_series.index.tolist() + ["Total"],
            columns=[clst_weight],
        )
        weight_tbl = pd.concat([weight_df, weight_pct_df], axis=1)
        weight_tbl.sort_values(weight_pct_name, inplace=True, ascending=False)
        weight_tbl.to_latex(join(output_dir, "cluster_weight.tex"))

        if num_stats_df is not None:
            # Round cluster stats
            output_stats_df = num_stats_df.round(2).fillna(0.0)
            output_stats_df.to_latex(join(output_dir, "cluster_stats.tex"))
            output_stats_df.to_csv(join(output_dir, "cluster_stats.csv"))
            output_stats_df.to_html(join(output_dir, "cluster_stats.html"))

        if categorical_stats_dict is not None:
            html = []
            for name, df in categorical_stats_dict.items():
                fp = join(output_dir, "{} categorical stats".format(name))
                df.to_latex(fp + ".tex", index=False)
                html.append("
{}
".format(name) + df.to_html())
            with open(join(output_dir, "ClusterStats.html"), mode="w") as f:
                f.write("

".join(html))

        if ttests_df is not None:
            # Round it
            ttests_df.to_latex(join(output_dir, "ttests.tex"))
            ttests_df.to_html(join(output_dir, "ttests.html"))

        if vtests_dict is not None:
            tests_html = []
            for name, tests in vtests_dict.items():
                tests_html.append("
{}
".format(name) + tests.to_html())
            with open(join(output_dir, "V-tests.html"), mode="w") as f:
                f.write("

".join(tests_html))
            """
            tests_path = join(output_dir, 'V-tests {}'.format(cluster_name))
            clst_tests.to_latex(tests_path + '.tex', index=False)
            clst_tests.to_csv(tests_path + '.csv')
            """

    def visualize(
        self,
        output_dir: str,
        stats_df: pd.DataFrame,
        transparent: bool,
        histo_cols: Tuple = (),
        weight_figsize=(16, 9),
        histo_figsize=(32, 32),
    ):
        """ Plot weights, numeric stats, """
        histo_cols = histo_cols or self._num_cols
        for mask, name in zip(self._masks, self._names):
            hist_path = join(output_dir, "Hist {}.png".format(name))
            data, nonull = remove_missing(
                self._df.loc[mask, histo_cols], return_mask=True, raise_error=False
            )
            weights = self.sample_weight.values[mask][nonull]
            data.hist(bins=100, figsize=histo_figsize, weights=weights)
            plt.savefig(hist_path, transparent=transparent)
            plt.close()
            plt.clf()

        fp = join(output_dir, "clusters_weight.png")
        self.cluster_weight_series.plot.bar(figsize=weight_figsize)
        plt.ylabel("No. households")
        plt.tight_layout()
        plt.savefig(fp, transparent=transparent)
        plt.close()
        plt.clf()

        fp = join(output_dir, "clusters_weight_pie.png")
        make_annotated_pie(self.cluster_weight_series, self.cluster_weight_series)
        plt.tight_layout()
        plt.savefig(fp, transparent=transparent)
        plt.close()
        plt.clf()

        if stats_df is not None:
            # Barplot for each attribute, showing global average
            n = len(self._num_cols)
            fig, ax = plt.subplots(
                n, 1, sharex=True, squeeze=False, figsize=(16, 9 * n)
            )
            ax = ax.ravel()
            for i, column in enumerate(self._num_cols):
                ylim = (20, 35) if column == "bmi" else None
                stats_df.loc[:, (column, "mean")].plot.bar(
                    yerr=stats_df.loc[:, (column, "std")],
                    title=column,
                    ylim=ylim,
                    ax=ax[i],
                )
                ax[i].axhline(stats_df.loc["Global", column]["mean"])
            fig.tight_layout()
            plt.savefig(join(output_dir, "Numerical attributes barplots.png"))
            plt.close()
            plt.clf()

            heatmap_path = join(output_dir, "stats_average_heatmap.png")
            fig, ax = plt.subplots(1, 1, figsize=(30, 30))
            normed_stats_df = normalize(
                stats_df.loc[:, (slice(None), "mean")], copy=False
            )
            heatmap(normed_stats_df, ax=ax, linewidths=0.5, annot=True)
            plt.yticks(rotation=0)
            plt.xticks(rotation=90)
            plt.tight_layout()
            plt.savefig(heatmap_path, transparent=transparent)
            plt.title("Average quantities purchased in tons (cluster x category)")
            plt.close()
            plt.clf()

        # Boxplot for each attribute, showing global average
        n = len(self._num_cols)
        fig, ax = plt.subplots(n, 1, sharex=True, squeeze=False, figsize=(16, 9 * n))
        ax = ax.ravel()  # Only squeezing 2nd dim
        names_df = pd.DataFrame(
            np.array(self._names)[self._labels],
            index=self._data.index,
            columns=["Cluster"],
        )
        data_with_names = pd.concat([self._data, names_df], axis=1)
        for i, column in enumerate(self._num_cols):
            boxplot(x="Cluster", y=column, data=data_with_names, ax=ax[i])
        fig.tight_layout()
        plt.savefig(join(output_dir, "Numerical attributes boxplots.png"))
        plt.close()
        plt.clf()

    def numerical_stats(self, columns=None):
        """Combine cluster-wise and population statistics"""
        columns = columns or self._num_cols
        logger.info("[Numerical stats] Attributes\n{}".format(columns))
        stats_columns = pd.MultiIndex.from_product(
            [columns, ["mean", "std"]], names=["attributes", "stats"]
        )
        stats_df = pd.DataFrame(
            index=self._names + ["Global"], columns=stats_columns, dtype=np.float64
        )
        stats_df.loc["Global", (slice(None), "mean")] = self.pop_stats.mean
        stats_df.loc["Global", (slice(None), "std")] = self.pop_stats.std
        for i, (stats, name) in enumerate(zip(self.cluster_stats, self._names)):
            # sample_stats.data = np.ma.masked_array(
            # sample_df.values, mask=sample_df.isnull())
            stats_df.loc[name, (slice(None), "mean")] = stats.mean
            stats_df.loc[name, (slice(None), "std")] = stats.std
        return stats_df

    def categorical_stats(self, columns=None, topk=None):
        columns = columns or self._cat_cols
        df = self._df.loc[:, columns]
        clst_stats = dict()
        logger.info("Categorical stats on attributes\n{}".format(df.columns.tolist()))
        for name, mask in zip(self._names, self._masks):
            cluster_desc_df = self._categorical_stats(
                df.loc[mask, :], self.sample_weight[mask], topk
            )
            clst_stats[name] = cluster_desc_df
        clst_stats["Population"] = self._categorical_stats(df, self.sample_weight, topk)
        return clst_stats

    @staticmethod
    def _categorical_stats(df, weights, topk=None):
        df = df.apply(weighted_value_counts, axis=0, weights=weights, pretty=True)
        df = df.head(topk).fillna("-")
        return df

    def ttests(self, columns=None):
        """Compute t-tests for numerical variables
        Notes
        -----
        The df passed here has to have columns that are aligned with the
        `pop_stats` structure. For the time being can't let the user choose
        which columns to test on: if we only want a subset of columns we
        first have to resolve the corresponding indices in `pop_stats`."""
        columns = columns or self._num_cols
        return ttests(
            self._df.loc[:, columns], self.pop_stats, self.cluster_stats, self._names
        )

    def vtests(self, columns=None, topk=None):
        columns = columns or self._cat_cols
        """Compute value tests for categorical variables."""
        return vtests(
            self._df.loc[:, columns],
            self._labels,
            self.sample_weight,
            self._names,
            topk,
        )

max_stars_repo_name: tritas/koino

max_issues_repo_name: tritas/koino

max_forks_repo_name: tritas/koino

reason: import numpy,from statsmodels


text:
#! /usr/bin/env python

import sys
import json
import re
import logging
import argparse
import time
import datetime
import pprint
import numpy
from urlparse import urlparse
from collections import defaultdict

CACHEABLE_CATEGORIES = ('image', 'text', 'css', 'javascript', 'flash', 'pdf',\
                        'xml', 'json', 'audio', 'video', 'font')
CACHE_CONTROL_CACHEABLE = ('public', 'max-age', 's-maxage', 'must-revalidate',\
                           'proxy-revalidate', 'no-transform')
CACHE_CONTROL_NOT_CACHEABLE = ('private', 'no-cache', 'no-store')

class HarError(Exception):
    pass

class HarObject(object):
    '''Encapsulates a single HAR request'''

    def __init__(self, object_json):
        self.json = object_json

        def process_headers(headers):
            # FIXME: don't discard multiple headers of same type
            header_dict = {}
            for header in headers:
                if header['name'] in header_dict:
                    logging.getLogger(__name__)\
                        .debug('Header "%s" already exists in this response.',
                        header['name'])
                else:
                    header_dict[header['name']] = header['value']
            return header_dict

        # make dicts for request and response headers
        self.request_headers = process_headers(self.json['request']['headers'])
        self.response_headers = process_headers(self.json['response']['headers'])

    def sanity_check(self, print_report=True):
        report = ''
        safe = True

        if 'timings' not in self.json:
            report += '-> No timings object'
            safe = False

        if self.category == 'unknown' and self.mime_type != '':
            report += '-> Unknown MIME type: %s\n' % self.mime_type

        if self.category == 'unknown' and self.content_size > 0:
            report += '-> Object of unknown type %s has nonzero size %d' % (self.mime_type, self.content_size)

        if self.content_size - self.content_compression != self.body_size:
            report += '-> Body size (%d) does not match content size (%d) minus compression (%d)\n'\
                % (self.body_size, self.content_size, self.content_compresion)

        if report != '' and print_report:
            print '\n========== SANITY CHECK REPORT ==========\n'
            print report + '\n'
            print self.json

        return safe

    def _get_mime_type(self):
        return self.json['response']['content']['mimeType']
    mime_type = property(_get_mime_type)

    def _get_category(self):
        if 'image' in self.mime_type:
            return 'image'
        elif 'audio' in self.mime_type:
            return 'audio'
        elif 'video' in self.mime_type:
            return 'video'
        elif 'css' in self.mime_type:
            return 'css'
        elif 'html' in self.mime_type:
            return 'html'
        elif 'javascript' in self.mime_type:
            return 'javascript'
        elif any(t in self.mime_type for t in ['text/plain', 'text/rtf']):
            return 'text'
        elif 'flash' in self.mime_type:
            return 'flash'
        elif any(t in self.mime_type for t in ['text/xml', 'application/xml']):
            return 'xml'
        elif 'json' in self.mime_type:
            return 'json'
        elif 'font' in self.mime_type:
            return 'font'
        elif 'octet-stream' in self.mime_type:
            return 'binary'
        else:
            return 'unknown'
    category = property(_get_category)

    @property
    def url(self):
        return self.json['request']['url']

    @property
    def host(self):
        return urlparse(self.url).netloc

    @property
    def path(self):
        return urlparse(self.url).path

    @property
    def filename(self):
        return self.url.split('/')[-1]

    def _get_protocol(self):
        return self.url.split('://')[0]
    protocol = property(_get_protocol)

    @property
    def response_code(self):
        return int(self.json['response']['status'])

    @property
    def object_start_time(self):
        return datetime.datetime.strptime(\
            self.json['startedDateTime'], '%Y-%m-%dT%H:%M:%S.%fZ')

    def _get_timings(self):
        return self.json['timings']
    timings = property(_get_timings)

    @property
    def request_headers_size(self):
        return int(self.json['request']['headersSize'])

    @property
    def request_body_size(self):
        return int(self.json['request']['bodySize'])

    @property
    def response_headers_size(self):
        return int(self.json['response']['headersSize'])

    @property
    def content_size(self):
        '''Size of original content (before compression)'''
        content_size = int(self.json['response']['content']['size'])
        return content_size

    @property
    def content_compression(self):
        '''Number of bytes saved by compression'''
        return int(self.json['response']['content']['compression'])

    def _get_body_size(self):
        '''Size of response body (possibly compressed)'''
        body_size = int(self.json['response']['bodySize'])
        return body_size
    response_body_size = property(_get_body_size)
    body_size = property(_get_body_size)

    def _get_explicitly_cacheable(self):
        '''Based on response headers, is this cacheable?'''
        try:
            if 'Expires' in self.response_headers:
                if self.response_headers['Expires'] in ['0', '-1']:
                    return False
                elif 'Date' in self.response_headers:
                    try:
                        expires = time.strptime(self.response_headers['Expires'][:-4].replace('-', ' '),\
                            '%a, %d %b %Y %H:%M:%S')
                        date = time.strptime(self.response_headers['Date'][:-4].replace('-', ' '),\
                            '%a, %d %b %Y %H:%M:%S')
                        return expires > date
                    except Exception as e:
                        logging.getLogger(__name__).warn('Error parsing date: %s', e)
                        pass
            elif 'Cache-Control' in self.response_headers:
                if any(t in self.response_headers['Cache-Control'] for t in CACHE_CONTROL_NOT_CACHEABLE):
                    return False
                elif 'max-age=' in self.response_headers['Cache-Control']:
                    max_age = int(self.response_headers['Cache-Control'].split('max-age=')[-1].split(',')[0])
                    return max_age > 0
                elif 's-maxage=' in self.response_headers['Cache-Control']:
                    max_age = int(self.response_headers['Cache-Control'].split('s-maxage=')[1].split(',')[0])
                    return max_age > 0
                elif any(t in self.response_headers['Cache-Control'] for t in CACHE_CONTROL_CACHEABLE):
                    return True
            else:
                return False
        except Exception as e:
            logging.warn('Error parsing Cache-Control or Expires header: %s', e)
        return False  # if there was an error, just say not cacheable
    explicitly_cacheable = property(_get_explicitly_cacheable)

    def _get_implicitly_cacheable(self):
        '''Based on MIME type, do we think this is cacheable?'''
        return self.category in CACHEABLE_CATEGORIES
    implicitly_cacheable = property(_get_implicitly_cacheable)

    def _get_tcp_handshake(self):
        return self.timings['connect'] > 0
    tcp_handshake = property(_get_tcp_handshake)

    def _get_ssl_handshake(self):
        return self.timings['ssl'] > 0
    ssl_handshake = property(_get_ssl_handshake)

    def __str__(self):
        return self.url
    def __repr__(self):
        return self.url

class Har(object):
    '''Encapsulates an HTTP Archive (HAR)'''

    def __init__(self, har_json):
        if har_json['log']['pages'] == [] or har_json['log']['entries'] == []:
            raise HarError('HAR is empty: %s' % har_json)

        self.data = har_json
        self.objects = []  # all objects, in order
        self.object_lists = defaultdict(list)
        self._sizes = []  # all object sizes, for computing mean/median
        self._hosts = set()
        self.page_start_time = datetime.datetime.strptime(\
            self.data['log']['pages'][0]['startedDateTime'],\
            '%Y-%m-%dT%H:%M:%S.%fZ')

        self._num_objects = 0
        self._num_bytes = 0
        self._num_explicitly_cacheable_objects = 0
        self._num_implicitly_cacheable_objects = 0
        self._num_explicitly_cacheable_bytes = 0
        self._num_implicitly_cacheable_bytes = 0
        self._num_http_objects = 0
        self._num_https_objects = 0
        self._num_tcp_handshakes = 0
        self._num_ssl_handshakes = 0
        self._total_tcp_handshake_ms = 0
        self._total_ssl_handshake_ms = 0
        self._total_handshake_ms = 0

        for obj_json in self.data['log']['entries']:
            try:
                obj = HarObject(obj_json)
                if not obj.sanity_check(print_report=False): continue
                #print '%d\t%s (%s)\t%s' % (obj.content_size, obj.mime_type, obj.category, obj.domain)

                self.objects.append(obj)
                self.object_lists[obj.category].append(obj)
                self._sizes.append(obj.content_size)
                self._hosts.add(obj.host)

                self._num_objects += 1
                self._num_bytes += obj.content_size

                if obj.protocol == 'http':
                    self._num_http_objects += 1
                elif obj.protocol == 'https':
                    self._num_https_objects += 1

                if obj.explicitly_cacheable:
                    self._num_explicitly_cacheable_objects += 1
                    self._num_explicitly_cacheable_bytes += obj.body_size
                if obj.implicitly_cacheable:
                    self._num_implicitly_cacheable_objects += 1
                    self._num_implicitly_cacheable_bytes += obj.body_size
                if obj.tcp_handshake:
                    self._num_tcp_handshakes += 1
                if obj.ssl_handshake:
                    self._num_ssl_handshakes += 1
                if obj.timings['connect'] >= 0:
                    self._total_tcp_handshake_ms += obj.timings['connect']
                    self._total_handshake_ms += obj.timings['connect']
                if obj.timings['ssl'] >= 0:
                    self._total_ssl_handshake_ms += obj.timings['ssl']
                    self._total_handshake_ms += obj.timings['ssl']
            except Exception as e:
                logging.warn('Error parsing HAR object:%s\n%s', e, obj_json)

    def sanity_check(self):
        for obj in [o for sublist in self.object_lists.values() for o in sublist]:
            obj.sanity_check()

    @classmethod
    def from_file(cls, path):
        with open(path, 'r') as f:
            data = json.load(f)
        f.closed
        return Har(data)

    @classmethod
    def sanitize_url(cls, url):
        return re.sub(r'[/\;,><&*:%=+@!#^()|?^]', '-', url)

    def get_by_name(self, name):
        try:
            return getattr(self, name)
        except:
            logging.warn('No property named "%s"' % name)
            return None

    def _get_url(self):
        id = self.data['log']['pages'][0]['id']
        title = self.data['log']['pages'][0]['title']

        return id if '://' in id else title
    url = property(_get_url)

    def _get_base_url(self):
        return self.url.split('://')[-1]
    base_url = property(_get_base_url)

    @property
    def on_load(self):
        return float(self.data['log']['pages'][0]['pageTimings']['onLoad'])

    @property
    def on_content_load(self):
        return float(self.data['log']['pages'][0]['pageTimings']['onContentLoad'])

    def _get_file_types(self):
        return self.object_lists.keys()
    file_types = property(_get_file_types)

    def get_objects(self, obj_type):
        return self.object_lists[obj_type]

    def _get_hosts(self):
        return self._hosts
    hosts = property(_get_hosts)

    def _get_num_hosts(self):
        return len(self._hosts)
    num_hosts = property(_get_num_hosts)

    def get_num_objects_by_type(self, obj_type):
        return len(self.object_lists[obj_type])

    def get_num_bytes_by_type(self, obj_type):
        ''' Returns total size, in bytes, of all objects of the specified type'''
        size = 0
        for obj in self.object_lists[obj_type]:
            size += obj.content_size
        return size

    def _get_num_objects(self):
        return self._num_objects
    num_objects = property(_get_num_objects)

    def _get_num_bytes(self):
        return self._num_bytes
    num_bytes = property(_get_num_bytes)

    def _get_num_mbytes(self):
        return self.num_bytes / 1000000.0
    num_mbytes = property(_get_num_mbytes)

    @property
    def mean_object_size(self):
        return numpy.mean(self._sizes)

    @property
    def median_object_size(self):
        return numpy.median(self._sizes)

    def _get_num_explicitly_cacheable_objects(self):
        return self._num_explicitly_cacheable_objects
    num_explicitly_cacheable_objects = property(_get_num_explicitly_cacheable_objects)

    def _get_num_explicitly_cacheable_bytes(self):
        return self._num_explicitly_cacheable_bytes
    num_explicitly_cacheable_bytes = property(_get_num_explicitly_cacheable_bytes)

    def _get_num_implicitly_cacheable_objects(self):
        return self._num_implicitly_cacheable_objects
    num_implicitly_cacheable_objects = property(_get_num_implicitly_cacheable_objects)

    def _get_num_implicitly_cacheable_bytes(self):
        return self._num_implicitly_cacheable_bytes
    num_implicitly_cacheable_bytes = property(_get_num_implicitly_cacheable_bytes)

    def _get_num_http_objects(self):
        return self._num_http_objects
    num_http_objects = property(_get_num_http_objects)

    def _get_num_https_objects(self):
        return self._num_https_objects
    num_https_objects = property(_get_num_https_objects)

    def _get_num_tcp_handshakes(self):
        return self._num_tcp_handshakes
    num_tcp_handshakes = property(_get_num_tcp_handshakes)

    def _get_num_ssl_handshakes(self):
        return self._num_ssl_handshakes
    num_ssl_handshakes = property(_get_num_ssl_handshakes)

    def _get_total_tcp_handshake_ms(self):
        return self._total_tcp_handshake_ms
    total_tcp_handshake_ms = property(_get_total_tcp_handshake_ms)

    def _get_total_ssl_handshake_ms(self):
        return self._total_ssl_handshake_ms
    total_ssl_handshake_ms = property(_get_total_ssl_handshake_ms)

    def _get_total_handshake_ms(self):
        return self._total_handshake_ms
    total_handshake_ms = property(_get_total_handshake_ms)

    def _get_profile(self):
        profile = {'num-objects-by-type':{}, 'num-bytes-by-type':{}}
        for t in self.file_types:
            profile['num-objects-by-type'][t] = self.get_num_objects_by_type(t)
            profile['num-bytes-by-type'][t] = self.get_num_bytes_by_type(t)
        profile['num-objects'] = self.num_objects
        profile['num-bytes'] = self.num_bytes
        profile['mean-object-size'] = self.mean_object_size
        profile['median-object-size'] = self.median_object_size
        profile['num-explicitly-cacheable-objects'] = self.num_explicitly_cacheable_objects
        profile['num-explicitly-cacheable-bytes'] = self.num_explicitly_cacheable_bytes
        profile['num-implicitly-cacheable-objects'] = self.num_implicitly_cacheable_objects
        profile['num-implicitly-cacheable-bytes'] = self.num_implicitly_cacheable_bytes
        profile['num-hosts'] = len(self.hosts)
        profile['num-tcp-handshakes'] = self.num_tcp_handshakes
        profile['num-ssl-handshakes'] = self.num_ssl_handshakes
        profile['total-tcp-handshake-ms'] = self.total_tcp_handshake_ms
        profile['total-ssl-handshake-ms'] = self.total_ssl_handshake_ms
        profile['total-handshake-ms'] = self.total_handshake_ms
        profile['percentage-explicitly-cacheable-bytes'] = \
            self.num_explicitly_cacheable_bytes / float(self.num_bytes)\
            if self.num_bytes else 0
        profile['percentage-implicitly-cacheable-bytes'] = \
            self.num_implicitly_cacheable_bytes / float(self.num_bytes)\
            if self.num_bytes else 0
        return profile
    profile = property(_get_profile)

    def __str__(self):
        return '%s (%s)' % (self.data['log']['pages'][0]['title'],\
                            self.data['log']['creator']['name'])
    def __repr__(self):
        return self.__str__()

def main():
    with open(args.har, 'r') as f:
        data = json.load(f)
    f.closed

    h = Har(data)

    if args.sanity_check:
        h.sanity_check()

    print h
    print pprint.pformat(h.profile)

if __name__ == '__main__':
    # set up command line args
    parser = argparse.ArgumentParser(description='Analyze a HAR file.')
    parser.add_argument('har', help='HAR file to analyze')
    parser.add_argument('-s', '--sanity_check', action='store_true', default=False, help='Check for problems in the HAR file')
    parser.add_argument('-q', '--quiet', action='store_true', default=False, help='only print errors')
    parser.add_argument('-v', '--verbose', action='store_true', default=False, help='print debug info. --quiet wins if both are present')
    args = parser.parse_args()

    # set up logging
    if args.quiet:
        level = logging.WARNING
    elif args.verbose:
        level = logging.DEBUG
    else:
        level = logging.INFO
    logging.basicConfig(
        format = "%(levelname) -10s %(asctime)s %(module)s:%(lineno) -7s %(message)s",
        level = level
    )

    main()

ext: py

max_stars_repo_name: eaufavor/chrome-webpage-profiler

max_issues_repo_name: eaufavor/chrome-webpage-profiler

max_forks_repo_name: eaufavor/chrome-webpage-profiler

reason: import numpy


text:
#!/usr/bin/python3

import sklearn.linear_model
import sklearn.model_selection
import sklearn.preprocessing
import numpy as np
import matplotlib.pyplot as plt
import argparse

COLOR = ['#1453ad', '#ad1453']

def train_test(x_train: np.ndarray, y_train: np.ndarray,
        x_test: np.ndarray, y_test: np.ndarray,
        reg_lambda: float = 1.0, std: bool = False) -> tuple:
    x_train_std, x_test_std = x_train, x_test
    if std:
        scaler = sklearn.preprocessing.StandardScaler()
        x_train_std = scaler.fit_transform(x_train)
        x_test_std = scaler.transform(x_test)
    clf = sklearn.linear_model.LogisticRegression(C = 1 / reg_lambda, max_iter = 1000).fit(x_train_std, y_train)
    error_rate_train = 1 - clf.score(x_train_std, y_train)
    error_rate_test =  1 - clf.score(x_test_std, y_test)
    return error_rate_train, error_rate_test

def cross_validation(x: np.ndarray, y: np.ndarray, reg_lambda: float = 1.0, std: bool = False) -> float:
    skf = sklearn.model_selection.StratifiedKFold(n_splits = 5, shuffle = True)
    error_rates = []
    for train_index, test_index in skf.split(x, y):
        _, error_rate = train_test(x[train_index], y[train_index], x[test_index], y[test_index],
            reg_lambda = reg_lambda, std = std)
        error_rates.append(error_rate)
    return np.mean(error_rates)

def part_a(x_train: np.ndarray, y_train: np.ndarray,
        x_test: np.ndarray, y_test: np.ndarray) -> None:
    print('STD')
    reg_lambdas = [1e-4, 1e-3, 1e-2, 5e-2, 1e-1, 5e-1, 1, 1e1]
    error_rates = []
    for l in reg_lambdas:
        er = cross_validation(x_train, y_train, reg_lambda = l, std = True)
        error_rates.append(er)
        print('%.5f\t%.6f' %  (l, er))

    opt_lambda = 1e-2
    print('Selected Lambda: %.5f' % opt_lambda)

    scaler = sklearn.preprocessing.StandardScaler()
    x_train_std = scaler.fit_transform(x_train)
    x_test_std = scaler.transform(x_test)
    error_rate_train, error_rate_test = train_test(x_train_std, y_train, x_test_std, y_test, reg_lambda = opt_lambda)
    print('ER Tr: %.6f, ER Test %.6f' %  (error_rate_train, error_rate_test))

    print('\nLOG')
    x_train_log = np.log(x_train + 0.1)
    x_test_log = np.log(x_test + 0.1)
    error_rate_cv = cross_validation(x_train_log, y_train, reg_lambda = opt_lambda)
    error_rate_train, error_rate_test = train_test(x_train_log, y_train, x_test_log, y_test, reg_lambda = opt_lambda)
    print('ER CV: %.6f, ER Tr: %.6f, ER Test %.6f' %  (error_rate_cv, error_rate_train, error_rate_test))

    print('\nBINARY')
    x_train_binary = np.where(x_train > 0, 1, 0)
    x_test_binary = np.where(x_test > 0, 1, 0)
    error_rate_cv = cross_validation(x_train_binary, y_train, reg_lambda = opt_lambda)
    error_rate_train, error_rate_test = train_test(x_train_binary, y_train, x_test_binary, y_test, reg_lambda = opt_lambda)
    print('ER CV: %.6f, ER Tr: %.6f, ER Test %.6f' %  (error_rate_cv, error_rate_train, error_rate_test))

def hist(x: np.ndarray, y: np.ndarray) -> tuple:
    bins = (np.unique(x), np.unique(y))
    hist, x_edges, y_edges = np.histogram2d(x, y, bins = bins)
    hist_x, hist_y = np.meshgrid(x_edges[:-1], y_edges[:-1])
    return hist_x.ravel(), hist_y.ravel(), hist.T.ravel()

def part_b(x: np.ndarray, y: np.ndarray,
        scatter_fname: str, non_spam_hist_fname: str, spam_hist_fname: str) -> None:
    x_binary = np.where(x > 0, 1, 0)
    x_scatter = np.sum(x_binary[:, 0:48], axis = 1)
    y_scatter = np.sum(x_binary[:, 48:54], axis = 1)

    x_scatter_non_spam = x_scatter[y == 0]
    y_scatter_non_spam = y_scatter[y == 0]
    x_scatter_spam = x_scatter[y == 1]
    y_scatter_spam = y_scatter[y == 1]

    fig, ax = plt.subplots()
    ax.scatter(x_scatter_non_spam, y_scatter_non_spam, label = 'Non-Spam',
        marker = 'o', c = COLOR[0], s = 12, alpha = 0.3)
    ax.scatter(x_scatter_spam, y_scatter_spam, label = 'Spam',
        marker = 'x', c = COLOR[1], s = 24, alpha = 0.3)
    ax.grid()
    ax.legend()
    fig.savefig(scatter_fname, dpi = 600, transparent = True)
    plt.clf()

    hist_x, hist_y, hist_z =  hist(x_scatter_non_spam, y_scatter_non_spam)
    dx = 0.75
    dy = 0.2

    fig = plt.figure()
    ax = fig.add_subplot(projection = '3d')
    ax.bar3d(hist_x, hist_y, 0, dx, dy, hist_z, label = 'Non-Spam', color = COLOR[0])
    ax.grid()
    fig.savefig(non_spam_hist_fname, dpi = 600, transparent = True)
    plt.clf()

    hist_x, hist_y, hist_z =  hist(x_scatter_spam, y_scatter_spam)
    dx = 0.75
    dy = 0.2

    fig = plt.figure()
    ax = fig.add_subplot(projection = '3d')
    ax.bar3d(hist_x, hist_y, 0, dx, dy, hist_z, label = 'Spam', color = COLOR[1])
    ax.grid()
    fig.savefig(spam_hist_fname, dpi = 600, transparent = True)
    plt.clf()

def main(x_train: np.ndarray, y_train: np.ndarray,
        x_test: np.ndarray, y_test: np.ndarray,
        scatter_fname: str, non_spam_hist_fname: str, spam_hist_fname: str) -> None:
    part_a(x_train, y_train, x_test, y_test)
    part_b(x_test, y_test, scatter_fname, non_spam_hist_fname, spam_hist_fname)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--x-train', help = 'x_train file', required = True)
    parser.add_argument('--y-train', help = 'y_train file', required = True)
    parser.add_argument('--x-test', help = 'x_test file', required = True)
    parser.add_argument('--y-test', help = 'y_test file', required = True)
    parser.add_argument('--scatter', help = 'scatter plot file', default = './scatter.png')
    parser.add_argument('--non-spam-hist', help = 'non spam hist plot file', default = './non_spam_hist.png')
    parser.add_argument('--spam-hist', help = 'spam hist plot file', default = './spam_hist.png')
    args = parser.parse_args()

    x_train, y_train = np.genfromtxt(args.x_train, delimiter = ','), np.genfromtxt(args.y_train, delimiter = ',')
    x_test, y_test = np.genfromtxt(args.x_test, delimiter = ','), np.genfromtxt(args.y_test, delimiter = ',')
    main(x_train, y_train, x_test, y_test, scatter_fname = args.scatter,
        non_spam_hist_fname = args.non_spam_hist, spam_hist_fname = args.spam_hist)

max_stars_repo_name: chenying-wang/usc-ee-coursework-public

max_issues_repo_name: chenying-wang/usc-ee-coursework-public

max_forks_repo_name: chenying-wang/usc-ee-coursework-public


text:
import unittest
import numpy
from kelvin.fci import FCI
from kelvin.ccsd import ccsd
try:
    from kelvin.hubbard_site_system import hubbard_site_system
    from kelvin.hubbard_system import HubbardSystem
    from lattice.hubbard import Hubbard1D
    from lattice.fci import FCISimple
    has_lattice = True
except ImportError:
    has_lattice = False

def compute_FCISimple(hub, nelec):
    myfci = FCISimple(hub, nelec, m_s=0)
    e, v = myfci.run()
    return e[0]

def compute_fci_kelvin(hub, nelec):
    nb = nelec//2
    na = nelec - nb
    Pa = None
    Pb = None
    sys = hubbard_site_system(0.0, hub, Pa, Pb, na=na, nb=nb)
    myfci = FCI(sys, nalpha=na, nbeta=nb)
    return myfci.run()[0]

def compute_fci_kelvinT(hub, T, mu):
    Pa = None
    Pb = None
    sys = hubbard_site_system(T, hub, Pa, Pb, mu=mu)
    myfci = FCI(sys, T=T, mu=mu)
    return myfci.run()[0]

@unittest.skipUnless(has_lattice, "Lattice module cannot be found")
class HubbardTest(unittest.TestCase):
    def setUp(self):
        self.thresh = 1e-10

    def _compare(self, nelec, L, U, t=1.0):
        hub = Hubbard1D(L, t, U, boundary='o')
        Es = compute_FCISimple(hub, nelec)
        Ec = compute_fci_kelvin(hub, nelec)
        return (Es, Ec)

    def _compare_pbc(self, nelec, L, U, t=1.0):
        hub = Hubbard1D(L, t, U)
        Es = compute_FCISimple(hub, nelec)
        Ec = compute_fci_kelvin(hub, nelec)
        return (Es, Ec)

    def test_2_1(self):
        nelec = 2
        L = 2
        U = 1.0
        Es, Ec = self._compare(nelec, L, U)
        diff = abs(Es - Ec)
        msg = "Expected: {}  Actual: {}".format(Es, Ec)
        self.assertTrue(diff < self.thresh, msg)

    def test_4_1(self):
        nelec = 4
        L = 4
        U = 1.0
        Es, Ec = self._compare(nelec, L, U)
        diff = abs(Es - Ec)
        msg = "Expected: {}  Actual: {}".format(Es, Ec)
        self.assertTrue(diff < self.thresh, msg)

    def test_4_1_pbc(self):
        nelec = 4
        L = 4
        U = 1.0
        Es, Ec = self._compare_pbc(nelec, L, U)
        diff = abs(Es - Ec)
        msg = "Expected: {}  Actual: {}".format(Es, Ec)
        self.assertTrue(diff < self.thresh, msg)

    def test_6_2_pbc(self):
        nelec = 6
        L = 6
        U = 2.0
        Es, Ec = self._compare_pbc(nelec, L, U)
        diff = abs(Es - Ec)
        msg = "Expected: {}  Actual: {}".format(Es, Ec)
        self.assertTrue(diff < self.thresh, msg)

    def test_ccsd_site(self):
        nelec = 2
        L = 2
        U = 1.0
        hub = Hubbard1D(L, 1.0, U, boundary='o')
        Eref = compute_FCISimple(hub, nelec)

        Oa = numpy.zeros((2))
        Ob = numpy.zeros((2))
        Oa[0] = 1.0
        Ob[1] = 1.0
        Pa = numpy.einsum('i,j->ij', Oa, Oa)
        Pb = numpy.einsum('i,j->ij', Ob, Ob)
        sys = hubbard_site_system(0.0, hub, Pa, Pb, na=1, nb=1)
        cc = ccsd(sys, iprint=0, damp=0.6, max_iter=80, econv=1e-11)
        T2aa = numpy.zeros((1, 1, 1, 1))
        T2ab = numpy.zeros((1, 1, 1, 1))
        T2bb = numpy.zeros((1, 1, 1, 1))
        T1a = numpy.zeros((1, 1))
        T1b = numpy.zeros((1, 1))
        T1a[0, 0] = 1.0
        T1b[0, 0] = 1.0
        T1 = (T1a, T1b)
        T2 = (T2aa, T2ab, T2bb)
        Eout, Ecc = cc.run(T1=T1, T2=T2)
        diff = abs(Eout - Eref)
        msg = "Expected: {} Actual: {}".format(Eref, Eout)
        self.assertTrue(diff < self.thresh, msg)

    def test_ccsd(self):
        nelec = 2
        L = 2
        U = 1.0
        hub = Hubbard1D(L, 1.0, U, boundary='o')
        Eref = compute_FCISimple(hub, nelec)

        Oa = numpy.zeros((2))
        Ob = numpy.zeros((2))
        Oa[0] = 1.0
        Ob[1] = 1.0
        Pa = numpy.einsum('i,j->ij', Oa, Oa)
        Pb = numpy.einsum('i,j->ij', Ob, Ob)
        sys = HubbardSystem(0.0, hub, Pa, Pb, na=1, nb=1)
        cc = ccsd(sys, iprint=0, max_iter=80, econv=1e-11)
        Eout, Ecc = cc.run()
        diff = abs(Eout - Eref)
        msg = "Expected: {} Actual: {}".format(Eref, Eout)
        self.assertTrue(diff < self.thresh, msg)

    def test_ft_ccsd(self):
        L = 2
        U = 1.0
        T = 2.0
        mu = 0.0
        hub = Hubbard1D(L, 1.0, U, boundary='o')
        Eref = compute_fci_kelvinT(hub, T, mu)
        ng = 8

        Oa = numpy.zeros((2))
        Ob = numpy.zeros((2))
        Oa[0] = 1.0
        Ob[1] = 1.0
        Pa = numpy.einsum('i,j->ij', Oa, Oa)
        Pb = numpy.einsum('i,j->ij', Ob, Ob)
        sys = HubbardSystem(T, hub, Pa, Pb, mu=mu)
        cc = ccsd(sys, iprint=0, max_iter=80,
                  econv=1e-11, T=T, mu=mu, ngrid=ng)
        Eout, Ecc = cc.run()
        diff = abs(Eout - Eref)
        msg = "Expected: {} Actual: {}".format(Eref, Eout)
        self.assertTrue(diff < 1e-4, msg)

    def test_ccsd_u_g(self):
        L = 2
        U = 1.0
        hub = Hubbard1D(L, 1.0, U, boundary='o')
        Oa = numpy.zeros((2))
        Ob = numpy.zeros((2))
        Oa[0] = 1.0
        Ob[1] = 1.0
        Pa = numpy.einsum('i,j->ij', Oa, Oa)
        Pb = numpy.einsum('i,j->ij', Ob, Ob)
        sys = HubbardSystem(0.0, hub, Pa, Pb, na=1, nb=1, orbtype='g')
        cc = ccsd(sys, iprint=0, max_iter=80, econv=1e-11)
        Eoutg, Eccg = cc.run()
        sys = HubbardSystem(0.0, hub, Pa, Pb, na=1, nb=1, orbtype='u')
        cc = ccsd(sys, iprint=0, max_iter=80, econv=1e-11)
        Eoutu, Eccu = cc.run()
        diff = abs(Eoutg - Eoutu)
        msg = "General: {} Unrestricted: {}".format(Eoutg, Eoutu)
        self.assertTrue(diff < self.thresh, msg)

    def test_ft_ccsd_u_g(self):
        L = 2
        U = 1.0
        T = 2.0
        mu = 0.0
        hub = Hubbard1D(L, 1.0, U, boundary='o')

        Oa = numpy.zeros((2))
        Ob = numpy.zeros((2))
        Oa[0] = 1.0
        Ob[1] = 1.0
        ng = 8
        Pa = numpy.einsum('i,j->ij', Oa, Oa)
        Pb = numpy.einsum('i,j->ij', Ob, Ob)
        sys = HubbardSystem(T, hub, Pa, Pb, mu=mu, orbtype='g')
        cc = ccsd(sys, iprint=0, max_iter=80,
                  econv=1e-11, T=T, mu=mu, ngrid=ng)
        Eoutg, Eccg = cc.run()
        sys = HubbardSystem(T, hub, Pa, Pb, mu=mu, orbtype='u')
        cc = ccsd(sys, iprint=0, max_iter=80,
                  econv=1e-11, T=T, mu=mu, ngrid=ng)
        Eoutu, Eccu = cc.run()
        diff = abs(Eoutg - Eoutu)
        msg = "General: {} Unrestricted: {}".format(Eoutg, Eoutu)
        self.assertTrue(diff < self.thresh, msg)

if __name__ == '__main__':
    unittest.main()

max_stars_repo_name: MoleOrbitalHybridAnalyst/kelvin

max_issues_repo_name: MoleOrbitalHybridAnalyst/kelvin

max_forks_repo_name: MoleOrbitalHybridAnalyst/kelvin


text:
import numpy as np
import json
import pkgutil
from flavio.classes import Parameter
from flavio.statistics.probability import MultivariateNormalDistribution

FFs = ["A0","A1","A12","V","T1","T2","T23"]
ai = ["a0","a1","a2"]
ff_a  = [(ff,a) for ff in FFs for a in ai]
a_ff_string  = [a + '_' + ff for ff in FFs for a in ai]

tex_a = {'a0': 'a_0', 'a1': 'a_1', 'a2': 'a_2', }
tex_ff = {'A0': 'A_0', 'A1': 'A_1', 'A12': r'A_{12}','V': 'V', 'T1': 'T_1','T2': 'T_2', 'T23': r'T_{23}', }

def get_ffpar(filename):
    f = pkgutil.get_data('flavio.physics', filename)
    data = json.loads(f.decode('utf-8'))
    central = np.array([data['central'][ff][a] for ff, a in ff_a])
    unc = np.array([data['uncertainty'][ff][a] for ff, a in ff_a])
    corr = np.array([[data['correlation'][ff1 + ff2][a1 + a2] for ff1, a1 in ff_a] for ff2, a2 in ff_a])
    # delete the parameters a0_A0 and a0_T2, which are instead fixed
    # using the exact kinematical relations, cf. eq. (16) of arXiv:1503.05534
    pos_a0_A0 = ff_a.index(('A0', 'a0'))
    pos_a0_T2 = ff_a.index(('T2', 'a0'))
    central = np.delete(central, [pos_a0_A0, pos_a0_T2])
    unc = np.delete(unc, [pos_a0_A0, pos_a0_T2])
    corr = np.delete(corr, [pos_a0_A0, pos_a0_T2], axis=0)
    corr = np.delete(corr, [pos_a0_A0, pos_a0_T2], axis=1)
    return [central, unc, corr]

def load_parameters(filename, process, constraints):
    implementation_name = process + ' BSZ'
    parameter_names = [implementation_name + ' ' + coeff_name for coeff_name in a_ff_string]
    # a0_A0 and a0_T2 are not treated as independent parameters!
    parameter_names.remove(implementation_name + ' a0_A0')
    parameter_names.remove(implementation_name + ' a0_T2')
    for parameter_name in parameter_names:
        try: # check if parameter object already exists
            p = Parameter[parameter_name]
        except: # otherwise, create a new one
            p = Parameter(parameter_name)
            # get LaTeX representation of coefficient and form factor names
            _tex_a = tex_a[parameter_name.split(' ')[-1].split('_')[0]]
            _tex_ff = tex_ff[parameter_name.split(' ')[-1].split('_')[-1]]
            p.tex = r'$' + _tex_a + r'^{' + _tex_ff + r'}$'
            p.description = r'BSZ form factor parametrization coefficient $' + _tex_a + r'$ of $' + _tex_ff + r'$'
        else: # if parameter exists, remove existing constraints
            constraints.remove_constraint(parameter_name)
    [central, unc, corr] = get_ffpar(filename)
    constraints.add_constraint(parameter_names,
            MultivariateNormalDistribution(central_value=central, covariance=np.outer(unc, unc)*corr) )

def bsz_load_v1_lcsr(constraints):
    load_parameters('data/arXiv-1503-05534v1/BKstar_LCSR.json', 'B->K*', constraints)
    load_parameters('data/arXiv-1503-05534v1/Bomega_LCSR.json', 'B->omega', constraints)
    load_parameters('data/arXiv-1503-05534v1/Brho_LCSR.json', 'B->rho', constraints)
    load_parameters('data/arXiv-1503-05534v1/Bsphi_LCSR.json', 'Bs->phi', constraints)
    load_parameters('data/arXiv-1503-05534v1/BsKstar_LCSR.json', 'Bs->K*', constraints)

def bsz_load_v1_combined(constraints):
    load_parameters('data/arXiv-1503-05534v1/BKstar_LCSR-Lattice.json', 'B->K*', constraints)
    load_parameters('data/arXiv-1503-05534v1/Bsphi_LCSR-Lattice.json', 'Bs->phi', constraints)
    load_parameters('data/arXiv-1503-05534v1/BsKstar_LCSR-Lattice.json', 'Bs->K*', constraints)

def bsz_load_v2_lcsr(constraints):
    load_parameters('data/arXiv-1503-05534v2/BKstar_LCSR.json', 'B->K*', constraints)
    load_parameters('data/arXiv-1503-05534v2/Bomega_LCSR.json', 'B->omega', constraints)
    load_parameters('data/arXiv-1503-05534v2/Brho_LCSR.json', 'B->rho', constraints)
    load_parameters('data/arXiv-1503-05534v2/Bsphi_LCSR.json', 'Bs->phi', constraints)
    load_parameters('data/arXiv-1503-05534v2/BsKstar_LCSR.json', 'Bs->K*', constraints)

def bsz_load_v2_combined(constraints):
    load_parameters('data/arXiv-1503-05534v2/BKstar_LCSR-Lattice.json', 'B->K*', constraints)
    load_parameters('data/arXiv-1503-05534v2/Bsphi_LCSR-Lattice.json', 'Bs->phi', constraints)
    load_parameters('data/arXiv-1503-05534v2/BsKstar_LCSR-Lattice.json', 'Bs->K*', constraints)

max_stars_repo_name: jackypheno/flavio

max_issues_repo_name: jackypheno/flavio

max_forks_repo_name: jackypheno/flavio


text:
import numpy as np
from baseline.remote import (
    RemoteModelREST,
    RemoteModelGRPC,
    register_remote,
    RemoteRESTClassifier,
    RemoteRESTTagger,
    RemoteRESTSeq2Seq,
    RemoteRESTEmbeddings,
    RemoteGRPCClassifier,
    RemoteGRPCTagger,
    RemoteGRPCSeq2Seq,
    RemoteGRPCEmbeddings,
)

class RemoteRESTTensorFlowMixin(RemoteModelREST):

    def create_request(self, examples):
        inputs = {}
        for feature in self.input_keys:
            tensor = examples[feature]
            if isinstance(tensor, np.ndarray):
                inputs[feature] = tensor.tolist()
            else:
                inputs[feature] = tensor
        request = {'signature_name': self.signature, 'inputs': inputs}
        return request

@register_remote('http-classify')
class RemoteRESTTensorFlowClassifier(RemoteRESTTensorFlowMixin, RemoteRESTClassifier): pass

@register_remote('http-tagger')
class RemoteRESTTensorFlowTagger(RemoteRESTTensorFlowMixin, RemoteRESTTagger): pass

@register_remote('http-seq2seq')
class RemoteRESTTensorFlowSeq2Seq(RemoteRESTTensorFlowMixin, RemoteRESTSeq2Seq): pass

@register_remote('http-servable-embeddings')
class RemoteRESTTensorFlowEmbeddings(RemoteRESTTensorFlowMixin, RemoteRESTEmbeddings): pass

@register_remote('grpc')
class RemoteGRPCTensorFlowMixin(RemoteModelGRPC): pass

@register_remote('grpc-classify')
class RemoteGRPCTensorFlowClassifier(RemoteGRPCTensorFlowMixin, RemoteGRPCClassifier): pass

@register_remote('grpc-tagger')
class RemoteGRPCTensorFlowTagger(RemoteGRPCTensorFlowMixin, RemoteGRPCTagger): pass

@register_remote('grpc-seq2seq')
class RemoteGRPCTensorFlowSeq2Seq(RemoteGRPCTensorFlowMixin, RemoteGRPCSeq2Seq): pass

@register_remote('grpc-servable-embeddings')
class RemoteGRPCTensorFlowEmbeddings(RemoteGRPCTensorFlowMixin, RemoteGRPCEmbeddings): pass

@register_remote('grpc-preproc')
class RemoteGRPCTensorFlowPreprocMixin(RemoteModelGRPC):

    def create_request(self, examples):
        # TODO: Remove TF dependency client side
        import tensorflow as tf

        request = self.predictpb.PredictRequest()
        request.model_spec.name = self.name
        request.model_spec.signature_name = self.signature
        if self.version is not None:
            request.model_spec.version.value = self.version

        for key in examples:
            if key.endswith('lengths'):
                shape = examples[key].shape
                tensor_proto = tf.contrib.util.make_tensor_proto(examples[key], shape=shape, dtype=tf.int32)
                request.inputs[key].CopyFrom(
                    tensor_proto
                )
            else:
                request.inputs[key].CopyFrom(
                    tf.contrib.util.make_tensor_proto(examples[key], shape=[len(examples[key]), 1])
                )
        return request

@register_remote('grpc-preproc-classify')
class RemoteGRPCTensorFlowPreprocClassifier(RemoteGRPCTensorFlowPreprocMixin, RemoteGRPCClassifier): pass

@register_remote('grpc-preproc-tagger')
class RemoteGRPCTensorFlowPreprocTagger(RemoteGRPCTensorFlowPreprocMixin, RemoteGRPCTagger): pass

@register_remote('grpc-preproc-seq2seq')
class RemoteGRPCTensorFlowPreprocSeq2Seq(RemoteGRPCTensorFlowPreprocMixin, RemoteGRPCSeq2Seq): pass

@register_remote('grpc-preproc-servable-embeddings')
class RemoteGRPCTensorFlowPreprocEmbeddings(RemoteGRPCTensorFlowPreprocMixin, RemoteGRPCEmbeddings): pass

@register_remote('http-preproc')
class RemoteRESTTensorFlowPreprocMixin(RemoteModelREST):

    def create_request(self, examples):
        inputs = {}
        if isinstance(examples['tokens'], np.ndarray):
            inputs['tokens'] = examples['tokens'].tolist()
        else:
            inputs['tokens'] = examples['tokens']
        for feature in self.input_keys:
            if feature.endswith('lengths'):
                if isinstance(examples[feature], np.ndarray):
                    inputs[feature] = examples[feature].tolist()
                else:
                    inputs[feature] = examples[feature]
        return {'signature_name': self.signature, 'inputs': inputs}

@register_remote('http-preproc-classify')
class RemoteRESTTensorFlowPreprocClassifier(RemoteRESTTensorFlowPreprocMixin, RemoteRESTClassifier): pass

@register_remote('http-preproc-tagger')
class RemoteRESTTensorFlowPreprocTagger(RemoteRESTTensorFlowPreprocMixin, RemoteRESTTagger): pass

@register_remote('http-preproc-seq2seq')
class RemoteRESTTensorFlowPreprocSeq2Seq(RemoteRESTTensorFlowPreprocMixin, RemoteRESTSeq2Seq): pass

@register_remote('http-preproc-servable-embeddings')
class RemoteRESTTensorFlowPreprocEmbeddings(RemoteRESTTensorFlowPreprocMixin, RemoteRESTEmbeddings): pass

max_stars_repo_name: sagnik/baseline

max_issues_repo_name: sagnik/baseline

max_forks_repo_name: sagnik/baseline


text:
import numpy as np
from ..utils.exceptions import ParameterError, ErrorMsgs

# init vars
F0 = 0
FSP = 200 / 3
BARK_FREQ = 1000
BARK_PT = (BARK_FREQ - F0) / FSP
LOGSTEP = np.exp(np.log(6.4) / 27.0)

def hz2erb(f):
    """
    Convert Hz frequencies to Bark.

    Args:
        f (np.array) : input frequencies [Hz].

    Returns:
        (np.array): frequencies in Bark [Bark].
    """
    return 24.7 * (4.37 * (f / 1000) + 1)

def erb2hz(fe):
    """
    Convert Bark frequencies to Hz.

    Args:
        fb (np.array) : input frequencies [Bark].

    Returns:
        (np.array)  : frequencies in Hz [Hz].
    """
    return ((fe / 24.7) - 1) * (1000. / 4.37)

def fft2erb(fft, fs, nfft):
    """
    Convert Bark frequencies to Hz.

    Args:
        fft (np.array) : fft bin numbers.

    Returns:
        (np.array): frequencies in Bark [Bark].
    """
    return hz2erb((fft * fs) / (nfft + 1))

def erb2fft(fb, fs, nfft):
    """
    Convert Bark frequencies to fft bins.

    Args:
        fb (np.array):  frequencies in Bark [Bark].
        fs (int):       sample rate/ sampling frequency of the signal.
        nfft (int):     the FFT size.
    Returns:
        (np.array) : fft bin numbers.
    """
    return (nfft + 1) * erb2hz(fb) / fs

def hz2bark(f):
    """
    Convert Hz frequencies to Bark acoording to Wang, Sekey & Gersho, 1992.

    Args:
        f (np.array) : input frequencies [Hz].

    Returns:
        (np.array): frequencies in Bark [Bark].
    """
    return 6. * np.arcsinh(f / 600.)

def bark2hz(fb):
    """
    Convert Bark frequencies to Hz.

    Args:
        fb (np.array) : input frequencies [Bark].

    Returns:
        (np.array)  : frequencies in Hz [Hz].
    """
    return 600. * np.sinh(fb / 6.)

def fft2hz(fft, fs, nfft):
    """
    Convert Bark frequencies to Hz.

    Args:
        fft (np.array): fft bin numbers.
        fs (int):       sample rate/ sampling frequency of the signal.
        nfft (int):     the FFT size.
    Returns:
        (np.array): frequencies in Bark [Bark].
    """
    return (fft * fs) / (nfft + 1)

def hz2fft(fb, fs, nfft):
    """
    Convert Bark frequencies to fft bins.

    Args:
        fb (np.array):  frequencies in Bark [Bark].
        fs (int):       sample rate/ sampling frequency of the signal.
        nfft (int):     the FFT size.
    Returns:
        (np.array) : fft bin numbers.
    """
    return (nfft + 1) * fb / fs

def fft2bark(fft, fs, nfft):
    """
    Convert Bark frequencies to Hz.

    Args:
        fft (np.array): fft bin numbers.
        fs (int):       sample rate/ sampling frequency of the signal.
        nfft (int):     the FFT size.

    Returns:
        (np.array): frequencies in Bark [Bark].
    """
    return hz2bark((fft * fs) / (nfft + 1))

def bark2fft(fb, fs, nfft):
    """
    Convert Bark frequencies to fft bins.

    Args:
        fb (np.array):  frequencies in Bark [Bark].
        fs (int):       sample rate/ sampling frequency of the signal.
        nfft (int):     the FFT size.
    Returns:
        (np.array) : fft bin numbers.
    """
    return (nfft + 1) * bark2hz(fb) / fs

def hz2mel(hz, htk=1):
    """
    Convert a value in Hertz to Mels

    Args:
         hz: a value in Hz. This can also be a numpy array, conversion proceeds element-wise.
         htk: Optional variable, if htk = 1 uses the mel axis defined in the HTKBook otherwise use Slaney's formula.

    Returns:
        a value in Mels. If an array was passed in, an identical sized array is returned.
    """
    if htk == 1:
        return 2595 * np.log10(1 + hz / 700.)

    else:
        # format variable
        hz = np.array(hz, ndmin=1)

        # definee lambda functions to simplify code
        def e(i):
            return (hz[i] - F0) / FSP

        def g(i):
            return BARK_PT + (np.log(hz[i] / BARK_FREQ) / np.log(LOGSTEP))

        mel = [e(i) if hz[i] < BARK_PT else g(i) for i in range(hz.shape[0])]
        return np.array(mel)

def mel2hz(mel, htk=1):
    """
    Convert a value in Mels to Hertz

    Args:
        mel: a value in Mels. This can also be a numpy array, conversion
             proceeds element-wise.
        htk: Optional variable, if htk = 1 uses the mel axis defined in the
             HTKBook otherwise use Slaney's formula.

    Returns:
        a value in Hertz. If an array was passed in, an identical sized array is returned.
    """
    if htk == 1:
        return 700 * (10**(mel / 2595.0) - 1)
    else:
        # format variable
        mel = np.array(mel, ndmin=1)

        # definee lambda functions to simplify code
        def e(i):
            return F0 + FSP * mel[i]

        def g(i):
            return BARK_FREQ * np.exp(np.log(LOGSTEP) * (mel[i] - BARK_PT))

        f = [e(i) if mel[i] < BARK_PT else g(i) for i in range(mel.shape[0])]
        return np.array(f)

def fft2melmx(nfft,
              fs,
              nfilts=0,
              bwidth=1,
              low_freq=0,
              high_freq=0,
              htk=False,
              constamp=False):
    """
    Generate a matrix of weights to combine FFT bins into Mel bins.

    Args:
        nfft      (int) : the FFT size.
                          (Default is 512)
        fs        (int) : sample rate/ sampling frequency of the signal.
                          (Default 16000 Hz)
        nfilts    (int) : the number of filters in the filterbank.
                          (Default 20)
        bwidth    (int) : the constant width of each band relative to standard Mel (default 1).
                          Default is 1.
        low_freq  (int) : lowest band edge of mel filters.
                          (Default 0 Hz)
        high_freq (int) : highest band edge of mel filters.
                          (Default samplerate/2)
        htkmel   (bool) : use HTK's version of the mel curve, not Slaney's.
                          Default is False.
        constamp (bool) : if True then make integration windows peak at 1, not sum to 1.
                          Default is False.

    Notes
        `low_freq` default is 0, but 133.33 is a common standard (to skip low frequencies).
        `high_freq` default is fs/2
        You can exactly duplicate the mel matrix in Slaney'ss using
        `fft2melmx(nfft=512, fs=8000, nfilts=40, bwidth=1, low_freq=133.33, high_freq=6855.5, 0)`

    Returns:
        matrix of weights to combine FFT bins into Mel bins.

    """
    if high_freq == 0:
        high_freq = fs / 2

    if nfilts == 0:
        nfilts = int(np.ceil(hz2mel(high_freq, htk) / 2))

    if not isinstance(nfilts, int):
        raise ParameterError(ErrorMsgs["nfilts"])

    if not isinstance(nfft, int):
        raise ParameterError(ErrorMsgs["nfft"])

    wts = np.zeros((nfilts, nfft))
    fftfrqs = (fs / nfft) * np.arange(0, nfft / 2 + 1)

    min_mel = hz2mel(low_freq, htk)
    max_mel = hz2mel(high_freq, htk)
    dif_mel = max_mel - min_mel
    binfrqs = mel2hz(
        min_mel + np.arange(0, nfilts + 2) * dif_mel / (nfilts + 1), htk)

    for i in range(nfilts):
        fs_tmp = binfrqs[np.arange(0, 3) + i]
        fs_tmp = fs_tmp[1] + bwidth * (fs_tmp - fs_tmp[1])
        # slopes
        loslope = (fftfrqs - fs_tmp[0]) / (fs_tmp[1] - fs_tmp[0])
        hislope = (fs_tmp[2] - fftfrqs) / (fs_tmp[2] - fs_tmp[1])
        wts[i, 0:nfft // 2 + 1] = np.maximum(0, np.minimum(loslope, hislope))

    if not constamp:
        wts = np.matmul(
            np.diag(2 / (binfrqs[2:nfilts + 2] - binfrqs[0:nfilts])), wts)

    return wts

def fft2barkmx(nfft, fs, nfilts=0, bwidth=1, low_freq=0, high_freq=0):
    """
    Generate a matrix of weights to combine FFT bins into Bark bins.

    Args:
        nfft      (int) : the FFT size.
                          (Default is 512)
        fs        (int) : sample rate/ sampling frequency of the signal.
                          (Default 16000 Hz)
        nfilts    (int) : the number of filters in the filterbank.
                          (Default 20)
        bwidth    (int) : the constant width of each band relative to standard Mel (default 1).
                          Default is 1.
        low_freq  (int) : lowest band edge of mel filters.
                          (Default 0 Hz)
        high_freq (int) : highest band edge of mel filters.
                          (Default sample rate/2)

    Notes:
        Optional nfilts specifies the number of output bands required
        (else one per bark), and width is the constant width of each
        band in Bark (default 1).

    Returns:
        matrix of weights to combine FFT bins into Bark bins.
    """
    if high_freq == 0:
        high_freq = fs / 2

    min_bark = hz2bark(low_freq)
    nyqbark = hz2bark(high_freq) - min_bark

    if nfilts == 0:
        nfilts = int(np.add(np.ceil(nyqbark), 1))

    if not isinstance(nfilts, int):
        raise ParameterError(ErrorMsgs["nfilts"])

    if not isinstance(nfft, int):
        raise ParameterError(ErrorMsgs["nfft"])

    wts = np.zeros((nfilts, nfft))
    step_barks = nyqbark / (nfilts - 1)
    binbarks = hz2bark((fs / nfft) * np.arange(0, nfft / 2 + 1))

    for i in range(nfilts):
        f_bark_mid = min_bark + i * step_barks
        lof = binbarks - f_bark_mid - 0.5
        hif = binbarks - f_bark_mid + 0.5
        wts[i, 0:nfft // 2 + 1] = 10**np.minimum(
            0,
            np.minimum(hif, -2.5 * lof) / bwidth)
    return wts

max_stars_repo_name: SuperKogito/cautious-palm-tree

max_issues_repo_name: SuperKogito/cautious-palm-tree

max_forks_repo_name: SuperKogito/cautious-palm-tree


text:
import os
import numpy as np
import datetime as dt
import pytest
from dotenv import load_dotenv
from postgresql_interface.postgresql_interface import *

@pytest.fixture(scope='session')
def create_env_variables():
    if os.path.isfile('.env'):
        load_dotenv(dotenv_path='.env')
    return os.environ['DATABASE_URL']

@pytest.fixture(scope='module')
def execute_insert_query(create_env_variables):
    heroku_db = postgres_factory(vendor='heroku', database_url=create_env_variables)
    statement_create_table = """
                            Drop table if exists test.simple;
                            drop schema if exists test;

                            CREATE SCHEMA test

                            create table test.simple (
                                Id INT not null,
                                Name VARCHAR(100) NOT NULL,
                                Activated BOOLEAN not null,
                                Date DATE NOT NULL
                            )
    """
    heroku_db.execute(statement_create_table)
    to_insert = pd.DataFrame.from_dict({'id': [1, 2, 3, 4],
                                        'name': ['Mercedes', 'Toyota', 'Suzuki', 'BMW'],
                                        'activated': [True, False, False, True],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2),
                                                 dt.date(2020, 3, 3), dt.date(2020, 4, 4)]})
    heroku_db.insert_table('test.simple', to_insert.copy())
    simple = heroku_db.query("SELECT * FROM test.simple")

    yield to_insert.sort_index().sort_index(axis=1) == simple.sort_index().sort_index(axis=1)

    heroku_db.execute("Drop table if exists test.simple; drop schema if exists test")

def test_execute_insert_query(execute_insert_query):
    """
    GIVEN a dataframe to insert into a heroku database
    WHEN it is inserted into the database
    THEN check that the values on the dataframe have been correctly created in the database
    :param execute_insert_query: fixture above
    :return:
    """
    assert execute_insert_query.all().all()

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

@pytest.fixture(scope='module')
def update(create_env_variables):
    heroku_db = postgres_factory(vendor='heroku', database_url=create_env_variables)
    statement_create_table = """
                            Drop table if exists test.simple;
                            drop schema if exists test;

                            CREATE SCHEMA test

                            create table test.simple (
                                Id INT not null,
                                Name VARCHAR(100) NOT NULL,
                                Activated BOOLEAN not null,
                                Date DATE NOT NULL
                            )
    """
    heroku_db.execute(statement_create_table)

    to_insert = pd.DataFrame.from_dict({'id': [1, 2, 3, 4],
                                        'name': ['Mercedes', 'Toyota', 'Suzuki', 'BMW'],
                                        'activated': [True, False, False, True],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2),
                                                 dt.date(2020, 3, 3), dt.date(2020, 4, 4)]})
    heroku_db.insert_table('test.simple', to_insert.copy())

    to_update = pd.DataFrame.from_dict({'id': [1, 2, 3, 4],
                                        'name': ['Ford', 'Vauxhall', 'SEAT', 'FIAT'],
                                        'activated': [False, True, True, False],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2),
                                                 dt.date(2020, 3, 3), dt.date(2020, 4, 4)]})
    heroku_db.update_table('test.simple', to_update, ['id', 'date'])

    simple = heroku_db.query("SELECT * FROM test.simple")
    yield to_update.sort_index().sort_index(axis=1) == simple.sort_index().sort_index(axis=1)

    heroku_db.execute("Drop table if exists test.simple; drop schema if exists test")

def test_execute_insert_update_query(update):
    """
    GIVEN a table in a heroku database
    WHEN it is updated
    THEN check that the values in the database are correctly updated
    :param update: fixture above
    :return:
    """
    assert update.all().all()

###############################################################################################
@pytest.fixture(scope='function')
def heroku_conn(create_env_variables):
    heroku_db = postgres_factory(vendor='heroku', database_url=create_env_variables)
    statement_create_table = """
                            Drop table if exists test.simple;
                            drop schema if exists test;

                            CREATE SCHEMA test

                            create table test.simple (
                                Id INT not null,
                                Name VARCHAR(100)  NULL,
                                Activated BOOLEAN not null,
                                Date DATE NOT NULL
                            )
    """
    heroku_db.execute(statement_create_table)

    to_insert = pd.DataFrame.from_dict({'id': [1, 2, 3, 4],
                                        'name': ['Mercedes', np.nan, 'Suzuki', 'BMW'],
                                        'activated': [True, False, False, True],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2),
                                                 dt.date(2020, 3, 3), dt.date(2020, 4, 4)]})
    heroku_db.insert_table('test.simple', to_insert.copy())

    yield heroku_db

    heroku_db.execute("Drop table if exists test.simple; drop schema if exists test")

def test_delete_several_cols(heroku_conn):
    """
    GIVEN a table in a heroku database
    WHEN some rows are deleted using several columns in the where clause
    THEN check that actually those rows are deleted
    :param heroku_conn: fixture above
    :return:
    """
    to_delete = pd.DataFrame.from_dict({'id': [1, 2, 3, 4],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2),
                                                 dt.date(2020, 3, 3), dt.date(2020, 4, 4)]})
    before_delete = heroku_conn.query("SELECT * FROM test.simple")
    heroku_conn.delete_from_table('test.simple', to_delete)
    simple = heroku_conn.query("SELECT * FROM test.simple")

    assert before_delete.shape[0] == 4
    assert simple.shape[0] == 0

def test_delete_one_col(heroku_conn):
    """
    GIVEN a table in a heroku database
    WHEN some rows are deleted using one column in the where clause
    THEN check that actually those rows are deleted
    :param heroku_conn: fixture above
    :return:
    """
    to_delete = pd.DataFrame.from_dict({'id': [1, 2, 3, 4]})
    before_delete = heroku_conn.query("SELECT * FROM test.simple")
    heroku_conn.delete_from_table('test.simple', to_delete)
    simple = heroku_conn.query("SELECT * FROM test.simple")

    assert before_delete.shape[0] == 4
    assert simple.shape[0] == 0

def test_truncate_true(heroku_conn):
    """
    GIVEN a table in a heroku database
    WHEN an insert is done after a truncate
    THEN check that previous values are actually deleted
    :param heroku_conn: fixture above
    :return:
    """
    # this is to make sure that actually there is something in the table before truncating it
    n_rows = heroku_conn.query("SELECT COUNT(*) AS count FROM test.simple").loc[0, 'count']

    # actual test
    to_insert = pd.DataFrame.from_dict({'id': [1, 2], 'name': ['Ford', 'Tesla'], 'activated': [True, False],
                                        'date': [dt.date(2020, 1, 1), dt.date(2020, 2, 2)]})
    heroku_conn.insert_table('test.simple', to_insert.copy(), truncate=True)
    simple = heroku_conn.query("SELECT * FROM test.simple")
    result = to_insert.sort_index().sort_index(axis=1) == simple.sort_index().sort_index(axis=1)

    # first assert is to make sure that actually there is something before truncating the table
    assert n_rows > 0
    assert result.all().all()


text:
import xlrd
from numpy import sqrt, abs, sin, cos, pi

import sys
sys.path.append('../..') # 如果最终要从main.py调用，则删掉这句
from GeneralMethod.PyCalcLib import Method, Simplified
from GeneralMethod.Report import Report

class Spectrometer:

    PREVIEW_FILENAME = "Preview.pdf"  # 预习报告模板文件的名称
    DATA_SHEET_FILENAME = "data.xlsx"  # 数据填写表格的名称
    REPORT_TEMPLATE_FILENAME = "GMR_empty.docx"  # 实验报告模板（未填数据）的名称
    REPORT_OUTPUT_FILENAME = "../../Report/Experiment2/2171Report.docx"  # 最后生成实验报告的相对路径

    def __init__(self):
        self.data = {}  # 存放实验中的各个物理量
        self.uncertainty = {}  # 存放物理量的不确定度
        self.report_data = {}  # 存放需要填入实验报告的
        print("1081 光的干涉实验\n1. 实验预习\n2. 数据处理")
        while True:
            try:
                oper = input("请选择: ").strip()
            except EOFError:
                sys.exit(0)
            if oper != '1' and oper != '2':
                print("输入内容非法！请输入一个数字1或2")
            else:
                break
        if oper == '1':
            print("现在开始实验预习")
            print("正在打开预习报告......")
            Method.start_file(self.PREVIEW_FILENAME)
        elif oper == '2':
            print("该实验没有自动数据处理")
            print("正在生成实验报告......")
            # 生成实验报告
            self.fill_report()
            print("实验报告生成完毕，正在打开......")
            Method.start_file(self.REPORT_OUTPUT_FILENAME)
            print("Done!")

    '''
    填充实验报告
    调用Report类，将数据填入Word文档格式的实验报告中
    '''
    def fill_report(self):
        # 调用ReportWriter类
        RW = Report()
        RW.load_replace_kw(self.report_data)
        RW.fill_report(self.REPORT_TEMPLATE_FILENAME, self.REPORT_OUTPUT_FILENAME)

if __name__ == '__main__':
    mc = Spectrometer()

reason: from numpy


text:
import pytest
from numpy.testing import assert_almost_equal

from pyproj.crs import GeographicCRS
from pyproj.crs.coordinate_operation import (
    AlbersEqualAreaConversion,
    AzimuthalEquidistantConversion,
    AzumuthalEquidistantConversion,
    EquidistantCylindricalConversion,
    GeostationarySatelliteConversion,
    HotineObliqueMercatorBConversion,
    LambertAzimuthalEqualAreaConversion,
    LambertAzumuthalEqualAreaConversion,
    LambertConformalConic1SPConversion,
    LambertConformalConic2SPConversion,
    LambertCylindricalEqualAreaConversion,
    LambertCylindricalEqualAreaScaleConversion,
    MercatorAConversion,
    MercatorBConversion,
    OrthographicConversion,
    PlateCarreeConversion,
    PolarStereographicAConversion,
    PolarStereographicBConversion,
    RotatedLatitudeLongitudeConversion,
    SinusoidalConversion,
    StereographicConversion,
    ToWGS84Transformation,
    TransverseMercatorConversion,
    UTMConversion,
    VerticalPerspectiveConversion,
)
from pyproj.exceptions import CRSError

def _to_dict(operation):
    param_dict = {}
    for param in operation.params:
        param_dict[param.name] = param.value
    return param_dict

def test_albers_equal_area_operation__defaults():
    aeaop = AlbersEqualAreaConversion(
        latitude_first_parallel=1, latitude_second_parallel=2
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Albers Equal Area"
    assert _to_dict(aeaop) == {
        "Easting at false origin": 0.0,
        "Latitude of 1st standard parallel": 1.0,
        "Latitude of 2nd standard parallel": 2.0,
        "Latitude of false origin": 0.0,
        "Longitude of false origin": 0.0,
        "Northing at false origin": 0.0,
    }

def test_albers_equal_area_operation():
    aeaop = AlbersEqualAreaConversion(
        latitude_first_parallel=1,
        latitude_second_parallel=2,
        latitude_false_origin=3,
        longitude_false_origin=4,
        easting_false_origin=5,
        northing_false_origin=6,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Albers Equal Area"
    assert _to_dict(aeaop) == {
        "Easting at false origin": 5.0,
        "Latitude of 1st standard parallel": 1.0,
        "Latitude of 2nd standard parallel": 2.0,
        "Latitude of false origin": 3.0,
        "Longitude of false origin": 4.0,
        "Northing at false origin": 6.0,
    }

def test_azumuthal_equidistant_operation__defaults():
    with pytest.warns(FutureWarning, match="deprecated"):
        aeop = AzumuthalEquidistantConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Modified Azimuthal Equidistant"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_azimuthal_equidistant_operation__defaults():
    aeop = AzimuthalEquidistantConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Modified Azimuthal Equidistant"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_azimuthal_equidistant_operation():
    aeop = AzimuthalEquidistantConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert aeop.name == "unknown"
    assert aeop.method_name == "Modified Azimuthal Equidistant"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_geostationary_operation__defaults():
    geop = GeostationarySatelliteConversion(sweep_angle_axis="x", satellite_height=10)
    assert geop.name == "unknown"
    assert geop.method_name == "Geostationary Satellite (Sweep X)"
    assert _to_dict(geop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Satellite height": 10.0,
    }

def test_geostationary_operation():
    with pytest.warns(UserWarning):
        geop = GeostationarySatelliteConversion(
            sweep_angle_axis="y",
            satellite_height=11,
            latitude_natural_origin=1,
            longitude_natural_origin=2,
            false_easting=3,
            false_northing=4,
        )
    assert geop.name == "unknown"
    assert geop.method_name == "Geostationary Satellite (Sweep Y)"
    assert _to_dict(geop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Satellite height": 11.0,
    }

def test_geostationary_operation__invalid_sweep():
    with pytest.raises(CRSError):
        GeostationarySatelliteConversion(sweep_angle_axis="P", satellite_height=10)

def test_lambert_azumuthal_equal_area_operation__defaults():
    with pytest.warns(FutureWarning, match="deprecated"):
        aeop = LambertAzumuthalEqualAreaConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Lambert Azimuthal Equal Area"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_lambert_azimuthal_equal_area_operation__defaults():
    aeop = LambertAzimuthalEqualAreaConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Lambert Azimuthal Equal Area"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_lambert_azimuthal_equal_area_operation():
    aeop = LambertAzimuthalEqualAreaConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert aeop.name == "unknown"
    assert aeop.method_name == "Lambert Azimuthal Equal Area"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_lambert_conformat_conic_2sp_operation__defaults():
    aeaop = LambertConformalConic2SPConversion(
        latitude_first_parallel=1, latitude_second_parallel=2
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Lambert Conic Conformal (2SP)"
    assert _to_dict(aeaop) == {
        "Easting at false origin": 0.0,
        "Latitude of 1st standard parallel": 1.0,
        "Latitude of 2nd standard parallel": 2.0,
        "Latitude of false origin": 0.0,
        "Longitude of false origin": 0.0,
        "Northing at false origin": 0.0,
    }

def test_lambert_conformat_conic_2sp_operation():
    aeaop = LambertConformalConic2SPConversion(
        latitude_first_parallel=1,
        latitude_second_parallel=2,
        latitude_false_origin=3,
        longitude_false_origin=4,
        easting_false_origin=5,
        northing_false_origin=6,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Lambert Conic Conformal (2SP)"
    assert _to_dict(aeaop) == {
        "Easting at false origin": 5.0,
        "Latitude of 1st standard parallel": 1.0,
        "Latitude of 2nd standard parallel": 2.0,
        "Latitude of false origin": 3.0,
        "Longitude of false origin": 4.0,
        "Northing at false origin": 6.0,
    }

def test_lambert_conformat_conic_1sp_operation__defaults():
    aeaop = LambertConformalConic1SPConversion()
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Lambert Conic Conformal (1SP)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Scale factor at natural origin": 1.0,
    }

def test_lambert_conformat_conic_1sp_operation():
    aeaop = LambertConformalConic1SPConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Lambert Conic Conformal (1SP)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Scale factor at natural origin": 0.5,
    }

def test_lambert_cylindrical_area_operation__defaults():
    lceaop = LambertCylindricalEqualAreaConversion()
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Lambert Cylindrical Equal Area"
    assert _to_dict(lceaop) == {
        "Latitude of 1st standard parallel": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_lambert_cylindrical_equal_area_operation():
    lceaop = LambertCylindricalEqualAreaConversion(
        latitude_first_parallel=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Lambert Cylindrical Equal Area"
    assert _to_dict(lceaop) == {
        "Latitude of 1st standard parallel": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_mercator_a_operation__defaults():
    aeaop = MercatorAConversion()
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Mercator (variant A)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Scale factor at natural origin": 1.0,
    }

def test_mercator_a_operation():
    aeaop = MercatorAConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Mercator (variant A)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Scale factor at natural origin": 0.5,
    }

def test_mercator_b_operation__defaults():
    lceaop = MercatorBConversion()
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Mercator (variant B)"
    assert _to_dict(lceaop) == {
        "Latitude of 1st standard parallel": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_mercator_b_operation():
    lceaop = MercatorBConversion(
        latitude_first_parallel=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Mercator (variant B)"
    assert _to_dict(lceaop) == {
        "Latitude of 1st standard parallel": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_hotline_oblique_mercator_b_operation__defaults():
    hop = HotineObliqueMercatorBConversion(
        latitude_projection_centre=0,
        longitude_projection_centre=0,
        azimuth_initial_line=0,
        angle_from_rectified_to_skew_grid=0,
    )
    assert hop.name == "unknown"
    assert hop.method_name == "Hotine Oblique Mercator (variant B)"
    assert _to_dict(hop) == {
        "Latitude of projection centre": 0.0,
        "Longitude of projection centre": 0.0,
        "Azimuth of initial line": 0.0,
        "Angle from Rectified to Skew Grid": 0.0,
        "Scale factor on initial line": 1.0,
        "Easting at projection centre": 0.0,
        "Northing at projection centre": 0.0,
    }

def test_hotline_oblique_mercator_b_operation():
    hop = HotineObliqueMercatorBConversion(
        latitude_projection_centre=1,
        longitude_projection_centre=2,
        azimuth_initial_line=3,
        angle_from_rectified_to_skew_grid=4,
        scale_factor_on_initial_line=0.5,
        easting_projection_centre=6,
        northing_projection_centre=7,
    )
    assert hop.name == "unknown"
    assert hop.method_name == "Hotine Oblique Mercator (variant B)"
    assert _to_dict(hop) == {
        "Latitude of projection centre": 1.0,
        "Longitude of projection centre": 2.0,
        "Azimuth of initial line": 3.0,
        "Angle from Rectified to Skew Grid": 4.0,
        "Scale factor on initial line": 0.5,
        "Easting at projection centre": 6.0,
        "Northing at projection centre": 7.0,
    }

def test_orthographic_operation__defaults():
    aeop = OrthographicConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Orthographic"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_orthographic_operation():
    aeop = OrthographicConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert aeop.name == "unknown"
    assert aeop.method_name == "Orthographic"
    assert _to_dict(aeop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_polar_stereographic_a_operation__defaults():
    aeaop = PolarStereographicAConversion(90)
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Polar Stereographic (variant A)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 90.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Scale factor at natural origin": 1.0,
    }

def test_polar_stereographic_a_operation():
    aeaop = PolarStereographicAConversion(
        latitude_natural_origin=-90,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Polar Stereographic (variant A)"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": -90.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Scale factor at natural origin": 0.5,
    }

def test_polar_stereographic_b_operation__defaults():
    aeop = PolarStereographicBConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Polar Stereographic (variant B)"
    assert _to_dict(aeop) == {
        "Latitude of standard parallel": 0.0,
        "Longitude of origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_polar_stereographic_b_operation():
    aeop = PolarStereographicBConversion(
        latitude_standard_parallel=1,
        longitude_origin=2,
        false_easting=3,
        false_northing=4,
    )
    assert aeop.name == "unknown"
    assert aeop.method_name == "Polar Stereographic (variant B)"
    assert _to_dict(aeop) == {
        "Latitude of standard parallel": 1.0,
        "Longitude of origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_sinusoidal_operation__defaults():
    aeop = SinusoidalConversion()
    assert aeop.name == "unknown"
    assert aeop.method_name == "Sinusoidal"
    assert _to_dict(aeop) == {
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_sinusoidal_operation():
    aeop = SinusoidalConversion(
        longitude_natural_origin=2, false_easting=3, false_northing=4
    )
    assert aeop.name == "unknown"
    assert aeop.method_name == "Sinusoidal"
    assert _to_dict(aeop) == {
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_stereographic_operation__defaults():
    aeaop = StereographicConversion()
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Stereographic"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Scale factor at natural origin": 1.0,
    }

def test_stereographic_operation():
    aeaop = StereographicConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Stereographic"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Scale factor at natural origin": 0.5,
    }

def test_utm_operation__defaults():
    aeop = UTMConversion(zone=2)
    assert aeop.name == "UTM zone 2N"
    assert aeop.method_name == "Transverse Mercator"

def test_utm_operation():
    aeop = UTMConversion(zone=2, hemisphere="s")
    assert aeop.name == "UTM zone 2S"
    assert aeop.method_name == "Transverse Mercator"

def test_transverse_mercator_operation__defaults():
    aeaop = TransverseMercatorConversion()
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Transverse Mercator"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
        "Scale factor at natural origin": 1.0,
    }

def test_transverse_mercator_operation():
    aeaop = TransverseMercatorConversion(
        latitude_natural_origin=1,
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Transverse Mercator"
    assert _to_dict(aeaop) == {
        "Latitude of natural origin": 1.0,
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
        "Scale factor at natural origin": 0.5,
    }

def test_vertical_perspective_operation__defaults():
    aeaop = VerticalPerspectiveConversion(viewpoint_height=10)
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Vertical Perspective"
    assert _to_dict(aeaop) == {
        "Latitude of topocentric origin": 0.0,
        "Longitude of topocentric origin": 0.0,
        "Ellipsoidal height of topocentric origin": 0.0,
        "Viewpoint height": 10.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_vertical_perspective_operation():
    aeaop = VerticalPerspectiveConversion(
        viewpoint_height=10,
        latitude_topocentric_origin=1,
        longitude_topocentric_origin=2,
        false_easting=3,
        false_northing=4,
        ellipsoidal_height_topocentric_origin=5,
    )
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "Vertical Perspective"
    assert _to_dict(aeaop) == {
        "Latitude of topocentric origin": 1.0,
        "Longitude of topocentric origin": 2.0,
        "Ellipsoidal height of topocentric origin": 5.0,
        "Viewpoint height": 10.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

def test_rotated_latitude_longitude_operation__defaults():
    aeaop = RotatedLatitudeLongitudeConversion(o_lat_p=1, o_lon_p=2)
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "PROJ ob_tran o_proj=longlat"
    assert _to_dict(aeaop) == {"o_lat_p": 1.0, "o_lon_p": 2.0, "lon_0": 0.0}

def test_rotated_latitude_longitude_operation():
    aeaop = RotatedLatitudeLongitudeConversion(o_lat_p=1, o_lon_p=2, lon_0=3)
    assert aeaop.name == "unknown"
    assert aeaop.method_name == "PROJ ob_tran o_proj=longlat"
    assert _to_dict(aeaop) == {"o_lat_p": 1.0, "o_lon_p": 2.0, "lon_0": 3.0}

def test_lambert_cylindrical_equal_area_scale_operation__defaults():
    lceaop = LambertCylindricalEqualAreaScaleConversion()
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Lambert Cylindrical Equal Area"
    assert _to_dict(lceaop) == {
        "Latitude of 1st standard parallel": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

def test_lambert_cylindrical_equal_area_scale_operation():
    lceaop = LambertCylindricalEqualAreaScaleConversion(
        longitude_natural_origin=2,
        false_easting=3,
        false_northing=4,
        scale_factor_natural_origin=0.999,
    )
    assert lceaop.name == "unknown"
    assert lceaop.method_name == "Lambert Cylindrical Equal Area"
    op_dict = _to_dict(lceaop)
    assert_almost_equal(
        op_dict.pop("Latitude of 1st standard parallel"), 2.57, decimal=2
    )
    assert op_dict == {
        "Longitude of natural origin": 2.0,
        "False easting": 3.0,
        "False northing": 4.0,
    }

@pytest.mark.parametrize(
    "eqc_class", [EquidistantCylindricalConversion, PlateCarreeConversion]
)
def test_equidistant_cylindrical_conversion__defaults(eqc_class):
    eqc = eqc_class()
    assert eqc.name == "unknown"
    assert eqc.method_name == "Equidistant Cylindrical"
    assert _to_dict(eqc) == {
        "Latitude of 1st standard parallel": 0.0,
        "Latitude of natural origin": 0.0,
        "Longitude of natural origin": 0.0,
        "False easting": 0.0,
        "False northing": 0.0,
    }

@pytest.mark.parametrize(
    "eqc_class", [EquidistantCylindricalConversion, PlateCarreeConversion]
)
def test_equidistant_cylindrical_conversion(eqc_class):
    eqc = eqc_class(
        latitude_first_parallel=1.0,
        latitude_natural_origin=2.0,
        longitude_natural_origin=3.0,
        false_easting=4.0,
        false_northing=5.0,
    )
    assert eqc.name == "unknown"
    assert eqc.method_name == "Equidistant Cylindrical"
    assert _to_dict(eqc) == {
        "Latitude of 1st standard parallel": 1.0,
        "Latitude of natural origin": 2.0,
        "Longitude of natural origin": 3.0,
        "False easting": 4.0,
        "False northing": 5.0,
    }

def test_towgs84_transformation():
    transformation = ToWGS84Transformation(GeographicCRS(), 1, 2, 3, 4, 5, 6, 7)
    assert transformation.towgs84 == [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0]
    assert _to_dict(transformation) == {
        "Scale difference": 7.0,
        "X-axis rotation": 4.0,
        "X-axis translation": 1.0,
        "Y-axis rotation": 5.0,
        "Y-axis translation": 2.0,
        "Z-axis rotation": 6.0,
        "Z-axis translation": 3.0,
    }

def test_towgs84_transformation__defaults():
    transformation = ToWGS84Transformation(GeographicCRS())
    assert transformation.towgs84 == [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
    assert _to_dict(transformation) == {
        "Scale difference": 0.0,
        "X-axis rotation": 0.0,
        "X-axis translation": 0.0,
        "Y-axis rotation": 0.0,
        "Y-axis translation": 0.0,
        "Z-axis rotation": 0.0,
        "Z-axis translation": 0.0,
    }

max_stars_repo_name: bjlittle/pyproj

max_issues_repo_name: bjlittle/pyproj

max_forks_repo_name: bjlittle/pyproj


text:
import numpy as np
import scipy as sp
import matplotlib.pyplot as plt

from scipy.sparse.linalg import spsolve
from mpl_toolkits.mplot3d import Axes3D
from matplotlib import cm

import matplotlib.cm
import scipy.sparse

############################  problema 2.4

############### 2.4 Parte 1

'''
d_dt u - d2_dt2 u = 0 ;   (t,x) en (0,infinito)x(0,1)

u(t,x+1) = u(t,x)     ;

u(0,x)   = u_(x)

N_T entero, un paso tempora dt > 0,
N   entero, paso espacial   dx = 1/N

x_j = j dx, j en {0,...,N}
t_n = n dt, n en {0,...,N_T}

obs t_n = N_t*dt

'''

u_00 = lambda x: np.sin(4*np.pi*x)

def calcula_bloque_C(dt,N_T,N,u_0):
    dx = 1/N

    # |{0,...,N}| = N+1 valores
    diagonal_interior  = np.full(N+1, 2*( 1/(dx**2) + 1/dt ) )
    diagonal_exterior = np.full(N,   -1/(dx**2) )

    diagonales = [diagonal_exterior,diagonal_interior,diagonal_exterior]
    offset = [-1,0,1]

    # las dimensiones del bloque son  (0 hasta N) dos veces
    C = sp.sparse.diags(diagonales,offset)
    return C

def calcula_b(dt,N_T,N,u_0):
    dx = 1/N
    b = np.zeros((N+1)*(N_T+1))

    for j in range(0,N+1):
        b[j] = u_0(j*dx)

    return b

def calcula_A(dt,N_T,N,u_0):
    C = calcula_bloque_C(dt,N_T,N,u_0)

    filas = []
    fila_vacía = np.full(N_T+1,None)

    fila_identidad = fila_vacía.copy()
    fila_identidad[0] = np.identity(N+1)

    filas.append(fila_identidad)

    for i in range(0,N_T):
        fila_iter = fila_vacía.copy()
        fila_iter[i]   =  C
        fila_iter[i+1] = -C

        filas.append(fila_iter)

    A = sp.sparse.bmat(filas)

    return A

def des_empaca_u(u,N,N_T):
    u_nuevo = np.zeros([N_T+1,N+1])
    for n in range(0,N_T+1):
        for j in range(0,N+1):
            u_nuevo[n][j] = u[n*(N+1)+j]
    return u_nuevo

def make_grid(N,N_T,dt):
    tt = np.linspace(0,N_T*dt,num=N_T+1,endpoint=True)
    xx = np.linspace(0,1,num=N+1,endpoint=True)

    assert (len(xx)==N+1) and (len(tt)==N_T+1)

    X,T = np.meshgrid(xx,tt)

    return T,X

############### 2.4 Parte 2

'''
Considere ∆x = 0.05 y ∆t ∈ {0.00625,0.025,0.1}.
Calcule la soluci´on num´erica para cada caso y comp´arelas con la
soluci´on real con una gr´aﬁca en T = 0.5.
'''

# dx = 0.05 <-> 1/N = 0.05 <-> N = int(1/0.05)
N = int(1/0.05)
# T = 0.5 <-> N_T*dt = 0.5 <-> N_T = int(0.5/dt)
arr_dt = [0.00625,0.025,0.1]
arr_N_T  = np.divide(0.5*np.ones(len(arr_dt)),arr_dt)

arr_A  = []
arr_b  = []
arr_u  = []
arr_tx = []
arr_U  = []

L = len(arr_dt)

for i in range(0,L):
    dt = arr_dt[i]
    N_T = int(arr_N_T[i])

    A = calcula_A(dt,N_T,N,u_00).tocsr()
    b = calcula_b(dt,N_T,N,u_00)
    u = spsolve(A,b)

    T,X = make_grid(N,N_T,dt)
    U   = des_empaca_u(u,N,N_T)

    arr_A.append(A)
    arr_b.append(b)
    arr_u.append(u)

    arr_tx.append([T,X])
    arr_U.append(U)

gráficos = plt.figure()
ejes = np.full(L,None)

for i in range(0,L):
    ejes[i] = gráficos.add_subplot()

    CS = ejes[i].contourf(arr_tx[i][1] , arr_tx[i][0] , arr_U[i], cmap=cm.coolwarm)

    ejes[i].set_xlabel("variable x")
    ejes[i].set_ylabel("variable t")

gráficos.colorbar(CS)

reason: import numpy,import scipy,from scipy