text:
# Copyright (c) 2021 Oleg Polakow. All rights reserved.
# This code is licensed under Apache 2.0 with Commons Clause license (see LICENSE.md for details)

"""Mixin for building statistics out of performance metrics."""

import inspect
import string
import warnings
from collections import Counter

import numpy as np
import pandas as pd

from vectorbt import _typing as tp
from vectorbt.base.array_wrapper import Wrapping
from vectorbt.utils import checks
from vectorbt.utils.attr_ import get_dict_attr
from vectorbt.utils.config import Config, merge_dicts, get_func_arg_names
from vectorbt.utils.tags import match_tags
from vectorbt.utils.template import deep_substitute

class MetaStatsBuilderMixin(type):
    """Meta class that exposes a read-only class property `StatsBuilderMixin.metrics`."""

    @property
    def metrics(cls) -> Config:
        """Metrics supported by `StatsBuilderMixin.stats`."""
        return cls._metrics

class StatsBuilderMixin(metaclass=MetaStatsBuilderMixin):
    """Mixin that implements `StatsBuilderMixin.stats`.

    Required to be a subclass of `vectorbt.base.array_wrapper.Wrapping`."""

    def __init__(self) -> None:
        checks.assert_instance_of(self, Wrapping)

        # Copy writeable attrs
        self._metrics = self.__class__._metrics.copy()

    @property
    def writeable_attrs(self) -> tp.Set[str]:
        """Set of writeable attributes that will be saved/copied along with the config."""
        return {'_metrics'}

    @property
    def stats_defaults(self) -> tp.Kwargs:
        """Defaults for `StatsBuilderMixin.stats`.

        See `stats_builder` in `vectorbt._settings.settings`."""
        from vectorbt._settings import settings
        stats_builder_cfg = settings['stats_builder']

        return merge_dicts(
            stats_builder_cfg,
            dict(settings=dict(freq=self.wrapper.freq))
        )

    _metrics: tp.ClassVar[Config] = Config(
        dict(
            start=dict(
                title='Start',
                calc_func=lambda self: self.wrapper.index[0],
                agg_func=None,
                tags='wrapper'
            ),
            end=dict(
                title='End',
                calc_func=lambda self: self.wrapper.index[-1],
                agg_func=None,
                tags='wrapper'
            ),
            period=dict(
                title='Period',
                calc_func=lambda self: len(self.wrapper.index),
                apply_to_timedelta=True,
                agg_func=None,
                tags='wrapper'
            )
        ),
        copy_kwargs=dict(copy_mode='deep')
    )

    @property
    def metrics(self) -> Config:
        """Metrics supported by `${cls_name}`.

        ```json
        ${metrics}
        ```

        Returns `${cls_name}._metrics`, which gets (deep) copied upon creation of each instance.
        Thus, changing this config won't affect the class.

        To change metrics, you can either change the config in-place, override this property,
        or overwrite the instance variable `${cls_name}._metrics`."""
        return self._metrics

    def stats(self,
              metrics: tp.Optional[tp.MaybeIterable[tp.Union[str, tp.Tuple[str, tp.Kwargs]]]] = None,
              tags: tp.Optional[tp.MaybeIterable[str]] = None,
              column: tp.Optional[tp.Label] = None,
              group_by: tp.GroupByLike = None,
              agg_func: tp.Optional[tp.Callable] = np.mean,
              silence_warnings: tp.Optional[bool] = None,
              template_mapping: tp.Optional[tp.Mapping] = None,
              settings: tp.KwargsLike = None,
              filters: tp.KwargsLike = None,
              metric_settings: tp.KwargsLike = None) -> tp.Optional[tp.SeriesFrame]:
        """Compute various metrics on this object.

        Args:
            metrics (str, tuple, iterable, or dict): Metrics to calculate.

                Each element can be either:

                * a metric name (see keys in `StatsBuilderMixin.metrics`)
                * a tuple of a metric name and a settings dict as in `StatsBuilderMixin.metrics`.

                The settings dict can contain the following keys:

                * `title`: Title of the metric. Defaults to the name.
                * `tags`: Single or multiple tags to associate this metric with.
                    If any of these tags is in `tags`, keeps this metric.
                * `check_{filter}` and `inv_check_{filter}`: Whether to check this metric against a
                    filter defined in `filters`. True (or False for inverse) means to keep this metric.
                * `calc_func` (required): Calculation function for custom metrics.
                    Should return either a scalar for one column/group, pd.Series for multiple columns/groups,
                    or a dict of such for multiple sub-metrics.
                * `resolve_calc_func`: whether to resolve `calc_func`. If the function can be accessed
                    by traversing attributes of this object, you can specify the path to this function
                    as a string (see `vectorbt.utils.attr_.deep_getattr` for the path format).
                    If `calc_func` is a function, arguments from merged metric settings are matched with
                    arguments in the signature (see below). If `resolve_calc_func` is False, `calc_func`
                    should accept (resolved) self and dictionary of merged metric settings.
                    Defaults to True.
                * `post_calc_func`: Function to post-process the result of `calc_func`.
                    Should accept (resolved) self, output of `calc_func`, and dictionary of merged metric settings,
                    and return whatever is acceptable to be returned by `calc_func`. Defaults to None.
                * `fill_wrap_kwargs`: Whether to fill `wrap_kwargs` with `to_timedelta` and `silence_warnings`.
                    Defaults to False.
                * `apply_to_timedelta`: Whether to apply `vectorbt.base.array_wrapper.ArrayWrapper.to_timedelta`
                    on the result. To disable this globally, pass `to_timedelta=False` in `settings`.
                    Defaults to False.
                * `pass_{arg}`: Whether to pass any argument from the settings (see below). Defaults to True if
                    this argument was found in the function's signature. Set to False to not pass.
                    If argument to be passed was not found, `pass_{arg}` is removed.
                * `resolve_path_{arg}`: Whether to resolve an argument that is meant to be an attribute of
                    this object and is the first part of the path of `calc_func`. Passes only optional arguments.
                    Defaults to True. See `vectorbt.utils.attr_.AttrResolver.resolve_attr`.
                * `resolve_{arg}`: Whether to resolve an argument that is meant to be an attribute of
                    this object and is present in the function's signature. Defaults to False.
                    See `vectorbt.utils.attr_.AttrResolver.resolve_attr`.
                * `template_mapping`: Mapping to replace templates in metric settings. Used across all settings.
                * Any other keyword argument that overrides the settings or is passed directly to `calc_func`.

                If `resolve_calc_func` is True, the calculation function may "request" any of the
                following arguments by accepting them or if `pass_{arg}` was found in the settings dict:

                * Each of `vectorbt.utils.attr_.AttrResolver.self_aliases`: original object
                    (ungrouped, with no column selected)
                * `group_by`: won't be passed if it was used in resolving the first attribute of `calc_func`
                    specified as a path, use `pass_group_by=True` to pass anyway
                * `column`
                * `metric_name`
                * `agg_func`
                * `silence_warnings`
                * `to_timedelta`: replaced by True if None and frequency is set
                * Any argument from `settings`
                * Any attribute of this object if it meant to be resolved
                    (see `vectorbt.utils.attr_.AttrResolver.resolve_attr`)

                Pass `metrics='all'` to calculate all supported metrics.
            tags (str or iterable): Tags to select.

                See `vectorbt.utils.tags.match_tags`.
            column (str): Name of the column/group.

                !!! hint
                    There are two ways to select a column: `obj['a'].stats()` and `obj.stats(column='a')`.
                    They both accomplish the same thing but in different ways: `obj['a'].stats()` computes
                    statistics of the column 'a' only, while `obj.stats(column='a')` computes statistics of
                    all columns first and only then selects the column 'a'. The first method is preferred
                    when you have a lot of data or caching is disabled. The second method is preferred when
                    most attributes have already been cached.
            group_by (any): Group or ungroup columns. See `vectorbt.base.column_grouper.ColumnGrouper`.
            agg_func (callable): Aggregation function to aggregate statistics across all columns.
                Defaults to mean.

                Should take `pd.Series` and return a const.

                Has only effect if `column` was specified or this object contains only one column of data.

                If `agg_func` has been overridden by a metric:

                * it only takes effect if global `agg_func` is not None
                * will raise a warning if it's None but the result of calculation has multiple values
            silence_warnings (bool): Whether to silence all warnings.
            template_mapping (mapping): Global mapping to replace templates.

                Gets merged over `template_mapping` from `StatsBuilderMixin.stats_defaults`.

                Applied on `settings` and then on each metric settings.
            filters (dict): Filters to apply.

                Each item consists of the filter name and settings dict.

                The settings dict can contain the following keys:

                * `filter_func`: Filter function that should accept resolved self and
                    merged settings for a metric, and return either True or False.
                * `warning_message`: Warning message to be shown when skipping a metric.
                    Can be a template that will be substituted using merged metric settings as mapping.
                    Defaults to None.
                * `inv_warning_message`: Same as `warning_message` but for inverse checks.

                Gets merged over `filters` from `StatsBuilderMixin.stats_defaults`.
            settings (dict): Global settings and resolution arguments.

                Extends/overrides `settings` from `StatsBuilderMixin.stats_defaults`.
                Gets extended/overridden by metric settings.
            metric_settings (dict): Keyword arguments for each metric.

                Extends/overrides all global and metric settings.

        For template logic, see `vectorbt.utils.template`.

        For defaults, see `StatsBuilderMixin.stats_defaults`.

        !!! hint
            There are two types of arguments: optional (or resolution) and mandatory arguments.
            Optional arguments are only passed if they are found in the function's signature.
            Mandatory arguments are passed regardless of this. Optional arguments can only be defined
            using `settings` (that is, globally), while mandatory arguments can be defined both using
            default metric settings and `{metric_name}_kwargs`. Overriding optional arguments using default
            metric settings or `{metric_name}_kwargs` won't turn them into mandatory. For this, pass `pass_{arg}=True`.

        !!! hint
            Make sure to resolve and then to re-use as many object attributes as possible to
            utilize built-in caching (even if global caching is disabled).

        Usage:
            See `vectorbt.portfolio.base` for examples.
        """
        # Resolve defaults
        if silence_warnings is None:
            silence_warnings = self.stats_defaults.get('silence_warnings', False)
        template_mapping = merge_dicts(self.stats_defaults.get('template_mapping', {}), template_mapping)
        filters = merge_dicts(self.stats_defaults.get('filters', {}), filters)
        settings = merge_dicts(self.stats_defaults.get('settings', {}), settings)
        metric_settings = merge_dicts(self.stats_defaults.get('metric_settings', {}), metric_settings)

        # Replace templates globally (not used at metric level)
        if len(template_mapping) > 0:
            sub_settings = deep_substitute(settings, mapping=template_mapping)
        else:
            sub_settings = settings

        # Resolve self
        reself = self.resolve_self(
            cond_kwargs=sub_settings,
            impacts_caching=False,
            silence_warnings=silence_warnings
        )

        # Prepare metrics
        if metrics is None:
            metrics = reself.stats_defaults.get('metrics', 'all')
        if metrics == 'all':
            metrics = reself.metrics
        if isinstance(metrics, dict):
            metrics = list(metrics.items())
        if isinstance(metrics, (str, tuple)):
            metrics = [metrics]

        # Prepare tags
        if tags is None:
            tags = reself.stats_defaults.get('tags', 'all')
        if isinstance(tags, str) and tags == 'all':
            tags = None
        if isinstance(tags, (str, tuple)):
            tags = [tags]

        # Bring to the same shape
        new_metrics = []
        for i, metric in enumerate(metrics):
            if isinstance(metric, str):
                metric = (metric, reself.metrics[metric])
            if not isinstance(metric, tuple):
                raise TypeError(f"Metric at index {i} must be either a string or a tuple")
            new_metrics.append(metric)
        metrics = new_metrics

        # Handle duplicate names
        metric_counts = Counter(list(map(lambda x: x[0], metrics)))
        metric_i = {k: -1 for k in metric_counts.keys()}
        metrics_dct = {}
        for i, (metric_name, _metric_settings) in enumerate(metrics):
            if metric_counts[metric_name] > 1:
                metric_i[metric_name] += 1
                metric_name = metric_name + '_' + str(metric_i[metric_name])
            metrics_dct[metric_name] = _metric_settings

        # Check metric_settings
        missed_keys = set(metric_settings.keys()).difference(set(metrics_dct.keys()))
        if len(missed_keys) > 0:
            raise ValueError(f"Keys {missed_keys} in metric_settings could not be matched with any metric")

        # Merge settings
        opt_arg_names_dct = {}
        custom_arg_names_dct = {}
        resolved_self_dct = {}
        mapping_dct = {}
        for metric_name, _metric_settings in list(metrics_dct.items()):
            opt_settings = merge_dicts(
                {name: reself for name in reself.self_aliases},
                dict(
                    column=column,
                    group_by=group_by,
                    metric_name=metric_name,
                    agg_func=agg_func,
                    silence_warnings=silence_warnings,
                    to_timedelta=None
                ),
                settings
            )
            _metric_settings = _metric_settings.copy()
            passed_metric_settings = metric_settings.get(metric_name, {})
            merged_settings = merge_dicts(
                opt_settings,
                _metric_settings,
                passed_metric_settings
            )
            metric_template_mapping = merged_settings.pop('template_mapping', {})
            template_mapping_merged = merge_dicts(template_mapping, metric_template_mapping)
            template_mapping_merged = deep_substitute(template_mapping_merged, mapping=merged_settings)
            mapping = merge_dicts(template_mapping_merged, merged_settings)
            merged_settings = deep_substitute(merged_settings, mapping=mapping)

            # Filter by tag
            if tags is not None:
                in_tags = merged_settings.get('tags', None)
                if in_tags is None or not match_tags(tags, in_tags):
                    metrics_dct.pop(metric_name, None)
                    continue

            custom_arg_names = set(_metric_settings.keys()).union(set(passed_metric_settings.keys()))
            opt_arg_names = set(opt_settings.keys())
            custom_reself = reself.resolve_self(
                cond_kwargs=merged_settings,
                custom_arg_names=custom_arg_names,
                impacts_caching=True,
                silence_warnings=merged_settings['silence_warnings']
            )

            metrics_dct[metric_name] = merged_settings
            custom_arg_names_dct[metric_name] = custom_arg_names
            opt_arg_names_dct[metric_name] = opt_arg_names
            resolved_self_dct[metric_name] = custom_reself
            mapping_dct[metric_name] = mapping

        # Filter metrics
        for metric_name, _metric_settings in list(metrics_dct.items()):
            custom_reself = resolved_self_dct[metric_name]
            mapping = mapping_dct[metric_name]
            _silence_warnings = _metric_settings.get('silence_warnings')

            metric_filters = set()
            for k in _metric_settings.keys():
                filter_name = None
                if k.startswith('check_'):
                    filter_name = k[len('check_'):]
                elif k.startswith('inv_check_'):
                    filter_name = k[len('inv_check_'):]
                if filter_name is not None:
                    if filter_name not in filters:
                        raise ValueError(f"Metric '{metric_name}' requires filter '{filter_name}'")
                    metric_filters.add(filter_name)

            for filter_name in metric_filters:
                filter_settings = filters[filter_name]
                _filter_settings = deep_substitute(filter_settings, mapping=mapping)
                filter_func = _filter_settings['filter_func']
                warning_message = _filter_settings.get('warning_message', None)
                inv_warning_message = _filter_settings.get('inv_warning_message', None)
                to_check = _metric_settings.get('check_' + filter_name, False)
                inv_to_check = _metric_settings.get('inv_check_' + filter_name, False)

                if to_check or inv_to_check:
                    whether_true = filter_func(custom_reself, _metric_settings)
                    to_remove = (to_check and not whether_true) or (inv_to_check and whether_true)
                    if to_remove:
                        if to_check and warning_message is not None and not _silence_warnings:
                            warnings.warn(warning_message)
                        if inv_to_check and inv_warning_message is not None and not _silence_warnings:
                            warnings.warn(inv_warning_message)

                        metrics_dct.pop(metric_name, None)
                        custom_arg_names_dct.pop(metric_name, None)
                        opt_arg_names_dct.pop(metric_name, None)
                        resolved_self_dct.pop(metric_name, None)
                        mapping_dct.pop(metric_name, None)
                        break

        # Any metrics left?
        if len(metrics_dct) == 0:
            if not silence_warnings:
                warnings.warn("No metrics to calculate", stacklevel=2)
            return None

        # Compute stats
        arg_cache_dct = {}
        stats_dct = {}
        used_agg_func = False
        for i, (metric_name, _metric_settings) in enumerate(metrics_dct.items()):
            try:
                final_kwargs = _metric_settings.copy()
                opt_arg_names = opt_arg_names_dct[metric_name]
                custom_arg_names = custom_arg_names_dct[metric_name]
                custom_reself = resolved_self_dct[metric_name]

                # Clean up keys
                for k, v in list(final_kwargs.items()):
                    if k.startswith('check_') or k.startswith('inv_check_') or k in ('tags',):
                        final_kwargs.pop(k, None)

                # Get metric-specific values
                _column = final_kwargs.get('column')
                _group_by = final_kwargs.get('group_by')
                _agg_func = final_kwargs.get('agg_func')
                _silence_warnings = final_kwargs.get('silence_warnings')
                if final_kwargs['to_timedelta'] is None:
                    final_kwargs['to_timedelta'] = custom_reself.wrapper.freq is not None
                to_timedelta = final_kwargs.get('to_timedelta')
                title = final_kwargs.pop('title', metric_name)
                calc_func = final_kwargs.pop('calc_func')
                resolve_calc_func = final_kwargs.pop('resolve_calc_func', True)
                post_calc_func = final_kwargs.pop('post_calc_func', None)
                use_caching = final_kwargs.pop('use_caching', True)
                fill_wrap_kwargs = final_kwargs.pop('fill_wrap_kwargs', False)
                if fill_wrap_kwargs:
                    final_kwargs['wrap_kwargs'] = merge_dicts(
                        dict(to_timedelta=to_timedelta, silence_warnings=_silence_warnings),
                        final_kwargs.get('wrap_kwargs', None)
                    )
                apply_to_timedelta = final_kwargs.pop('apply_to_timedelta', False)

                # Resolve calc_func
                if resolve_calc_func:
                    if not callable(calc_func):
                        passed_kwargs_out = {}

                        def _getattr_func(obj: tp.Any,
                                          attr: str,
                                          args: tp.ArgsLike = None,
                                          kwargs: tp.KwargsLike = None,
                                          call_attr: bool = True,
                                          _final_kwargs: tp.Kwargs = final_kwargs,
                                          _opt_arg_names: tp.Set[str] = opt_arg_names,
                                          _custom_arg_names: tp.Set[str] = custom_arg_names,
                                          _arg_cache_dct: tp.Kwargs = arg_cache_dct) -> tp.Any:
                            if attr in final_kwargs:
                                return final_kwargs[attr]
                            if args is None:
                                args = ()
                            if kwargs is None:
                                kwargs = {}
                            if obj is custom_reself and _final_kwargs.pop('resolve_path_' + attr, True):
                                if call_attr:
                                    return custom_reself.resolve_attr(
                                        attr,
                                        args=args,
                                        cond_kwargs={k: v for k, v in _final_kwargs.items() if k in _opt_arg_names},
                                        kwargs=kwargs,
                                        custom_arg_names=_custom_arg_names,
                                        cache_dct=_arg_cache_dct,
                                        use_caching=use_caching,
                                        passed_kwargs_out=passed_kwargs_out
                                    )
                                return getattr(obj, attr)
                            out = getattr(obj, attr)
                            if callable(out) and call_attr:
                                return out(*args, **kwargs)
                            return out

                        calc_func = custom_reself.deep_getattr(
                            calc_func,
                            getattr_func=_getattr_func,
                            call_last_attr=False
                        )

                        if 'group_by' in passed_kwargs_out:
                            if 'pass_group_by' not in final_kwargs:
                                final_kwargs.pop('group_by', None)

                    # Resolve arguments
                    if callable(calc_func):
                        func_arg_names = get_func_arg_names(calc_func)
                        for k in func_arg_names:
                            if k not in final_kwargs:
                                if final_kwargs.pop('resolve_' + k, False):
                                    try:
                                        arg_out = custom_reself.resolve_attr(
                                            k,
                                            cond_kwargs=final_kwargs,
                                            custom_arg_names=custom_arg_names,
                                            cache_dct=arg_cache_dct,
                                            use_caching=use_caching
                                        )
                                    except AttributeError:
                                        continue
                                    final_kwargs[k] = arg_out
                        for k in list(final_kwargs.keys()):
                            if k in opt_arg_names:
                                if 'pass_' + k in final_kwargs:
                                    if not final_kwargs.get('pass_' + k):  # first priority
                                        final_kwargs.pop(k, None)
                                elif k not in func_arg_names:  # second priority
                                    final_kwargs.pop(k, None)
                        for k in list(final_kwargs.keys()):
                            if k.startswith('pass_') or k.startswith('resolve_'):
                                final_kwargs.pop(k, None)  # cleanup

                        # Call calc_func
                        out = calc_func(**final_kwargs)
                    else:
                        # calc_func is already a result
                        out = calc_func
                else:
                    # Do not resolve calc_func
                    out = calc_func(custom_reself, _metric_settings)

                # Call post_calc_func
                if post_calc_func is not None:
                    out = post_calc_func(custom_reself, out, _metric_settings)

                # Post-process and store the metric
                multiple = True
                if not isinstance(out, dict):
                    multiple = False
                    out = {None: out}
                for k, v in out.items():
                    # Resolve title
                    if multiple:
                        if title is None:
                            t = str(k)
                        else:
                            t = title + ': ' + str(k)
                    else:
                        t = title

                    # Check result type
                    if checks.is_any_array(v) and not checks.is_series(v):
                        raise TypeError("calc_func must return either a scalar for one column/group, "
                                        "pd.Series for multiple columns/groups, or a dict of such. "
                                        f"Not {type(v)}.")

                    # Handle apply_to_timedelta
                    if apply_to_timedelta and to_timedelta:
                        v = custom_reself.wrapper.to_timedelta(v, silence_warnings=_silence_warnings)

                    # Select column or aggregate
                    if checks.is_series(v):
                        if _column is not None:
                            v = custom_reself.select_one_from_obj(
                                v, custom_reself.wrapper.regroup(_group_by), column=_column)
                        elif _agg_func is not None and agg_func is not None:
                            v = _agg_func(v)
                            used_agg_func = True
                        elif _agg_func is None and agg_func is not None:
                            if not _silence_warnings:
                                warnings.warn(f"Metric '{metric_name}' returned multiple values "
                                              f"despite having no aggregation function", stacklevel=2)
                            continue

                    # Store metric
                    if t in stats_dct:
                        if not _silence_warnings:
                            warnings.warn(f"Duplicate metric title '{t}'", stacklevel=2)
                    stats_dct[t] = v
            except Exception as e:
                warnings.warn(f"Metric '{metric_name}' raised an exception", stacklevel=2)
                raise e

        # Return the stats
        if reself.wrapper.get_ndim(group_by=group_by) == 1:
            return pd.Series(stats_dct, name=reself.wrapper.get_name(group_by=group_by))
        if column is not None:
            return pd.Series(stats_dct, name=column)
        if agg_func is not None:
            if used_agg_func and not silence_warnings:
                warnings.warn(f"Object has multiple columns. Aggregating using {agg_func}. "
                              f"Pass column to select a single column/group.", stacklevel=2)
            return pd.Series(stats_dct, name='agg_func_' + agg_func.__name__)
        new_index = reself.wrapper.grouper.get_columns(group_by=group_by)
        stats_df = pd.DataFrame(stats_dct, index=new_index)
        return stats_df

    # ############# Docs ############# #

    @classmethod
    def build_metrics_doc(cls, source_cls: tp.Optional[type] = None) -> str:
        """Build metrics documentation."""
        if source_cls is None:
            source_cls = StatsBuilderMixin
        return string.Template(
            inspect.cleandoc(get_dict_attr(source_cls, 'metrics').__doc__)
        ).substitute(
            {'metrics': cls.metrics.to_doc(), 'cls_name': cls.__name__}
        )

    @classmethod
    def override_metrics_doc(cls, __pdoc__: dict, source_cls: tp.Optional[type] = None) -> None:
        """Call this method on each subclass that overrides `metrics`."""
        __pdoc__[cls.__name__ + '.metrics'] = cls.build_metrics_doc(source_cls=source_cls)

__pdoc__ = dict()
StatsBuilderMixin.override_metrics_doc(__pdoc__)

max_stars_repo_name: lmservas/vectorbt

max_issues_repo_name: Reed-Schimmel/vectorbt

max_forks_repo_name: Reed-Schimmel/vectorbt


text:
"""
Code borrowed from Xinshuo_PyToolbox: https://github.com/xinshuoweng/Xinshuo_PyToolbox
"""

import os
import shutil
import torch
import numpy as np
import random
import time
import copy
import glob, glob2
from torch import nn

class AverageMeter(object):
  """Computes and stores the average and current value"""
  def __init__(self):
    self.reset()

  def reset(self):
    self.val = 0
    self.avg = 0
    self.sum = 0
    self.count = 0
    self.list = list()

  def update(self, val, n=1):
    self.val = val
    self.sum += val * n
    self.count += n
    self.avg = self.sum / self.count
    self.list.append(val)

def isnparray(nparray_test):
	return isinstance(nparray_test, np.ndarray)

def isinteger(integer_test):
	if isnparray(integer_test): return False
	try: return isinstance(integer_test, int) or int(integer_test) == integer_test
	except ValueError: return False
	except TypeError: return False

def isfloat(float_test):
	return isinstance(float_test, float)

def isscalar(scalar_test):
	try: return isinteger(scalar_test) or isfloat(scalar_test)
	except TypeError: return False

def islogical(logical_test):
	return isinstance(logical_test, bool)

def isstring(string_test):
    return isinstance(string_test, str)

def islist(list_test):
	return isinstance(list_test, list)

def convert_secs2time(seconds):
    '''
    format second to human readable way
    '''
    assert isscalar(seconds), 'input should be a scalar to represent number of seconds'
    m, s = divmod(int(seconds), 60)
    h, m = divmod(m, 60)
    return '[%d:%02d:%02d]' % (h, m, s)

def get_timestring():
    return time.strftime('%Y%m%d_%Hh%Mm%Ss')

def recreate_dirs(*dirs):
    for d in dirs:
        if os.path.exists(d):
            shutil.rmtree(d)
        os.makedirs(d)

def is_path_valid(pathname):
	try:
		if not isstring(pathname) or not pathname: return False
	except TypeError: return False
	else: return True

def is_path_creatable(pathname):
	'''
	if any previous level of parent folder exists, returns true
	'''
	if not is_path_valid(pathname): return False
	pathname = os.path.normpath(pathname)
	pathname = os.path.dirname(os.path.abspath(pathname))

	# recursively to find the previous level of parent folder existing
	while not is_path_exists(pathname):
		pathname_new = os.path.dirname(os.path.abspath(pathname))
		if pathname_new == pathname: return False
		pathname = pathname_new
	return os.access(pathname, os.W_OK)

def is_path_exists(pathname):
	try: return is_path_valid(pathname) and os.path.exists(pathname)
	except OSError: return False

def is_path_exists_or_creatable(pathname):
	try: return is_path_exists(pathname) or is_path_creatable(pathname)
	except OSError: return False

def isfile(pathname):
	if is_path_valid(pathname):
		pathname = os.path.normpath(pathname)
		name = os.path.splitext(os.path.basename(pathname))[0]
		ext = os.path.splitext(pathname)[1]
		return len(name) > 0 and len(ext) > 0
	else: return False

def isfolder(pathname):
	'''
	if '.' exists in the subfolder, the function still justifies it as a folder. e.g., /mnt/dome/adhoc_0.5x/abc is a folder
	if '.' exists after all slashes, the function will not justify is as a folder. e.g., /mnt/dome/adhoc_0.5x is NOT a folder
	'''
	if is_path_valid(pathname):
		pathname = os.path.normpath(pathname)
		if pathname == './': return True
		name = os.path.splitext(os.path.basename(pathname))[0]
		ext = os.path.splitext(pathname)[1]
		return len(name) > 0 and len(ext) == 0
	else: return False

def mkdir_if_missing(input_path):
    folder = input_path if isfolder(input_path) else os.path.dirname(input_path)
    os.makedirs(folder, exist_ok=True)

def safe_list(input_data, warning=True, debug=True):
	'''
	copy a list to the buffer for use
	parameters:
		input_data:		a list
	outputs:
		safe_data:		a copy of input data
	'''
	if debug: assert islist(input_data), 'the input data is not a list'
	safe_data = copy.copy(input_data)
	return safe_data

def safe_path(input_path, warning=True, debug=True):
    '''
    convert path to a valid OS format, e.g., empty string '' to '.', remove redundant '/' at the end from 'aa/' to 'aa'
    parameters:
    	input_path:		a string
    outputs:
    	safe_data:		a valid path in OS format
    '''
    if debug: assert isstring(input_path), 'path is not a string: %s' % input_path
    safe_data = copy.copy(input_path)
    safe_data = os.path.normpath(safe_data)
    return safe_data

def prepare_seed(rand_seed):
	np.random.seed(rand_seed)
	random.seed(rand_seed)
	torch.manual_seed(rand_seed)
	torch.cuda.manual_seed_all(rand_seed)

def initialize_weights(modules):
    for m in modules:
        if isinstance(m, nn.Conv2d):
            nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
            if m.bias is not None: nn.init.constant_(m.bias, 0)
        elif isinstance(m, nn.BatchNorm2d):
            nn.init.constant_(m.weight, 1)
            if m.bias is not None: nn.init.constant_(m.bias, 0)
        elif isinstance(m, nn.Linear):
            nn.init.normal_(m.weight, 0, 0.01)
            if m.bias is not None: nn.init.constant_(m.bias, 0)

def print_log(print_str, log, same_line=False, display=True):
	'''
	print a string to a log file

	parameters:
		print_str:          a string to print
		log:                a opened file to save the log
		same_line:          True if we want to print the string without a new next line
		display:            False if we want to disable to print the string onto the terminal
	'''
	if display:
		if same_line: print('{}'.format(print_str), end='')
		else: print('{}'.format(print_str))

	if same_line: log.write('{}'.format(print_str))
	else: log.write('{}\n'.format(print_str))
	log.flush()

def find_unique_common_from_lists(input_list1, input_list2, warning=True, debug=True):
	'''
	find common items from 2 lists, the returned elements are unique. repetitive items will be ignored
	if the common items in two elements are not in the same order, the outputs follows the order in the first list

	parameters:
		input_list1, input_list2:		two input lists

	outputs:
		list_common:	a list of elements existing both in list_src1 and list_src2
		index_list1:	a list of index that list 1 has common items
		index_list2:	a list of index that list 2 has common items
	'''
	input_list1 = safe_list(input_list1, warning=warning, debug=debug)
	input_list2 = safe_list(input_list2, warning=warning, debug=debug)

	common_list = list(set(input_list1).intersection(input_list2))

	# find index
	index_list1 = []
	for index in range(len(input_list1)):
		item = input_list1[index]
		if item in common_list:
			index_list1.append(index)

	index_list2 = []
	for index in range(len(input_list2)):
		item = input_list2[index]
		if item in common_list:
			index_list2.append(index)

	return common_list, index_list1, index_list2

def load_txt_file(file_path, debug=True):
    '''
    load data or string from text file
    '''
    file_path = safe_path(file_path)
    if debug: assert is_path_exists(file_path), 'text file is not existing at path: %s!' % file_path
    with open(file_path, 'r') as file: data = file.read().splitlines()
    num_lines = len(data)
    file.close()
    return data, num_lines

def load_list_from_folder(folder_path, ext_filter=None, depth=1, recursive=False, sort=True, save_path=None, debug=True):
    '''
    load a list of files or folders from a system path

    parameters:
        folder_path:    root to search
        ext_filter:     a string to represent the extension of files interested
        depth:          maximum depth of folder to search, when it's None, all levels of folders will be searched
        recursive:      False: only return current level
                        True: return all levels till to the input depth

    outputs:
        fulllist:       a list of elements
        num_elem:       number of the elements
    '''
    folder_path = safe_path(folder_path)
    if debug: assert isfolder(folder_path), 'input folder path is not correct: %s' % folder_path
    if not is_path_exists(folder_path):
        print('the input folder does not exist\n')
        return [], 0
    if debug:
        assert islogical(recursive), 'recursive should be a logical variable: {}'.format(recursive)
        assert depth is None or (isinteger(depth) and depth >= 1), 'input depth is not correct {}'.format(depth)
        assert ext_filter is None or (islist(ext_filter) and all(isstring(ext_tmp) for ext_tmp in ext_filter)) or isstring(ext_filter), 'extension filter is not correct'
    if isstring(ext_filter): ext_filter = [ext_filter]                               # convert to a list
    # zxc

    fulllist = list()
    if depth is None:        # find all files recursively
        recursive = True
        wildcard_prefix = '**'
        if ext_filter is not None:
            for ext_tmp in ext_filter:
                # wildcard = os.path.join(wildcard_prefix, '*' + string2ext_filter(ext_tmp))
                wildcard = os.path.join(wildcard_prefix, '*' + ext_tmp)
                curlist = glob2.glob(os.path.join(folder_path, wildcard))
                if sort: curlist = sorted(curlist)
                fulllist += curlist
        else:
            wildcard = wildcard_prefix
            curlist = glob2.glob(os.path.join(folder_path, wildcard))
            if sort: curlist = sorted(curlist)
            fulllist += curlist
    else:                    # find files based on depth and recursive flag
        wildcard_prefix = '*'
        for index in range(depth-1): wildcard_prefix = os.path.join(wildcard_prefix, '*')
        if ext_filter is not None:
            for ext_tmp in ext_filter:
                # wildcard = wildcard_prefix + string2ext_filter(ext_tmp)
                wildcard = wildcard_prefix + ext_tmp
                curlist = glob.glob(os.path.join(folder_path, wildcard))
                if sort: curlist = sorted(curlist)
                fulllist += curlist
            # zxc
        else:
            wildcard = wildcard_prefix
            curlist = glob.glob(os.path.join(folder_path, wildcard))
            # print(curlist)
            if sort: curlist = sorted(curlist)
            fulllist += curlist
        if recursive and depth > 1:
            newlist, _ = load_list_from_folder(folder_path=folder_path, ext_filter=ext_filter, depth=depth-1, recursive=True)
            fulllist += newlist

    fulllist = [os.path.normpath(path_tmp) for path_tmp in fulllist]
    num_elem = len(fulllist)

    # save list to a path
    if save_path is not None:
        save_path = safe_path(save_path)
        if debug: assert is_path_exists_or_creatable(save_path), 'the file cannot be created'
        with open(save_path, 'w') as file:
            for item in fulllist: file.write('%s\n' % item)
        file.close()

    return fulllist, num_elem

ext: py

max_stars_repo_name: kikacaty/AgentFormer

max_issues_repo_name: kikacaty/AgentFormer

max_forks_repo_name: kikacaty/AgentFormer

reason: import numpy


text:
import abc
from collections import defaultdict
import contextlib
import functools
from logging import getLogger

import numpy as np
import tensorflow as tf

logger = getLogger("rl")

def lazy_function(function):
    attribute = '_cache_' + function.__name__

    @functools.wraps(function)
    def decorator(self):
        if not hasattr(self, attribute):
            setattr(self, attribute, function(self))
        return getattr(self, attribute)
    return decorator

def purge_orphaned_summaries(summary_writer, step):
  summary_writer.add_session_log(
    tf.SessionLog(status=tf.SessionLog.START), step)

def read_events(event_filename, data=None, purge_orphaned=True):
  data = data or defaultdict(dict)
  for e in tf.train.summary_iterator(event_filename):
    if purge_orphaned and e.session_log.status == tf.SessionLog.START:
      for tag in data.keys():
        data[tag] = {
            step_time: val
            for step_time, val in data[tag].items()
            if step_time[0] < e.step
        }

    for v in e.summary.value:
      data[v.tag][(e.step, e.wall_time)] = v.simple_value
  return data

def orthogonal_initializer(scale=1.0):
  # taken from https://github.com/openai/baselines/tree/master/baselines/ppo2
  def _initializer(shape, dtype, partition_info=None):
    shape = tuple(shape)
    if len(shape) == 2:
      flat_shape = shape
    elif len(shape) == 4:  # assumes NHWC
      flat_shape = (np.prod(shape[:-1]), shape[-1])
    else:
      raise NotImplementedError("Not supported shape: {}".format(shape))
    a = np.random.normal(0.0, 1.0, flat_shape)
    u, _, v = np.linalg.svd(a, full_matrices=False)
    q = u if u.shape == flat_shape else v
    q = q.reshape(shape)
    return (scale * q[:shape[0], :shape[1]]).astype(np.float32)
  return _initializer

def normalized_columns_initializer(stddev=1.0):
  def _initializer(shape, dtype=None, partition_info=None):
    out = np.random.randn(*shape).astype(np.float32)
    out *= stddev / np.sqrt(np.square(out).sum(axis=0, keepdims=True))
    return tf.constant(out)
  return _initializer

def torch_default_initializer():
  """Default initializer in torch.

   The initializer is similar to the He uniform initializer as
   described in  http://arxiv.org/abs/1502.01852, but uses
   different value in numerator.

   https://github.com/torch/nn/blob/master/Linear.lua#L21
   https://github.com/torch/nn/blob/master/SpatialConvolution.lua#L34
  """
  def _initializer(shape, dtype=None, partition_info=None):
    shape = tuple(shape)
    fan_in = np.prod(shape[:-1])
    stddev = 1. / np.sqrt(fan_in)
    return tf.random_uniform(shape, minval=-stddev, maxval=stddev, dtype=dtype)
  return _initializer

def partial_restore(variables, checkpoint, session=None, scope=None):
  meta_file = checkpoint + ".meta"
  with tf.Graph().as_default(), tf.Session() as sess, sess.as_default():
    saver = tf.train.import_meta_graph(meta_file)
    saver.restore(sess, checkpoint)
    checkpoint_vars = {
        v.name: v.eval()
        for v in tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES,
                                   scope=scope)
    }
  for v in variables:
    v.load(checkpoint_vars[v.name])

def explained_variance(targets, predictions):
  targets, predictions = tf.squeeze(targets), tf.squeeze(predictions)
  tf.assert_rank_in(targets, [0, 1])
  tf.assert_rank_in(predictions, [0, 1])
  var_targets = tf.cond(
      tf.equal(tf.rank(targets), 0),
      lambda: tf.constant(0, dtype=tf.float32),
      lambda: tf.nn.moments(targets, axes=[0])[1]
  )
  return tf.cond(
      tf.equal(var_targets, 0),
      lambda: tf.constant(np.nan),
      lambda: (1
               - tf.nn.moments(targets - predictions, axes=[0])[1]
               / var_targets)
  )

def huber_loss(x, delta=1.0):
  with tf.variable_scope("huber_loss"):
    abs_x = tf.abs(x)
    return tf.where(
        abs_x < delta,
        0.5 * tf.square(x),
        delta * (abs_x - 0.5 * delta)
    )

@contextlib.contextmanager
def variable_and_name_scopes(scope):
  """ Restores both variable and name scopes.

    Both scopes might need to be restored to allow continuation of building
    process. Restoring only variable scope might lead to the creation of
    new name scope which is not desirable in some cases.
  """
  # Manually reopen name_scope inside variable_scope of
  # the algorithm. See https://github.com/tensorflow/tensorflow/issues/6189.
  # This is also the way it is done in tf.layers.base.
  with tf.variable_scope(scope) as restored,\
      tf.name_scope(scope.original_name_scope):
    yield restored

def scoped(func):
  """ Decorator that adds variable and name scope for function of a class.

  The class need to have `scope` property of type `tf.VariableScope`. This
  variable scope will be restored together with the underlying name scope.
  """
  def _scoped(self, *args, **kwargs):
    with variable_and_name_scopes(self.scope):
      return func(self, *args, **kwargs)
  return _scoped

class NetworkStructure(abc.ABC):
  """ Conceptually separate structure in tensorflow graph. """
  def __init__(self, name=None):
    self._is_built = False
    self._name = name or self.__class__.__name__
    with tf.variable_scope(self._name) as captured_scope:
      self._scope = captured_scope
    self._before_build_hooks = []
    self._after_build_hooks = []

  @property
  def scope(self):
    return self._scope

  @property
  def is_built(self):
    return self._is_built

  def add_before_build_hook(self, hook):
    self._before_build_hooks.append(hook)

  def add_after_build_hook(self, hook):
    self._after_build_hooks.append(hook)

  @scoped
  def build(self, *args, **kwargs):
    if not self.is_built:
      logger.info("Building {}".format(self._name))
      for hook in self._before_build_hooks:
        hook(self, *args, **kwargs)
      self._build(*args, **kwargs)
      self._is_built = True
      for hook in self._after_build_hooks:
        hook(self, *args, **kwargs)

  @abc.abstractmethod
  def _build(self):
    ...

max_stars_repo_name: MichaelKonobeev/rl

max_issues_repo_name: MichaelKonobeev/rl

max_forks_repo_name: MichaelKonobeev/rl


text:
import itertools
import warnings
from numbers import Number

import numpy as np

from ..abc import SparseArray
from ..utils import NDArrayReprMixin
from ..utils import is_broadcastable
from ..utils import ravel_sparse_multi_index
from ..utils import unravel_sparse_index

__all__ = ['CoordinateArray']

class CoordinateArray(
    np.lib.mixins.NDArrayOperatorsMixin, NDArrayReprMixin, SparseArray
):
    # is there a real advantage to idxs *before* data? if so change order
    def __init__(
        self, data, idxs, shape=None, fill_value=None, dtype=None,
        copy=False, sum_duplicates=True
    ):
        multi_idx = np.atleast_2d(np.array(idxs, dtype=np.uint, copy=copy))
        if multi_idx.ndim != 2:
            raise ValueError(f"'idxs' must be 2d, not {multi_idx.ndim}d")

        try:
            min_shape = multi_idx.max(axis=1) + 1
        except ValueError:
            min_shape = ()

        if shape is None:
            self._shape = tuple(min_shape)
        else:
            shape = np.broadcast_arrays(shape).pop().astype(np.uint).ravel()
            if len(shape) != len(min_shape):
                raise ValueError(f"'shape' length does not match 'idxs' length")
            elif np.any(shape < min_shape):
                raise ValueError(f"'shape' {tuple(shape)} values are too small for 'idxs'")
            else:
                self._shape = tuple(shape)

        self._idxs = ravel_sparse_multi_index(multi_idx, self.shape)

        #TODO: if shape=() then data should be scalar array of value = fill_value
        if not np.iterable(data):
            data = tuple(itertools.repeat(data, self._idxs.shape[-1]))

        self._data = np.array(data, dtype=dtype, copy=copy)
        if self.data.ndim != 1:
            raise ValueError(f"'data' must be 1d, not {self.data.ndim}d")

        if self.data.shape[-1] != self.idxs.shape[-1]:
            raise ValueError("'data' does not have 1-1 correspondence with 'idxs'")

        if fill_value is None:
            fill_value = np.zeros((), self.dtype).item()
        self.fill_value = fill_value

        if sum_duplicates:
            self.sum_duplicates()
        else:
            self.sort_index()

    def __repr__(self):
        data = np.array2string(
            self.data,
            separator=", ",
            prefix=self._prefix_(0),
        )

        idxs = np.array2string(
            np.vstack(unravel_sparse_index(self.idxs, self.shape)),
            separator=", ",
            prefix=self._prefix_(1),
        )

        return self._repr_(
            data, idxs, self.shape, self.fill_value, self.dtype,
            ignore=('copy', 'sum_duplicates')
        )

    def __getitem__(self, idx):
        # TODO: slicing & fancy indexing & masks
        # TODO: negative indices
        try:
            ravel = ravel_sparse_multi_index(idx, self.shape)
        except ValueError:
            raise IndexError('...')
            # IndexError: index 9 is out of bounds for axis 0 with size 4
            # IndexError: too many indices for array
        loc = self.idxs.searchsorted(ravel)
        if self.idxs[loc] == ravel:
            return self.data[loc] # FIXME: returns an array; not a scalar...
        else:
            return self.fill_value

    def __contains__(self, item):
        return item in self.data

    def __eq__(self, other):
            ...

    def __array__(self):
        arr = np.full(self.shape, fill_value=self.fill_value, dtype=self.dtype)
        arr.ravel()[self._idxs] = self.data
        return arr

    def __array_ufunc__(self, ufunc, method, *inputs, **kwargs):
        if method == '__call__':
            pass
        return NotImplemented

    def astype(self, dtype, casting='unsafe', copy=True):
        data = self._data.astype(dtype, casting=casting, copy=copy)
        unraveled = np.vstack(unravel_sparse_index(self.idxs, self.shape))
        if copy:
            return type(self)(data, unraveled, self.shape, self.fill_value, copy=copy)
        else:
            self._data = data
            return self

    def reshape(self, *shape, copy=True):
        shape = super().reshape(*shape)
        if not shape.size:
            # TODO: copy=copy
            return self

        unraveled = np.unravel_index(self._idxs, shape)

        return type(self)(
            self.data, unraveled, shape, self.fill_value, self.dtype, copy=copy
        )

    def sort_index(self):
        """Sort indices and and their respective data values.
        This is an in-place operation.
        """
        sorter = self.idxs.argsort()
        self._idxs = self.idxs[sorter]
        self._data = self.data[sorter]

    def sum_duplicates(self):
        """Eliminate duplicate entries by adding them together.
        This is an in-place operation.
        """
        if not self.size:
            return

        self._idxs, inverse = np.unique(self.idxs, return_inverse=True)
        data = np.zeros_like(self.idxs, dtype=self.dtype)
        np.add.at(data, inverse, self.data)
        self._data = data

    def transpose(self, *axes):
        ...

    #TODO: override from ABC
    @property
    def nbytes(self):
        return self.data.nbytes + self.idxs.nbytes

    #TODO: SparseArray.dtype.setter since self.idxs used

max_stars_repo_name: MattEding/Array-Containers

max_issues_repo_name: MattEding/Array-Containers

max_forks_repo_name: MattEding/Array-Containers


text:
import tensorflow as tf
import numpy as np
import math
import sys
import os

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(BASE_DIR)
sys.path.append(os.path.join(BASE_DIR, '../utils'))
import tf_util
from transform_nets import feature_transform_net

def placeholder_inputs(batch_size, n_points, gmm):

    #Placeholders for the data
    n_gaussians = gmm.means_.shape[0]
    D = gmm.means_.shape[1]

    labels_pl = tf.placeholder(tf.int32, shape=(batch_size))

    w_pl = tf.placeholder(tf.float32, shape=(n_gaussians))
    mu_pl = tf.placeholder(tf.float32, shape=(n_gaussians, D))
    sigma_pl = tf.placeholder(tf.float32, shape=(n_gaussians, D)) # diagonal
    points_pl = tf.placeholder(tf.float32, shape=(batch_size, n_points, D))

    return points_pl, labels_pl, w_pl, mu_pl, sigma_pl

def get_model(points, w, mu, sigma, is_training, bn_decay=None, weigth_decay=0.005, add_noise=False, num_classes=40):
    """ Classification PointNet, input is BxNx3, output Bx40 """
    batch_size = points.get_shape()[0].value
    n_points = points.get_shape()[1].value
    n_gaussians = w.shape[0].value
    res = int(np.round(np.power(n_gaussians,1.0/3.0)))

    fv = tf_util.get_fv_minmax(points, w, mu, sigma, flatten=False)

    if add_noise:
        noise = tf.cond(is_training,
                        lambda: tf.random_normal(shape=tf.shape(fv), mean=0.0, stddev=0.01, dtype=tf.float32),
                        lambda:  tf.zeros(shape=tf.shape(fv)))
        #noise = tf.random_normal(shape=tf.shape(fv), mean=0.0, stddev=0.01, dtype=tf.float32)
        fv = fv + noise

    grid_fisher = tf.reshape(fv, [batch_size, -1, res, res, res])
    grid_fisher = tf.transpose(grid_fisher, [0, 2, 3, 4, 1])

    #3D Voxenet with pfv
    layer = 1
    net = tf_util.conv3d(grid_fisher, 32, [5, 5, 5], scope='conv'+str(layer),
                         stride=[2, 2, 2], padding='SAME', bn=True,
                         bn_decay=bn_decay, is_training=is_training)
    layer = layer + 1
    net = tf_util.conv3d(net, 32, [3, 3, 3], scope='conv'+str(layer),
                         stride=[1, 1, 1], padding='SAME', bn=True,
                         bn_decay=bn_decay, is_training=is_training)
    layer = layer + 1
    net = tf_util.max_pool3d(net, [2, 2, 2], scope='maxpool'+str(layer), stride=[2, 2, 2], padding='SAME')

    net = tf.reshape(net,[batch_size, -1])

    #Classifier
    net = tf_util.fully_connected(net, 128, bn=True, is_training=is_training,
                                  scope='fc1', bn_decay=bn_decay, weigth_decay=weigth_decay)
    net = tf_util.dropout(net, keep_prob=0.7, is_training=is_training,
                          scope='dp1')
    net = tf_util.fully_connected(net, num_classes, activation_fn=None, scope='fc4', is_training=is_training, weigth_decay=weigth_decay)

    return net, fv

def inception_module(input, n_filters=64, kernel_sizes=[3,5], is_training=None, bn_decay=None, scope='inception'):
    one_by_one =  tf_util.conv3d(input, n_filters, [1,1,1], scope= scope + '_conv1',
           stride=[1, 1, 1], padding='SAME', bn=True,
           bn_decay=bn_decay, is_training=is_training)
    three_by_three = tf_util.conv3d(one_by_one, int(n_filters/2), [kernel_sizes[0], kernel_sizes[0], kernel_sizes[0]], scope= scope + '_conv2',
           stride=[1, 1, 1], padding='SAME', bn=True,
           bn_decay=bn_decay, is_training=is_training)
    five_by_five = tf_util.conv3d(one_by_one, int(n_filters/2), [kernel_sizes[1], kernel_sizes[1], kernel_sizes[1]], scope=scope + '_conv3',
                          stride=[1, 1, 1], padding='SAME', bn=True,
                          bn_decay=bn_decay, is_training=is_training)
    average_pooling = tf_util.avg_pool3d(input, [kernel_sizes[0], kernel_sizes[0], kernel_sizes[0]], scope=scope+'_avg_pool', stride=[1, 1, 1], padding='SAME')
    average_pooling = tf_util.conv3d(average_pooling, n_filters, [1,1,1], scope= scope + '_conv4',
           stride=[1, 1, 1], padding='SAME', bn=True,
           bn_decay=bn_decay, is_training=is_training)

    output = tf.concat([ one_by_one, three_by_three, five_by_five, average_pooling], axis=4)
    #output = output + tf.tile(input) ??? #resnet
    return output

def get_loss(pred, label):
    """ pred: B*NUM_CLASSES,
        label: B, """

    weight_decay_losses = tf.get_collection('losses')
    weight_decay_loss = tf.reduce_sum(weight_decay_losses)

    loss = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=pred, labels=label)

    # loss = tf.losses.softmax_cross_entropy(logits=pred , onehot_labels=tf.one_hot(tf.to_int32(label),depth=40), label_smoothing=0.128)
    classify_loss = tf.reduce_mean(loss)
    tf.summary.scalar('classify loss', classify_loss)
    tf.summary.scalar('weight_decay_loss', weight_decay_loss)

    return classify_loss + weight_decay_loss

if __name__ == '__main__':
    with tf.Graph().as_default():
        inputs = tf.zeros((32, 1024, 3))
        outputs = get_model(inputs, tf.constant(True))
        print (outputs)


text:
import os
import numpy as np
import torch
from torch import sin, cos, atan2, acos
from math import pi
from pathlib import Path
from functools import wraps

from se3_transformer_pytorch.utils import exists, default, cast_torch_tensor, to_order
from se3_transformer_pytorch.spherical_harmonics import get_spherical_harmonics, clear_spherical_harmonics_cache

DATA_PATH = path = Path(os.path.dirname(__file__)) / 'data'

try:
    path = DATA_PATH / 'J_dense.pt'
    Jd = torch.load(str(path))
except:
    path = DATA_PATH / 'J_dense.npy'
    Jd_np = np.load(str(path), allow_pickle = True)
    Jd = list(map(torch.from_numpy, Jd_np))

def wigner_d_matrix(degree, alpha, beta, gamma, dtype = None, device = None):
    """Create wigner D matrices for batch of ZYZ Euler anglers for degree l."""
    J = Jd[degree].type(dtype).to(device)
    order = to_order(degree)
    x_a = z_rot_mat(alpha, degree)
    x_b = z_rot_mat(beta, degree)
    x_c = z_rot_mat(gamma, degree)
    res = x_a @ J @ x_b @ J @ x_c
    return res.view(order, order)

def z_rot_mat(angle, l):
    device, dtype = angle.device, angle.dtype
    order = to_order(l)
    m = angle.new_zeros((order, order))
    inds = torch.arange(0, order, 1, dtype=torch.long, device=device)
    reversed_inds = torch.arange(2 * l, -1, -1, dtype=torch.long, device=device)
    frequencies = torch.arange(l, -l - 1, -1, dtype=dtype, device=device)[None]

    m[inds, reversed_inds] = sin(frequencies * angle[None])
    m[inds, inds] = cos(frequencies * angle[None])
    return m

def irr_repr(order, alpha, beta, gamma, dtype = None):
    """
    irreducible representation of SO3
    - compatible with compose and spherical_harmonics
    """
    cast_ = cast_torch_tensor(lambda t: t)
    dtype = default(dtype, torch.get_default_dtype())
    alpha, beta, gamma = map(cast_, (alpha, beta, gamma))
    return wigner_d_matrix(order, alpha, beta, gamma, dtype = dtype)

@cast_torch_tensor
def rot_z(gamma):
    '''
    Rotation around Z axis
    '''
    return torch.tensor([
        [cos(gamma), -sin(gamma), 0],
        [sin(gamma), cos(gamma), 0],
        [0, 0, 1]
    ], dtype=gamma.dtype)

@cast_torch_tensor
def rot_y(beta):
    '''
    Rotation around Y axis
    '''
    return torch.tensor([
        [cos(beta), 0, sin(beta)],
        [0, 1, 0],
        [-sin(beta), 0, cos(beta)]
    ], dtype=beta.dtype)

@cast_torch_tensor
def x_to_alpha_beta(x):
    '''
    Convert point (x, y, z) on the sphere into (alpha, beta)
    '''
    x = x / torch.norm(x)
    beta = acos(x[2])
    alpha = atan2(x[1], x[0])
    return (alpha, beta)

def rot(alpha, beta, gamma):
    '''
    ZYZ Euler angles rotation
    '''
    return rot_z(alpha) @ rot_y(beta) @ rot_z(gamma)

def compose(a1, b1, c1, a2, b2, c2):
    """
    (a, b, c) = (a1, b1, c1) composed with (a2, b2, c2)
    """
    comp = rot(a1, b1, c1) @ rot(a2, b2, c2)
    xyz = comp @ torch.tensor([0, 0, 1.])
    a, b = x_to_alpha_beta(xyz)
    rotz = rot(0, -b, -a) @ comp
    c = atan2(rotz[1, 0], rotz[0, 0])
    return a, b, c

def spherical_harmonics(order, alpha, beta, dtype = None):
    return get_spherical_harmonics(order, theta = (pi - beta), phi = alpha)


text:
import numpy as np
from utils import average_list_of_lists

def generate_random_lists(length, range_floats, range_len):
    li = []
    for i in range(length):
        list_length = np.random.randint(range_len[0], high = range_len[1])
        li += [list(np.random.uniform(range_floats[0], high = range_floats[1], size = list_length))]
    return li

li = generate_random_lists(5, (1, 100), (10, 50))

for l in li:
    print(len(l))

li.sort(key = len)
li += [list(np.random.uniform(1, high = 100, size = len(li[-1])))]
print(li)

li_avg = average_list_of_lists(li)

print(len(li_avg))
print(li_avg)


text:
# -*- coding: utf-8 -*-
"""
Created on Thu Apr 28 13:00:04 2022

@author: rimmler
"""

from lmfit import Parameters, minimize
import modules.functions.stfmrAnglePlotFitFuncKarimeddiny as fk
from units import deg2rad
import numpy as np

default_torques = ['tau_xDL', 'tau_xFL', 'tau_yDL', 'tau_yFL', 'tau_zDL', 'tau_zFL']

def check_comps(comps):
    if comps not in ['x', 'y', 'z', 'xy', 'xz', 'yz', 'xyz']:
        raise ValueError

def V_Karimeddiny_Hall_fitting(comps, phi_data_deg, Vset_mode,
                               Vsxx_data, Vaxx_data, Vsxy_data, Vaxy_data,
                               phi_fit_deg, cps,
                               assume_arts=True, do_lmfit_report=False, do_check_fit=False):

    def Vsxx_dataset(params, phi):
        return fk.Vsxx(phi, params['Irf'], params['Ramr'], params['Tart'], params['alpha'],
                     params['w1'], params['w2'], params['wp'], params['W (m)'],
                     params['tau_xDL'], params['tau_xFL'], params['tau_yDL'],
                     params['tau_yFL'], params['tau_zDL'], params['tau_zFL'])

    def Vaxx_dataset(params, phi):
        return fk.Vaxx(phi, params['Irf'], params['Ramr'], params['alpha'],
                     params['w1'], params['w2'], params['wp'],
                     params['tau_xDL'], params['tau_xFL'], params['tau_yDL'],
                     params['tau_yFL'], params['tau_zDL'], params['tau_zFL'])

    def Vsxy_dataset(params, phi):
        return fk.Vsxy(phi, params['Irf'], params['Rphe'], params['Rahe'], params['Tart'], params['alpha'],
                     params['w1'], params['w2'], params['wp'], params['L (m)'],
                     params['tau_xDL'], params['tau_xFL'], params['tau_yDL'],
                     params['tau_yFL'], params['tau_zDL'], params['tau_zFL'])

    def Vaxy_dataset(params, phi):
        return fk.Vaxy(phi, params['Irf'], params['Rphe'], params['Rahe'], params['alpha'],
                     params['w1'], params['w2'], params['wp'],
                     params['tau_xDL'], params['tau_xFL'], params['tau_yDL'],
                     params['tau_yFL'], params['tau_zDL'], params['tau_zFL'])

    def objective(params, phi, Vset_mode, Vsxx_data, Vaxx_data, Vsxy_data, Vaxy_data):
        '''Calculate total residual for fits.'''
        V1 = Vsxx_data
        V2 = Vaxx_data
        if Vset_mode == 0:
            V3 = Vsxy_data
        elif Vset_mode == 1:
            V3 = Vaxy_data

        data = np.array([V1, V2, V3])
        resid = 0.*data[:]

        # make residual per data set
        resid[0, :] = data[0, :] - Vsxx_dataset(params, phi)
        resid[1, :] = data[1, :] - Vaxx_dataset(params, phi)
        if Vset_mode == 0:
            resid[2, :] = data[2, :] - Vsxy_dataset(params, phi)
        elif Vset_mode == 1:
            resid[2, :] = data[2, :] - Vaxy_dataset(params, phi)

        # now flatten this to a 1D array, as minimize() needs
        return resid.flatten()

    # Get fixed parameters from cps
    alpha = cps['alpha']
    H0 = cps['H0_SI (A/m)']
    B0 = fk.B0(H0)
    Meff = cps['Meff_SI (A/m)']
    gamma = fk.gamma(cps['g_e'])
    w1 = fk.w1(gamma, B0)
    w2 = fk.w2(gamma, B0, Meff)
    wp = fk.wp(w1, w2)
    Irf = cps['Irf (mA)']*1e-3 # to A
    Ramr = cps['Ramr (Ohm)']
    Rphe = cps['Rphe (Ohm)']
    Rahe = cps['Rahe (Ohm)']
    W = cps['W (m)']
    L = cps['L (m)']

    # Convert angle to rad
    check_comps(comps)
    phi_data = deg2rad(phi_data_deg)
    phi_plt = deg2rad(phi_fit_deg)

    # Fixed parameters
    fit_params = Parameters()

    fit_params.add('Irf', value=Irf)
    fit_params.add('Ramr', value=Ramr)
    fit_params.add('Rphe', value=Rphe)
    fit_params.add('Rahe', value=Rahe)
    fit_params.add('W', value=W)
    fit_params.add('L', value=L)

    fit_params.add('alpha', value=alpha)
    fit_params.add('w1', value=w1)
    fit_params.add('w2', value=w2)
    fit_params.add('wp', value=wp)

    for param in ['Irf', 'Ramr', 'Rphe', 'Rahe', 'W', 'L', 'alpha', 'w1', 'w2', 'wp']:
        fit_params[param].vary=False

    # Varying parameters (fit parameters)

    for comp in ['x', 'y', 'z']:
        if comp in comps:
            # Case: If the component is in the fitting components (e.g. x is in xyz),
            # set its initial value to 1 and let it vary.
            fit_params.add(f'tau_{comp}DL', value=1)
            fit_params.add(f'tau_{comp}FL', value=1)
        else:
            # Else, fix its value at 0
            fit_params.add(f'tau_{comp}DL', value=0)
            fit_params.add(f'tau_{comp}FL', value=0)
            fit_params[f'tau_{comp}DL'].vary=False
            fit_params[f'tau_{comp}FL'].vary=False

    ###
    # Iterative fitting to include artifacts
    if assume_arts is True:
        iterats = 2
    else:
        iterats = 1

    for i in range(iterats):
        if i == 0:
            # First iteration: No artifacts
            fit_params.add('Tart', value=0)
            fit_params['Tart'].vary=False

            out = minimize(objective, fit_params, args=(phi_data, Vsxx_data, Vaxx_data, Vsxy_data, Vaxy_data))
            # if do_lmfit_report is True:
            #     report_fit(out.params)

        elif i == 1:
            # Second iteration: if artifacts assumed: start from output params
            # of previous fit and let Tart vary
            fit_params['Tart'].value = -1
            fit_params['Tart'].min = -1e3
            fit_params['Tart'].max = 0
            fit_params['Tart'].vary = True

            for comp in ['x', 'y', 'z']:
                if comp in comps:
                    # Case: If the component is in the fitting components (e.g. x is in xyz),
                    # set its initial value to 1 and let it vary.
                    fit_params[f'tau_{comp}DL'] = out.params[f'tau_{comp}DL']
                    fit_params[f'tau_{comp}FL'] = out.params[f'tau_{comp}FL']

            out = minimize(objective, fit_params, args=(phi_data, Vset_mode, Vsxx_data, Vaxx_data, Vsxy_data, Vaxy_data))

    # _________________________________________________________________________
    # Run the global fit and show the fitting result

    # _________________________________________________________________________
    # Plot the data sets and fits
    Vsxx_fit = Vsxx_dataset(out.params, phi_data)
    Vaxx_fit = Vaxx_dataset(out.params, phi_data)
    Vsxx_plt = Vsxx_dataset(out.params, phi_plt)
    Vaxx_plt = Vaxx_dataset(out.params, phi_plt)

    Vsxy_fit = Vsxy_dataset(out.params, phi_data)
    Vaxy_fit = Vaxy_dataset(out.params, phi_data)
    Vsxy_plt = Vsxy_dataset(out.params, phi_plt)
    Vaxy_plt = Vaxy_dataset(out.params, phi_plt)

    # if do_check_fit is True:
    #     plt.figure()
    #     plt.plot(phi_data, Vs_data, 'o', phi_data, Vs_fit, '-')
    #     plt.plot(phi_data, Va_data, 'o', phi_data, Va_fit, '-')

    # Get parameters in dict
    params_dict = {}
    for key in out.params.keys():
        params_dict[key] = out.params[key].value

    return out.params, params_dict, Vsxx_fit, Vsxx_plt, Vaxx_fit, Vaxx_plt, Vsxy_fit, Vsxy_plt, Vaxy_fit, Vaxy_plt

def get_norm_torques_karimed(params, norm_to):
    ''' Get normalized torques from fitting parameter output '''
    if norm_to.split('_')[0] != 'tau':
        norm_to = 'tau_'+norm_to

    torques = {}
    torques_norm = {}
    for key in default_torques:
        tau = params[key].value
        tau_norm = tau / params[norm_to].value
        torques[key] = tau
        torques_norm[f'{key}_norm'] = tau_norm

    return torques, torques_norm

def calc_Ru(yobs, ycalc):
    return (np.sum(np.abs(yobs)**2) - np.sum(np.abs(ycalc)**2)) / np.sum(np.abs(yobs)**2)

max_stars_repo_name: BRimmler/STFMR-Analysis-code

max_issues_repo_name: BRimmler/STFMR-Analysis-code

max_forks_repo_name: BRimmler/STFMR-Analysis-code


text:
import matplotlib.pyplot as plt
import numpy
import pandas as pd
import control.matlab as cnt
import cp
import scipy.optimize as optim

dfdata = pd.read_csv('T1.out', sep='\t', header=None)
datadata = dfdata.values

plt.plot(datadata[:,0], 1000*datadata[:,6])
plt.title("QAero")
plt.show()

Jr = 115926
inert = datadata[:,7]*0.01745329*Jr
plt.plot(datadata[:,0], 1000*datadata[:,6]+inert)
plt.title("QAero")
plt.show()

plt.plot(datadata[:,0], 1000*datadata[:,8])
plt.title("POWER")
plt.show()

reason: import numpy,import scipy


text:
#################################################################################
# The Institute for the Design of Advanced Energy Systems Integrated Platform
# Framework (IDAES IP) was produced under the DOE Institute for the
# Design of Advanced Energy Systems (IDAES), and is copyright (c) 2018-2021
# by the software owners: The Regents of the University of California, through
# Lawrence Berkeley National Laboratory,  National Technology & Engineering
# Solutions of Sandia, LLC, Carnegie Mellon University, West Virginia University
# Research Corporation, et al.  All rights reserved.
#
# Please see the files COPYRIGHT.md and LICENSE.md for full copyright and
# license information.
#################################################################################
import pyomo.environ as pyo
from collections import deque
import pandas as pd
from idaes.apps.grid_integration import Tracker
from idaes.apps.grid_integration import Bidder
from idaes.apps.grid_integration import PlaceHolderForecaster

from pyomo.common.dependencies import attempt_import

Prescient, prescient_avail = attempt_import("prescient.simulator.Prescient")

class ThermalGenerator:

    """
    Simple thermal generator model (MIP). Equations models are from Gao, Knueven,
    Siirola, Miller, Dowling (2022). Multiscale Simulation of Integrated Energy
    System and Electricity Market Interactions. Applied Energy.
    """

    # Using 4 segments to be consistent with models in RTS-GMLC dataset
    segment_number = 4

    def __init__(self, rts_gmlc_dataframe, horizon=48, generator="102_STEAM_3"):

        """
        Initializes the class object by building the thermal generator model.

        Arguments:
            rts_gmlc_dataframe: the RTS-GMLC generator data in Pandas DataFrame
            horizon: the length of the planning horizon of the model.
            generator: a generator in RTS-GMLC

        Returns:
            None
        """

        self.generator = generator
        self.horizon = horizon
        self.model_data = self.assemble_model_data(
            generator_name=generator, gen_params=rts_gmlc_dataframe
        )
        self.result_list = []

    @staticmethod
    def assemble_model_data(generator_name, gen_params):

        """
        This function assembles the parameter data to build the thermal generator
        model, given a list of generator names and the RTS-GMLC data directory.

        Arguments:
            generator_names: a generator name in RTS-GMLC dataset.
            gen_params: the RTS-GMLC generator data in Pandas DataFrame

        Returns:
            model_data: a dictionary which has this structure
            {data type name: value}.
        """

        gen_params = gen_params.set_index("GEN UID", inplace=False)
        properties = [
            "PMin MW",
            "PMax MW",
            "Min Up Time Hr",
            "Min Down Time Hr",
            "Ramp Rate MW/Min",
            "Start Heat Warm MBTU",
            "Fuel Price $/MMBTU",
            "HR_avg_0",
            "HR_incr_1",
            "HR_incr_2",
            "HR_incr_3",
            "Output_pct_1",
            "Output_pct_2",
            "Output_pct_3",
        ]

        # to dict
        model_data = gen_params.loc[generator_name, properties].to_dict()

        model_data["RU"] = model_data["Ramp Rate MW/Min"] * 60
        model_data["RD"] = model_data["RU"]
        model_data["SU"] = min(model_data["PMin MW"], model_data["RU"])
        model_data["SD"] = min(model_data["PMin MW"], model_data["RD"])
        model_data["SU Cost"] = (
            model_data["Start Heat Warm MBTU"] * model_data["Fuel Price $/MMBTU"]
        )

        model_data["Min Load Cost"] = (
            model_data["HR_avg_0"]
            / 1000
            * model_data["Fuel Price $/MMBTU"]
            * model_data["PMin MW"]
        )

        model_data["Power Segments"] = {}
        model_data["Marginal Costs"] = {}

        model_data["Original Marginal Cost Curve"] = {}
        model_data["Original Marginal Cost Curve"][model_data["PMin MW"]] = (
            model_data["Min Load Cost"] / model_data["PMin MW"]
        )

        for l in range(1, ThermalGenerator.segment_number):
            model_data["Power Segments"][l] = (
                model_data["Output_pct_{}".format(l)] * model_data["PMax MW"]
            )
            model_data["Marginal Costs"][l] = (
                model_data["HR_incr_{}".format(l)]
                / 1000
                * model_data["Fuel Price $/MMBTU"]
            )
            model_data["Original Marginal Cost Curve"][
                model_data["Power Segments"][l]
            ] = model_data["Marginal Costs"][l]

        return model_data

    @staticmethod
    def _add_UT_DT_constraints(b):

        """
        This function adds the minimum up/down time constraints using eq. 4 - 5
        in "On mixed-integer programming formulations for the unit commitment
        problem". INFORMS Journal on Computing, 32(4), pp.857-876. Knueven, B.,
        Ostrowski, J. and Watson, J.P., 2020.

        Arguments:
            b: a pyomo block

        Returns:
            None
        """

        def pre_shut_down_trajectory_set_rule(b):
            return (t for t in range(-pyo.value(b.min_dw_time) + 1, 0))

        b.pre_shut_down_trajectory_set = pyo.Set(
            dimen=1, initialize=pre_shut_down_trajectory_set_rule, ordered=True
        )

        def pre_start_up_trajectory_set_rule(b):
            return (t for t in range(-pyo.value(b.min_up_time) + 1, 0))

        b.pre_start_up_trajectory_set = pyo.Set(
            dimen=1, initialize=pre_start_up_trajectory_set_rule, ordered=True
        )

        b.pre_shut_down_trajectory = pyo.Param(
            b.pre_shut_down_trajectory_set, initialize=0, mutable=True
        )
        b.pre_start_up_trajectory = pyo.Param(
            b.pre_start_up_trajectory_set, initialize=0, mutable=True
        )

        def min_down_time_rule(b, h):
            if h < pyo.value(b.min_dw_time):
                return (
                    sum(
                        b.pre_shut_down_trajectory[h0]
                        for h0 in range(h - pyo.value(b.min_dw_time) + 1, 0)
                    )
                    + sum(b.shut_dw[h0] for h0 in range(h + 1))
                    <= 1 - b.on_off[h]
                )
            else:
                return (
                    sum(
                        b.shut_dw[h0]
                        for h0 in range(h - pyo.value(b.min_dw_time) + 1, h + 1)
                    )
                    <= 1 - b.on_off[h]
                )

        b.min_down_time_con = pyo.Constraint(b.HOUR, rule=min_down_time_rule)

        def min_up_time_rule(b, h):
            if h < pyo.value(b.min_up_time):
                return (
                    sum(
                        b.pre_start_up_trajectory[h0]
                        for h0 in range(h - pyo.value(b.min_up_time) + 1, 0)
                    )
                    + sum(b.start_up[h0] for h0 in range(h + 1))
                    <= b.on_off[h]
                )
            else:
                return (
                    sum(
                        b.start_up[h0]
                        for h0 in range(h - pyo.value(b.min_up_time) + 1, h + 1)
                    )
                    <= b.on_off[h]
                )

        b.min_up_time_con = pyo.Constraint(b.HOUR, rule=min_up_time_rule)

        return

    def populate_model(self, b):

        """
        This function builds the model for a thermal generator.

        Arguments:
            b: a pyomo block

        Returns:
            b: the constructed block.
        """

        model_data = self.model_data

        ## define the sets
        b.HOUR = pyo.Set(initialize=list(range(self.horizon)))
        b.SEGMENTS = pyo.Set(initialize=list(range(1, self.segment_number)))

        ## define the parameters

        b.start_up_cost = pyo.Param(initialize=model_data["SU Cost"], mutable=False)

        # capacity of generators: upper bound (MW)
        b.Pmax = pyo.Param(initialize=model_data["PMax MW"], mutable=False)

        # minimum power of generators: lower bound (MW)
        b.Pmin = pyo.Param(initialize=model_data["PMin MW"], mutable=False)

        b.power_segment_bounds = pyo.Param(
            b.SEGMENTS, initialize=model_data["Power Segments"], mutable=False
        )

        # get the cost slopes
        b.F = pyo.Param(
            b.SEGMENTS, initialize=model_data["Marginal Costs"], mutable=False
        )

        b.min_load_cost = pyo.Param(
            initialize=model_data["Min Load Cost"], mutable=False
        )

        # Ramp up limits (MW/h)
        b.ramp_up = pyo.Param(initialize=model_data["RU"], mutable=False)

        # Ramp down limits (MW/h)
        b.ramp_dw = pyo.Param(initialize=model_data["RD"], mutable=False)

        # start up ramp limit
        b.ramp_start_up = pyo.Param(initialize=model_data["SU"], mutable=False)

        # shut down ramp limit
        b.ramp_shut_dw = pyo.Param(initialize=model_data["SD"], mutable=False)

        # minimum down time [hr]
        b.min_dw_time = pyo.Param(
            initialize=int(model_data["Min Down Time Hr"]), mutable=False
        )

        # minimum up time [hr]
        b.min_up_time = pyo.Param(
            initialize=int(model_data["Min Up Time Hr"]), mutable=False
        )

        # on/off status from previous day
        b.pre_on_off = pyo.Param(within=pyo.Binary, default=1, mutable=True)

        # define a function to initialize the previous power params
        def init_pre_pow_fun(b):
            return b.pre_on_off * b.Pmin

        b.pre_P_T = pyo.Param(initialize=init_pre_pow_fun, mutable=True)

        ## define the variables

        # generator power (MW)

        # power generated by thermal generator
        b.P_T = pyo.Var(
            b.HOUR, initialize=model_data["PMin MW"], within=pyo.NonNegativeReals
        )

        # binary variables indicating on/off
        b.on_off = pyo.Var(b.HOUR, initialize=True, within=pyo.Binary)

        # binary variables indicating  start_up
        b.start_up = pyo.Var(b.HOUR, initialize=False, within=pyo.Binary)

        # binary variables indicating shut down
        b.shut_dw = pyo.Var(b.HOUR, initialize=False, within=pyo.Binary)

        # power produced in each segment
        b.power_segment = pyo.Var(b.HOUR, b.SEGMENTS, within=pyo.NonNegativeReals)

        ## Constraints

        # bounds on gen_pow
        def lhs_bnd_gen_pow_fun(b, h):
            return b.on_off[h] * b.Pmin <= b.P_T[h]

        b.lhs_bnd_gen_pow = pyo.Constraint(b.HOUR, rule=lhs_bnd_gen_pow_fun)

        def rhs_bnd_gen_pow_fun(b, h):
            return b.P_T[h] <= b.on_off[h] * b.Pmax

        b.rhs_bnd_gen_pow = pyo.Constraint(b.HOUR, rule=rhs_bnd_gen_pow_fun)

        # linearized power
        def linear_power_fun(b, h):
            return (
                b.P_T[h]
                == sum(b.power_segment[h, l] for l in b.SEGMENTS) + b.Pmin * b.on_off[h]
            )

        b.linear_power = pyo.Constraint(b.HOUR, rule=linear_power_fun)

        # bounds on segment power
        def seg_pow_bnd_fun(b, h, l):
            if l == 1:
                return (
                    b.power_segment[h, l]
                    <= (b.power_segment_bounds[l] - b.Pmin) * b.on_off[h]
                )
            else:
                return (
                    b.power_segment[h, l]
                    <= (b.power_segment_bounds[l] - b.power_segment_bounds[l - 1])
                    * b.on_off[h]
                )

        b.seg_pow_bnd = pyo.Constraint(b.HOUR, b.SEGMENTS, rule=seg_pow_bnd_fun)

        # start up and shut down logic (Garver 1962)
        def start_up_shut_dw_fun(b, h):
            if h == 0:
                return b.start_up[h] - b.shut_dw[h] == b.on_off[h] - b.pre_on_off
            else:
                return b.start_up[h] - b.shut_dw[h] == b.on_off[h] - b.on_off[h - 1]

        b.start_up_shut_dw = pyo.Constraint(b.HOUR, rule=start_up_shut_dw_fun)

        # either start up or shut down
        def start_up_or_shut_dw_fun(b, h):
            return b.start_up[h] + b.shut_dw[h] <= 1

        b.start_up_or_shut_dw = pyo.Constraint(b.HOUR, rule=start_up_or_shut_dw_fun)

        # ramp up limits
        def ramp_up_fun(b, h):
            """
            h stand for hour
            """
            if h == 0:
                return (
                    b.P_T[h]
                    <= b.pre_P_T
                    + b.ramp_up * b.pre_on_off
                    + b.ramp_start_up * b.start_up[h]
                )
            else:
                return (
                    b.P_T[h]
                    <= b.P_T[h - 1]
                    + b.ramp_up * b.on_off[h - 1]
                    + b.ramp_start_up * b.start_up[h]
                )

        b.ramp_up_con = pyo.Constraint(b.HOUR, rule=ramp_up_fun)

        # ramp shut down limits
        def ramp_shut_dw_fun(b, h):
            """
            h stand for hour.
            """
            if h == 0:
                return b.pre_P_T <= b.Pmax * b.on_off[h] + b.ramp_shut_dw * b.shut_dw[h]
            else:
                return (
                    b.P_T[h - 1] <= b.Pmax * b.on_off[h] + b.ramp_shut_dw * b.shut_dw[h]
                )

        b.ramp_shut_dw_con = pyo.Constraint(b.HOUR, rule=ramp_shut_dw_fun)

        # ramp down limits
        def ramp_dw_fun(b, h):
            """
            h stand for hour.
            """
            if h == 0:
                return (
                    b.pre_P_T - b.P_T[h]
                    <= b.ramp_dw * b.on_off[h] + b.ramp_shut_dw * b.shut_dw[h]
                )
            else:
                return (
                    b.P_T[h - 1] - b.P_T[h]
                    <= b.ramp_dw * b.on_off[h] + b.ramp_shut_dw * b.shut_dw[h]
                )

        b.ramp_dw_con = pyo.Constraint(b.HOUR, rule=ramp_dw_fun)

        ## add min up and down time constraints
        self._add_UT_DT_constraints(b)

        ## Expression
        def prod_cost_fun(b, h):
            return b.min_load_cost * b.on_off[h] + sum(
                b.F[l] * b.power_segment[h, l] for l in b.SEGMENTS
            )

        b.prod_cost_approx = pyo.Expression(b.HOUR, rule=prod_cost_fun)

        # start up costs
        def start_cost_fun(b, h):
            return b.start_up_cost * b.start_up[h]

        b.start_up_cost_expr = pyo.Expression(b.HOUR, rule=start_cost_fun)

        # total cost
        def tot_cost_fun(b, h):
            return b.prod_cost_approx[h] + b.start_up_cost_expr[h]

        b.tot_cost = pyo.Expression(b.HOUR, rule=tot_cost_fun)

        return

    @staticmethod
    def _update_UT_DT(b, implemented_shut_down, implemented_start_up):
        """
        This method updates the parameters in the minimum up/down time
        constraints based on the implemented shut down and start up events.

        Arguments:
            b: the block that needs to be updated
            implemented_shut_down: realized shut down events: [].
            implemented_start_up: realized start up events: []

        Returns:
            None
        """

        pre_shut_down_trajectory_copy = deque([])
        pre_start_up_trajectory_copy = deque([])

        # copy old trajectory
        for t in b.pre_shut_down_trajectory_set:
            pre_shut_down_trajectory_copy.append(
                round(pyo.value(b.pre_shut_down_trajectory[t]))
            )
        for t in b.pre_start_up_trajectory_set:
            pre_start_up_trajectory_copy.append(
                round(pyo.value(b.pre_start_up_trajectory[t]))
            )

        # add implemented trajectory to the queue
        pre_shut_down_trajectory_copy += deque(implemented_shut_down)
        pre_start_up_trajectory_copy += deque(implemented_start_up)

        # pop out outdated trajectory
        while len(pre_shut_down_trajectory_copy) > pyo.value(b.min_dw_time) - 1:
            pre_shut_down_trajectory_copy.popleft()
        while len(pre_start_up_trajectory_copy) > pyo.value(b.min_up_time) - 1:
            pre_start_up_trajectory_copy.popleft()

        # actual update
        for t in b.pre_shut_down_trajectory_set:
            b.pre_shut_down_trajectory[t] = pre_shut_down_trajectory_copy.popleft()

        for t in b.pre_start_up_trajectory_set:
            b.pre_start_up_trajectory[t] = pre_start_up_trajectory_copy.popleft()

        return

    @staticmethod
    def _update_power(b, implemented_power_output):
        """
        This method updates the parameters in the ramping constraints based on
        the implemented power outputs.

        Arguments:
            b: the block that needs to be updated
            implemented_power_output: realized power outputs: []

         Returns:
             None
        """

        b.pre_P_T = round(implemented_power_output[-1], 2)
        b.pre_on_off = round(int(implemented_power_output[-1] > 1e-3))

        return

    def update_model(
        self, b, implemented_shut_down, implemented_start_up, implemented_power_output
    ):

        """
        This method updates the parameters in the model based on
        the implemented power outputs, shut down and start up events.

        Arguments:
            b: the block that needs to be updated
            implemented_shut_down: realized shut down events: [].
            implemented_start_up: realized start up events: []
            implemented_power_output: realized power outputs: []

         Returns:
             None
        """

        self._update_UT_DT(b, implemented_shut_down, implemented_start_up)
        self._update_power(b, implemented_power_output)

        return

    @staticmethod
    def get_implemented_profile(b, last_implemented_time_step):

        """
        This method gets the implemented variable profiles in the last optimization
        solve.

        Arguments:
            b: the block
            model_var: intended variable name in str
            last_implemented_time_step: time index for the last implemented time
                                        step

         Returns:
             profile: the intended profile, {unit: [...]}
        """

        implemented_shut_down = deque(
            [pyo.value(b.shut_dw[t]) for t in range(last_implemented_time_step + 1)]
        )
        implemented_start_up = deque(
            [pyo.value(b.start_up[t]) for t in range(last_implemented_time_step + 1)]
        )
        implemented_power_output = deque(
            [pyo.value(b.P_T[t]) for t in range(last_implemented_time_step + 1)]
        )

        return {
            "implemented_shut_down": implemented_shut_down,
            "implemented_start_up": implemented_start_up,
            "implemented_power_output": implemented_power_output,
        }

    @staticmethod
    def get_last_delivered_power(b, last_implemented_time_step):

        """
        Returns the last delivered power output.

        Arguments:
            None

        Returns:
            None
        """

        return pyo.value(b.P_T[last_implemented_time_step])

    def record_results(self, b, date=None, hour=None, **kwargs):

        """
        Record the operations stats for the model.

        Arguments:

            date: current simulation date
            hour: current simulation hour

        Returns:
            None

        """

        df_list = []

        for t in b.HOUR:

            result_dict = {}
            result_dict["Generator"] = self.generator
            result_dict["Date"] = date
            result_dict["Hour"] = hour

            # simulation inputs
            result_dict["Horizon [hr]"] = int(t)

            # model vars
            result_dict["Thermal Power Generated [MW]"] = float(
                round(pyo.value(b.P_T[t]), 2)
            )

            result_dict["On/off [bin]"] = int(round(pyo.value(b.on_off[t])))
            result_dict["Start Up [bin]"] = int(round(pyo.value(b.start_up[t])))
            result_dict["Shut Down [bin]"] = int(round(pyo.value(b.shut_dw[t])))

            result_dict["Production Cost [$]"] = float(
                round(pyo.value(b.prod_cost_approx[t]), 2)
            )
            result_dict["Start-up Cost [$]"] = float(
                round(pyo.value(b.start_up_cost_expr[t]), 2)
            )
            result_dict["Total Cost [$]"] = float(round(pyo.value(b.tot_cost[t]), 2))

            # calculate mileage
            if t == 0:
                result_dict["Mileage [MW]"] = float(
                    round(abs(pyo.value(b.P_T[t] - b.pre_P_T)), 2)
                )
            else:
                result_dict["Mileage [MW]"] = float(
                    round(abs(pyo.value(b.P_T[t] - b.P_T[t - 1])), 2)
                )

            for key in kwargs:
                result_dict[key] = kwargs[key]

            result_df = pd.DataFrame.from_dict(result_dict, orient="index")
            df_list.append(result_df.T)

        # save the result to object property
        # wait to be written when simulation ends
        self.result_list.append(pd.concat(df_list))

        return

    def write_results(self, path):

        """
        This methods writes the saved operation stats into an csv file.

        Arguments:
            path: the path to write the results.

        Return:
            None
        """

        pd.concat(self.result_list).to_csv(path, index=False)

    @property
    def power_output(self):
        return "P_T"

    @property
    def total_cost(self):
        return ("tot_cost", 1)

    @property
    def default_bids(self):
        return self.model_data["Original Marginal Cost Curve"]

    @property
    def pmin(self):
        return self.model_data["PMin MW"]

if __name__ == "__main__":

    generator = "102_STEAM_3"
    horizon = 4

    rts_gmlc_dataframe = pd.read_csv("gen.csv")
    solver = pyo.SolverFactory("cbc")

    run_tracker = True
    run_bidder = True
    run_prescient = True

    if run_tracker:

        # create a tracker model
        tracking_model_object = ThermalGenerator(
            rts_gmlc_dataframe=rts_gmlc_dataframe,
            horizon=horizon,
            generator="102_STEAM_3",
        )
        # make a tracker
        thermal_tracker = Tracker(
            tracking_model_object=tracking_model_object,
            n_tracking_hour=1,
            solver=solver,
        )

        market_dispatch = [30, 40, 50, 70]

        thermal_tracker.track_market_dispatch(
            market_dispatch=market_dispatch, date="2021-07-26", hour="17:00"
        )
        thermal_tracker.write_results(path="./")

    if run_bidder:

        # create a tracker model
        bidding_model_object = ThermalGenerator(
            rts_gmlc_dataframe=rts_gmlc_dataframe, horizon=48, generator="102_STEAM_3"
        )

        # create forecaster
        price_forecasts_df = pd.read_csv("lmp_forecasts_concat.csv")
        forecaster = PlaceHolderForecaster(price_forecasts_df=price_forecasts_df)

        thermal_bidder = Bidder(
            bidding_model_object=bidding_model_object,
            n_scenario=10,
            solver=solver,
            forecaster=forecaster,
        )

        date = "2020-07-10"
        hour = "13:00"
        bids = thermal_bidder.compute_bids(date, hour)
        thermal_bidder.write_results(path="./")

    if run_prescient and prescient_avail:

        options = {
            "data_path": "/home/xgao1/DowlingLab/RTS-GMLC/RTS_Data/SourceData",
            "input_format": "rts-gmlc",
            "simulate_out_of_sample": True,
            "run_sced_with_persistent_forecast_errors": True,
            "output_directory": "bidding_plugin_test_thermal_generator",
            "start_date": "07-10-2020",
            "num_days": 1,
            "sced_horizon": 4,
            "compute_market_settlements": True,
            "day_ahead_pricing": "LMP",
            "ruc_mipgap": 0.01,
            "symbolic_solver_labels": True,
            "reserve_factor": 0.0,
            "deterministic_ruc_solver": "cbc",
            "sced_solver": "cbc",
            "plugin": {
                "doubleloop": {
                    "module": "thermal_generator_prescient_plugin.py",
                    "bidding_generator": "102_STEAM_3",
                }
            },
        }

        Prescient().simulate(**options)

max_stars_repo_name: OOAmusat/idaes-pse

max_issues_repo_name: OOAmusat/idaes-pse

max_forks_repo_name: OOAmusat/idaes-pse

reason: import pyomo,from pyomo