text:
<?xml version="1.0" encoding="UTF-8"?>

<!--
   The bootstrap of the server
-->
<deployment xmlns="urn:jboss:bean-deployer:2.0">

   <bean name="SystemProperties" class="org.jboss.embedded.util.SystemProperties">
      <property name="properties">
         <map keyClass="java.lang.String" valueClass="java.lang.String">
            <entry>
               <key>jboss.bind.address</key>
               <value>0.0.0.0</value>
            </entry>
            <entry>
               <key>com.arjuna.ats.arjuna.common.propertiesFile</key>
               <value>${jboss.home.dir}/conf/jbossjta-properties.xml</value>
            </entry>
            <entry>
               <key>com.arjuna.ats.arjuna.objectstore.objectStoreDir</key>
               <value>${jboss.server.data.dir}/ObjectStore</value>
            </entry>
         </map>
      </property>
   </bean>

   <!-- The legacy JMX kernel -->
   <bean name="JMXKernel" class="org.jboss.embedded.adapters.JMXKernel">
      <property name="kernel">
         <inject bean="jboss.kernel:service=Kernel"/>
      </property>
      <property name="serverConfig">
         <inject bean="ServerConfig"/>
      </property>
   </bean>

   <!-- The MainDeployer -->
   <bean name="MainDeployer" class="org.jboss.deployers.plugins.main.MainDeployerImpl">
      <property name="structuralDeployers"><inject bean="StructuralDeployers"/></property>
      <property name="deployers"><inject bean="Deployers"/></property>
   </bean>

   <!-- The holder for deployers that determine structure -->
   <bean name="StructuralDeployers" class="org.jboss.deployers.vfs.plugins.structure.VFSStructuralDeployersImpl">
      <property name="structureBuilder">
         <!-- The consolidator of the structure information -->
         <bean name="StructureBuilder" class="org.jboss.deployers.vfs.plugins.structure.VFSStructureBuilder"/>
      </property>
      <!-- Accept any implementor of structure deployer -->
      <incallback method="addDeployer"/>
      <uncallback method="removeDeployer"/>
   </bean>

   <!-- The holder for deployers that do real deployment -->
   <bean name="Deployers" class="org.jboss.deployers.plugins.deployers.DeployersImpl">
      <constructor><parameter><inject bean="jboss.kernel:service=KernelController"/></parameter></constructor>
      <!-- Accept any implementor of deployer -->
      <incallback method="addDeployer"/>
      <uncallback method="removeDeployer"/>
   </bean>

   <!-- A declared structure descriptor deployer -->
   <bean name="DeclaredStructure" class="org.jboss.deployers.vfs.plugins.structure.explicit.DeclaredStructure"></bean>

    <!-- JAR Structure -->
    <bean name="JARStructure" class="org.jboss.deployers.vfs.plugins.structure.jar.JARStructure">
       <!-- Unless specified the default list of suffixes is .zip, .ear, .jar, ,.rar, .war, .sar, .har, .aop -->
       <constructor>
          <parameter>
             <set elementClass="java.lang.String">
                <value>.zip</value>
                <value>.ear</value>
                <value>.jar</value>
                <value>.rar</value>
                <value>.war</value>
                <value>.sar</value>
                <value>.har</value>
                <value>.aop</value>
                <value>.deployer</value>
                <value>.beans</value>
             </set>
          </parameter>
       </constructor>
       <property name="candidateStructureVisitorFactory">
          <!-- Any file that is not an ordinary directory is a candidate -->
          <bean name="JARStructureCandidates" class="org.jboss.deployers.vfs.spi.structure.helpers.DefaultCandidateStructureVisitorFactory">
             <!-- A filter to exclude some obvious non-subdeployments -->
             <property name="filter">
                <bean name="JARFilter" class="org.jboss.virtual.plugins.vfs.helpers.SuffixesExcludeFilter">
                   <constructor><parameter>
                      <list elementClass="java.lang.String">
                         <!-- Exclude class files as subdeployments -->
                         <value>.class</value>
                      </list>
                   </parameter></constructor>
                </bean>
             </property>
          </bean>
       </property>
    </bean>

    <!-- File Structure -->
    <bean name="FileStructure" class="org.jboss.deployers.vfs.plugins.structure.file.FileStructure">
       <!-- Unless specified the default list of suffixes is -service.xml, -beans.xml, -ds.xml, -aop.xml -->
       <constructor>
          <parameter>
             <set elementClass="java.lang.String">
                <value>-service.xml</value>
                <value>-beans.xml</value>
                <value>-ds.xml</value>
                <value>-aop.xml</value>
             </set>
          </parameter>
       </constructor>
    </bean>

   <!-- POJO Deployment -->
   <bean name="BeanDeployer" class="org.jboss.deployers.vfs.deployer.kernel.BeanDeployer">
      <property name="type">beans</property>
   </bean>
   <bean name="KernelDeploymentDeployer" class="org.jboss.deployers.vfs.deployer.kernel.KernelDeploymentDeployer">
      <property name="type">beans</property>
   </bean>
   <bean name="BeanMetaDataDeployer" class="org.jboss.deployers.vfs.deployer.kernel.BeanMetaDataDeployer">
      <constructor><parameter><inject bean="jboss.kernel:service=Kernel"/></parameter></constructor>
      <property name="type">beans</property>
   </bean>

   <!-- JMX Deployment -->
   <bean name="SARDeployer" class="org.jboss.system.deployers.SARDeployer">
      <property name="type">sar</property>
   </bean>
   <bean name="ServiceClassLoaderDeployer" class="org.jboss.embedded.ClassLoaderDeployer"/>
   <bean name="ServiceDeploymentDeployer" class="org.jboss.system.deployers.ServiceDeploymentDeployer">
       <property name="type">sar</property>
   </bean>
   <bean name="ServiceDeployer" class="org.jboss.system.deployers.ServiceDeployer">
       <constructor><parameter><inject bean="JMXKernel" property="serviceController"/></parameter></constructor>
       <property name="type">sar</property>
   </bean>

   <!-- A filter for excluding files from the scanner -->
   <bean name="DeploymentFilter" class="org.jboss.virtual.plugins.vfs.helpers.ExtensibleFilter">
      <!-- Files starting with theses strings are ignored -->
      <property name="prefixes">#,%,\,,.,_$</property>
      <!-- Files ending with theses strings are ignored -->
      <property name="suffixes">#,$,%,~,\,v,.BAK,.bak,.old,.orig,.tmp,.rej,.sh</property>
      <!-- Files matching with theses strings are ignored -->
      <property name="matches">.make.state,.nse_depinfo,CVS,CVS.admin,RCS,RCSLOG,SCCS,TAGS,core,tags</property>
   </bean>

   <!--
        JNDI Initialization of basic beans (NO REMOTING)
     -->
   <bean name="Naming" class="org.jnp.server.SingletonNamingServer"/>
   <bean name="java:comp" class="org.jboss.naming.JavaCompInitializer"/>

   <!-- register Naming bean with dispatcher just in case we want to have a remote JNDI connection -->
   <bean name="JndiRegistration" class="org.jboss.aspects.remoting.DispatcherRegistration">
      <property name="oid">JNDI</property>
      <property name="target"><inject bean="Naming"/></property>
   </bean>

   <!--
        Bootstrap old JMX-based core services
        setup deployers
        setup deploy/ directory
     -->
   <bean name="ResourcesToDeploy" class="org.jboss.embedded.DeploymentScanner">
      <property name="filter">
         <inject bean="DeploymentFilter"/>
      </property>
      <property name="mainDeployer">
         <inject bean="MainDeployer"/>
      </property>
      <property name="kernel">
         <inject bean="jboss.kernel:service=Kernel"/>
      </property>
      <property name="resources">
         <list elementClass="java.lang.String">
            <value>${jboss.embedded.bootstrap.resource.path}conf/jboss-service.xml</value>
         </list>
      </property>
   </bean>

   <bean name="UserTransaction" class="org.jboss.embedded.adapters.LocalOnlyUserTransaction"/>

   <bean name="UserTransactionBinding" class="org.jboss.naming.JndiBinder">
      <property name="target">
         <inject bean="UserTransaction"/>
      </property>
      <property name="bindTo">UserTransaction</property>
      <property name="serializable">false</property>
   </bean>

   <!-- FIXME EMB-2 -->
   <bean name="UserTransactionBinding2" class="org.jboss.naming.JndiBinder">
      <property name="target">
         <inject bean="UserTransaction"/>
      </property>
      <property name="bindTo">java:comp/UserTransaction</property>
      <property name="serializable">false</property>
   </bean>

   <bean name="ResourcesToDeploy2" class="org.jboss.embedded.DeploymentScanner">
      <property name="filter">
         <inject bean="DeploymentFilter"/>
      </property>
      <property name="mainDeployer">
         <inject bean="MainDeployer"/>
      </property>
      <property name="kernel">
         <inject bean="jboss.kernel:service=Kernel"/>
      </property>
      <property name="directoriesByResource">
         <list elementClass="java.lang.String">
            <value>${jboss.embedded.bootstrap.resource.path}conf/jboss-service.xml/../deployers</value>
         </list>
      </property>
   </bean>
   <bean name="ResourcesToDeploy3" class="org.jboss.embedded.DeploymentScanner">
      <property name="filter">
         <inject bean="DeploymentFilter"/>
      </property>
      <property name="mainDeployer">
         <inject bean="MainDeployer"/>
      </property>
      <property name="kernel">
         <inject bean="jboss.kernel:service=Kernel"/>
      </property>
      <property name="directoriesByResource">
         <list elementClass="java.lang.String">
            <value>${jboss.embedded.bootstrap.resource.path}conf/jboss-service.xml/../deploy</value>
         </list>
      </property>
   </bean>
</deployment>


text:
package com.qinyadan.monitor.network.stream;

import com.qinyadan.monitor.network.packet.stream.StreamCreatePacket;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

public class ClientStreamChannel extends StreamChannel {

	public ClientStreamChannel(Channel channel, int streamId, StreamChannelManager streamChannelManager) {
		super(channel, streamId, streamChannelManager);
	}

	public ChannelFuture sendCreate(byte[] payload) {
		assertState(StreamChannelStateCode.CONNECT_AWAIT);

		StreamCreatePacket packet = new StreamCreatePacket(getStreamId(), payload);
		return this.getChannel().write(packet);
	}

	boolean changeStateConnectAwait() {
		return changeStateTo(StreamChannelStateCode.CONNECT_AWAIT);
	}

}

redpajama_set_name: RedPajamaGithub


text:
from django.http import Http404
from django.contrib.auth.decorators import user_passes_test
from django.utils.decorators import method_decorator

from diogenis.schools.models import School

def user_is_school(user):
    try:
        request_user = School.objects.get(user=user)
        return user.is_authenticated() and request_user.is_school
    except:
        return False

class AuthenticatedSchoolMixin(object):

    @method_decorator(user_passes_test(user_is_school, login_url='/login/'))
    def dispatch(self, request, *args, **kwargs):
        username = kwargs.get('username')
        if username != request.user.username: raise Http404
        return super(AuthenticatedSchoolMixin, self).dispatch(request, *args, **kwargs)


text:
#include <stdio.h>
#include <string.h>
#include <stdarg.h>

#include "py/nlr.h"
#include "py/runtime.h"
#include "py/stream.h"
#include "py/mperrno.h"
#include "py/mphal.h"
#include "uart.h"
#include "irq.h"
#include "genhdr/pins.h"

/// \moduleref pyb
/// \class UART - duplex serial communication bus
///
/// UART implements the standard UART/USART duplex serial communications protocol.  At
/// the physical level it consists of 2 lines: RX and TX.  The unit of communication
/// is a character (not to be confused with a string character) which can be 8 or 9
/// bits wide.
///
/// UART objects can be created and initialised using:
///
///     from pyb import UART
///
///     uart = UART(1, 9600)                         # init with given baudrate
///     uart.init(9600, bits=8, parity=None, stop=1) # init with given parameters
///
/// Bits can be 8 or 9.  Parity can be None, 0 (even) or 1 (odd).  Stop can be 1 or 2.
///
/// A UART object acts like a stream object and reading and writing is done
/// using the standard stream methods:
///
///     uart.read(10)       # read 10 characters, returns a bytes object
///     uart.read()         # read all available characters
///     uart.readline()     # read a line
///     uart.readinto(buf)  # read and store into the given buffer
///     uart.write('abc')   # write the 3 characters
///
/// Individual characters can be read/written using:
///
///     uart.readchar()     # read 1 character and returns it as an integer
///     uart.writechar(42)  # write 1 character
///
/// To check if there is anything to be read, use:
///
///     uart.any()               # returns True if any characters waiting

#define CHAR_WIDTH_8BIT (0)
#define CHAR_WIDTH_9BIT (1)

struct _pyb_uart_obj_t {
    mp_obj_base_t base;
    UART_HandleTypeDef uart;            // this is 17 words big
    IRQn_Type irqn;
    pyb_uart_t uart_id : 8;
    bool is_enabled : 1;
    byte char_width;                    // 0 for 7,8 bit chars, 1 for 9 bit chars
    uint16_t char_mask;                 // 0x7f for 7 bit, 0xff for 8 bit, 0x1ff for 9 bit
    uint16_t timeout;                   // timeout waiting for first char
    uint16_t timeout_char;              // timeout waiting between chars
    uint16_t read_buf_len;              // len in chars; buf can hold len-1 chars
    volatile uint16_t read_buf_head;    // indexes first empty slot
    uint16_t read_buf_tail;             // indexes first full slot (not full if equals head)
    byte *read_buf;                     // byte or uint16_t, depending on char size
};

STATIC mp_obj_t pyb_uart_deinit(mp_obj_t self_in);

void uart_init0(void) {
    for (int i = 0; i < MP_ARRAY_SIZE(MP_STATE_PORT(pyb_uart_obj_all)); i++) {
        MP_STATE_PORT(pyb_uart_obj_all)[i] = NULL;
    }
}

// unregister all interrupt sources
void uart_deinit(void) {
    for (int i = 0; i < MP_ARRAY_SIZE(MP_STATE_PORT(pyb_uart_obj_all)); i++) {
        pyb_uart_obj_t *uart_obj = MP_STATE_PORT(pyb_uart_obj_all)[i];
        if (uart_obj != NULL) {
            pyb_uart_deinit(uart_obj);
        }
    }
}

STATIC bool uart_exists(int uart_id) {
    if (uart_id > MP_ARRAY_SIZE(MP_STATE_PORT(pyb_uart_obj_all))) {
        // safeguard against pyb_uart_obj_all array being configured too small
        return false;
    }
    switch (uart_id) {
        #if defined(MICROPY_HW_UART1_TX) && defined(MICROPY_HW_UART1_RX)
        case PYB_UART_1: return true;
        #endif

        #if defined(MICROPY_HW_UART2_TX) && defined(MICROPY_HW_UART2_RX)
        case PYB_UART_2: return true;
        #endif

        #if defined(MICROPY_HW_UART3_TX) && defined(MICROPY_HW_UART3_RX)
        case PYB_UART_3: return true;
        #endif

        #if defined(MICROPY_HW_UART4_TX) && defined(MICROPY_HW_UART4_RX)
        case PYB_UART_4: return true;
        #endif

        #if defined(MICROPY_HW_UART5_TX) && defined(MICROPY_HW_UART5_RX)
        case PYB_UART_5: return true;
        #endif

        #if defined(MICROPY_HW_UART6_TX) && defined(MICROPY_HW_UART6_RX)
        case PYB_UART_6: return true;
        #endif

        #if defined(MICROPY_HW_UART7_TX) && defined(MICROPY_HW_UART7_RX)
        case PYB_UART_7: return true;
        #endif

        #if defined(MICROPY_HW_UART8_TX) && defined(MICROPY_HW_UART8_RX)
        case PYB_UART_8: return true;
        #endif

        default: return false;
    }
}

// assumes Init parameters have been set up correctly
STATIC bool uart_init2(pyb_uart_obj_t *uart_obj) {
    USART_TypeDef *UARTx;
    IRQn_Type irqn;
    int uart_unit;

    const pin_obj_t *pins[4] = {0};

    switch (uart_obj->uart_id) {
        #if defined(MICROPY_HW_UART1_TX) && defined(MICROPY_HW_UART1_RX)
        case PYB_UART_1:
            uart_unit = 1;
            UARTx = USART1;
            irqn = USART1_IRQn;
            pins[0] = &MICROPY_HW_UART1_TX;
            pins[1] = &MICROPY_HW_UART1_RX;
            __USART1_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART2_TX) && defined(MICROPY_HW_UART2_RX)
        case PYB_UART_2:
            uart_unit = 2;
            UARTx = USART2;
            irqn = USART2_IRQn;
            pins[0] = &MICROPY_HW_UART2_TX;
            pins[1] = &MICROPY_HW_UART2_RX;
            #if defined(MICROPY_HW_UART2_RTS)
            if (uart_obj->uart.Init.HwFlowCtl & UART_HWCONTROL_RTS) {
                pins[2] = &MICROPY_HW_UART2_RTS;
            }
            #endif
            #if defined(MICROPY_HW_UART2_CTS)
            if (uart_obj->uart.Init.HwFlowCtl & UART_HWCONTROL_CTS) {
                pins[3] = &MICROPY_HW_UART2_CTS;
            }
            #endif
            __USART2_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART3_TX) && defined(MICROPY_HW_UART3_RX)
        case PYB_UART_3:
            uart_unit = 3;
            UARTx = USART3;
            irqn = USART3_IRQn;
            pins[0] = &MICROPY_HW_UART3_TX;
            pins[1] = &MICROPY_HW_UART3_RX;
            #if defined(MICROPY_HW_UART3_RTS)
            if (uart_obj->uart.Init.HwFlowCtl & UART_HWCONTROL_RTS) {
                pins[2] = &MICROPY_HW_UART3_RTS;
            }
            #endif
            #if defined(MICROPY_HW_UART3_CTS)
            if (uart_obj->uart.Init.HwFlowCtl & UART_HWCONTROL_CTS) {
                pins[3] = &MICROPY_HW_UART3_CTS;
            }
            #endif
            __USART3_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART4_TX) && defined(MICROPY_HW_UART4_RX)
        case PYB_UART_4:
            uart_unit = 4;
            UARTx = UART4;
            irqn = UART4_IRQn;
            pins[0] = &MICROPY_HW_UART4_TX;
            pins[1] = &MICROPY_HW_UART4_RX;
            __UART4_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART5_TX) && defined(MICROPY_HW_UART5_RX)
        case PYB_UART_5:
            uart_unit = 5;
            UARTx = UART5;
            irqn = UART5_IRQn;
            pins[0] = &MICROPY_HW_UART5_TX;
            pins[1] = &MICROPY_HW_UART5_RX;
            __UART5_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART6_TX) && defined(MICROPY_HW_UART6_RX)
        case PYB_UART_6:
            uart_unit = 6;
            UARTx = USART6;
            irqn = USART6_IRQn;
            pins[0] = &MICROPY_HW_UART6_TX;
            pins[1] = &MICROPY_HW_UART6_RX;
            __USART6_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART7_TX) && defined(MICROPY_HW_UART7_RX)
        case PYB_UART_7:
            uart_unit = 7;
            UARTx = UART7;
            irqn = UART7_IRQn;
            pins[0] = &MICROPY_HW_UART7_TX;
            pins[1] = &MICROPY_HW_UART7_RX;
            __UART7_CLK_ENABLE();
            break;
        #endif

        #if defined(MICROPY_HW_UART8_TX) && defined(MICROPY_HW_UART8_RX)
        case PYB_UART_8:
            uart_unit = 8;
            UARTx = UART8;
            irqn = UART8_IRQn;
            pins[0] = &MICROPY_HW_UART8_TX;
            pins[1] = &MICROPY_HW_UART8_RX;
            __UART8_CLK_ENABLE();
            break;
        #endif

        default:
            // UART does not exist or is not configured for this board
            return false;
    }

    uint32_t mode = MP_HAL_PIN_MODE_ALT;
    uint32_t pull = MP_HAL_PIN_PULL_UP;

    for (uint i = 0; i < 4; i++) {
        if (pins[i] != NULL) {
            bool ret = mp_hal_pin_config_alt(pins[i], mode, pull, AF_FN_UART, uart_unit);
            if (!ret) {
                return false;
            }
        }
    }

    uart_obj->irqn = irqn;
    uart_obj->uart.Instance = UARTx;

    // init UARTx
    HAL_UART_Init(&uart_obj->uart);

    uart_obj->is_enabled = true;

    return true;
}

/* obsolete and unused
bool uart_init(pyb_uart_obj_t *uart_obj, uint32_t baudrate) {
    UART_HandleTypeDef *uh = &uart_obj->uart;
    memset(uh, 0, sizeof(*uh));
    uh->Init.BaudRate = baudrate;
    uh->Init.WordLength = UART_WORDLENGTH_8B;
    uh->Init.StopBits = UART_STOPBITS_1;
    uh->Init.Parity = UART_PARITY_NONE;
    uh->Init.Mode = UART_MODE_TX_RX;
    uh->Init.HwFlowCtl = UART_HWCONTROL_NONE;
    uh->Init.OverSampling = UART_OVERSAMPLING_16;
    return uart_init2(uart_obj);
}
*/

mp_uint_t uart_rx_any(pyb_uart_obj_t *self) {
    int buffer_bytes = self->read_buf_head - self->read_buf_tail;
    if (buffer_bytes < 0) {
        return buffer_bytes + self->read_buf_len;
    } else if (buffer_bytes > 0) {
        return buffer_bytes;
    } else {
        return __HAL_UART_GET_FLAG(&self->uart, UART_FLAG_RXNE) != RESET;
    }
}

// Waits at most timeout milliseconds for at least 1 char to become ready for
// reading (from buf or for direct reading).
// Returns true if something available, false if not.
STATIC bool uart_rx_wait(pyb_uart_obj_t *self, uint32_t timeout) {
    uint32_t start = HAL_GetTick();
    for (;;) {
        if (self->read_buf_tail != self->read_buf_head || __HAL_UART_GET_FLAG(&self->uart, UART_FLAG_RXNE) != RESET) {
            return true; // have at least 1 char ready for reading
        }
        if (HAL_GetTick() - start >= timeout) {
            return false; // timeout
        }
        MICROPY_EVENT_POLL_HOOK
    }
}

// assumes there is a character available
int uart_rx_char(pyb_uart_obj_t *self) {
    if (self->read_buf_tail != self->read_buf_head) {
        // buffering via IRQ
        int data;
        if (self->char_width == CHAR_WIDTH_9BIT) {
            data = ((uint16_t*)self->read_buf)[self->read_buf_tail];
        } else {
            data = self->read_buf[self->read_buf_tail];
        }
        self->read_buf_tail = (self->read_buf_tail + 1) % self->read_buf_len;
        if (__HAL_UART_GET_FLAG(&self->uart, UART_FLAG_RXNE) != RESET) {
            // UART was stalled by flow ctrl: re-enable IRQ now we have room in buffer
            __HAL_UART_ENABLE_IT(&self->uart, UART_IT_RXNE);
        }
        return data;
    } else {
        // no buffering
        #if defined(MCU_SERIES_F7) || defined(MCU_SERIES_L4)
        return self->uart.Instance->RDR & self->char_mask;
        #else
        return self->uart.Instance->DR & self->char_mask;
        #endif
    }
}

// Waits at most timeout milliseconds for TX register to become empty.
// Returns true if can write, false if can't.
STATIC bool uart_tx_wait(pyb_uart_obj_t *self, uint32_t timeout) {
    uint32_t start = HAL_GetTick();
    for (;;) {
        if (__HAL_UART_GET_FLAG(&self->uart, UART_FLAG_TXE)) {
            return true; // tx register is empty
        }
        if (HAL_GetTick() - start >= timeout) {
            return false; // timeout
        }
        MICROPY_EVENT_POLL_HOOK
    }
}

// Waits at most timeout milliseconds for UART flag to be set.
// Returns true if flag is/was set, false on timeout.
STATIC bool uart_wait_flag_set(pyb_uart_obj_t *self, uint32_t flag, uint32_t timeout) {
    // Note: we don't use WFI to idle in this loop because UART tx doesn't generate
    // an interrupt and the flag can be set quickly if the baudrate is large.
    uint32_t start = HAL_GetTick();
    for (;;) {
        if (__HAL_UART_GET_FLAG(&self->uart, flag)) {
            return true;
        }
        if (timeout == 0 || HAL_GetTick() - start >= timeout) {
            return false; // timeout
        }
    }
}

// src - a pointer to the data to send (16-bit aligned for 9-bit chars)
// num_chars - number of characters to send (9-bit chars count for 2 bytes from src)
// *errcode - returns 0 for success, MP_Exxx on error
// returns the number of characters sent (valid even if there was an error)
STATIC size_t uart_tx_data(pyb_uart_obj_t *self, const void *src_in, size_t num_chars, int *errcode) {
    if (num_chars == 0) {
        *errcode = 0;
        return 0;
    }

    uint32_t timeout;
    if (self->uart.Init.HwFlowCtl & UART_HWCONTROL_CTS) {
        // CTS can hold off transmission for an arbitrarily long time. Apply
        // the overall timeout rather than the character timeout.
        timeout = self->timeout;
    } else {
        // The timeout specified here is for waiting for the TX data register to
        // become empty (ie between chars), as well as for the final char to be
        // completely transferred.  The default value for timeout_char is long
        // enough for 1 char, but we need to double it to wait for the last char
        // to be transferred to the data register, and then to be transmitted.
        timeout = 2 * self->timeout_char;
    }

    const uint8_t *src = (const uint8_t*)src_in;
    size_t num_tx = 0;
    USART_TypeDef *uart = self->uart.Instance;

    while (num_tx < num_chars) {
        if (!uart_wait_flag_set(self, UART_FLAG_TXE, timeout)) {
            *errcode = MP_ETIMEDOUT;
            return num_tx;
        }
        uint32_t data;
        if (self->char_width == CHAR_WIDTH_9BIT) {
            data = *((uint16_t*)src) & 0x1ff;
            src += 2;
        } else {
            data = *src++;
        }
        #if defined(MCU_SERIES_F4)
        uart->DR = data;
        #else
        uart->TDR = data;
        #endif
        ++num_tx;
    }

    // wait for the UART frame to complete
    if (!uart_wait_flag_set(self, UART_FLAG_TC, timeout)) {
        *errcode = MP_ETIMEDOUT;
        return num_tx;
    }

    *errcode = 0;
    return num_tx;
}

STATIC void uart_tx_char(pyb_uart_obj_t *uart_obj, int c) {
    uint16_t ch = c;
    int errcode;
    uart_tx_data(uart_obj, &ch, 1, &errcode);
}

void uart_tx_strn(pyb_uart_obj_t *uart_obj, const char *str, uint len) {
    int errcode;
    uart_tx_data(uart_obj, str, len, &errcode);
}

void uart_tx_strn_cooked(pyb_uart_obj_t *uart_obj, const char *str, uint len) {
    for (const char *top = str + len; str < top; str++) {
        if (*str == '\n') {
            uart_tx_char(uart_obj, '\r');
        }
        uart_tx_char(uart_obj, *str);
    }
}

// this IRQ handler is set up to handle RXNE interrupts only
void uart_irq_handler(mp_uint_t uart_id) {
    // get the uart object
    pyb_uart_obj_t *self = MP_STATE_PORT(pyb_uart_obj_all)[uart_id - 1];

    if (self == NULL) {
        // UART object has not been set, so we can't do anything, not
        // even disable the IRQ.  This should never happen.
        return;
    }

    if (__HAL_UART_GET_FLAG(&self->uart, UART_FLAG_RXNE) != RESET) {
        if (self->read_buf_len != 0) {
            uint16_t next_head = (self->read_buf_head + 1) % self->read_buf_len;
            if (next_head != self->read_buf_tail) {
                // only read data if room in buf
                #if defined(MCU_SERIES_F7) || defined(MCU_SERIES_L4)
                int data = self->uart.Instance->RDR; // clears UART_FLAG_RXNE
                #else
                int data = self->uart.Instance->DR; // clears UART_FLAG_RXNE
                #endif
                data &= self->char_mask;
                if (self->char_width == CHAR_WIDTH_9BIT) {
                    ((uint16_t*)self->read_buf)[self->read_buf_head] = data;
                } else {
                    self->read_buf[self->read_buf_head] = data;
                }
                self->read_buf_head = next_head;
            } else { // No room: leave char in buf, disable interrupt
                __HAL_UART_DISABLE_IT(&self->uart, UART_IT_RXNE);
            }
        }
    }
}

/******************************************************************************/
/* Micro Python bindings                                                      */

STATIC void pyb_uart_print(const mp_print_t *print, mp_obj_t self_in, mp_print_kind_t kind) {
    pyb_uart_obj_t *self = self_in;
    if (!self->is_enabled) {
        mp_printf(print, "UART(%u)", self->uart_id);
    } else {
        mp_int_t bits = (self->uart.Init.WordLength == UART_WORDLENGTH_8B ? 8 : 9);
        if (self->uart.Init.Parity != UART_PARITY_NONE) {
            bits -= 1;
        }
        mp_printf(print, "UART(%u, baudrate=%u, bits=%u, parity=",
            self->uart_id, self->uart.Init.BaudRate, bits);
        if (self->uart.Init.Parity == UART_PARITY_NONE) {
            mp_print_str(print, "None");
        } else {
            mp_printf(print, "%u", self->uart.Init.Parity == UART_PARITY_EVEN ? 0 : 1);
        }
        if (self->uart.Init.HwFlowCtl) {
            mp_printf(print, ", flow=");
            if (self->uart.Init.HwFlowCtl & UART_HWCONTROL_RTS) {
                mp_printf(print, "RTS%s", self->uart.Init.HwFlowCtl & UART_HWCONTROL_CTS ? "|" : "");
            }
            if (self->uart.Init.HwFlowCtl & UART_HWCONTROL_CTS) {
                mp_printf(print, "CTS");
            }
        }
        mp_printf(print, ", stop=%u, timeout=%u, timeout_char=%u, read_buf_len=%u)",
            self->uart.Init.StopBits == UART_STOPBITS_1 ? 1 : 2,
            self->timeout, self->timeout_char,
            self->read_buf_len == 0 ? 0 : self->read_buf_len - 1); // -1 to adjust for usable length of buffer
    }
}

/// \method init(baudrate, bits=8, parity=None, stop=1, *, timeout=1000, timeout_char=0, flow=0, read_buf_len=64)
///
/// Initialise the UART bus with the given parameters:
///
///   - `baudrate` is the clock rate.
///   - `bits` is the number of bits per byte, 7, 8 or 9.
///   - `parity` is the parity, `None`, 0 (even) or 1 (odd).
///   - `stop` is the number of stop bits, 1 or 2.
///   - `timeout` is the timeout in milliseconds to wait for the first character.
///   - `timeout_char` is the timeout in milliseconds to wait between characters.
///   - `flow` is RTS | CTS where RTS == 256, CTS == 512
///   - `read_buf_len` is the character length of the read buffer (0 to disable).
STATIC mp_obj_t pyb_uart_init_helper(pyb_uart_obj_t *self, mp_uint_t n_args, const mp_obj_t *pos_args, mp_map_t *kw_args) {
    static const mp_arg_t allowed_args[] = {
        { MP_QSTR_baudrate, MP_ARG_REQUIRED | MP_ARG_INT, {.u_int = 9600} },
        { MP_QSTR_bits, MP_ARG_INT, {.u_int = 8} },
        { MP_QSTR_parity, MP_ARG_OBJ, {.u_obj = mp_const_none} },
        { MP_QSTR_stop, MP_ARG_INT, {.u_int = 1} },
        { MP_QSTR_flow, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = UART_HWCONTROL_NONE} },
        { MP_QSTR_timeout, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 1000} },
        { MP_QSTR_timeout_char, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 0} },
        { MP_QSTR_read_buf_len, MP_ARG_KW_ONLY | MP_ARG_INT, {.u_int = 64} },
    };

    // parse args
    struct {
        mp_arg_val_t baudrate, bits, parity, stop, flow, timeout, timeout_char, read_buf_len;
    } args;
    mp_arg_parse_all(n_args, pos_args, kw_args,
        MP_ARRAY_SIZE(allowed_args), allowed_args, (mp_arg_val_t*)&args);

    // set the UART configuration values
    memset(&self->uart, 0, sizeof(self->uart));
    UART_InitTypeDef *init = &self->uart.Init;

    // baudrate
    init->BaudRate = args.baudrate.u_int;

    // parity
    mp_int_t bits = args.bits.u_int;
    if (args.parity.u_obj == mp_const_none) {
        init->Parity = UART_PARITY_NONE;
    } else {
        mp_int_t parity = mp_obj_get_int(args.parity.u_obj);
        init->Parity = (parity & 1) ? UART_PARITY_ODD : UART_PARITY_EVEN;
        bits += 1; // STs convention has bits including parity
    }

    // number of bits
    if (bits == 8) {
        init->WordLength = UART_WORDLENGTH_8B;
    } else if (bits == 9) {
        init->WordLength = UART_WORDLENGTH_9B;
    } else {
        nlr_raise(mp_obj_new_exception_msg_varg(&mp_type_ValueError, "unsupported combination of bits and parity"));
    }

    // stop bits
    switch (args.stop.u_int) {
        case 1: init->StopBits = UART_STOPBITS_1; break;
        default: init->StopBits = UART_STOPBITS_2; break;
    }

    // flow control
    init->HwFlowCtl = args.flow.u_int;

    // extra config (not yet configurable)
    init->Mode = UART_MODE_TX_RX;
    init->OverSampling = UART_OVERSAMPLING_16;

    // init UART (if it fails, it's because the port doesn't exist)
    if (!uart_init2(self)) {
        nlr_raise(mp_obj_new_exception_msg_varg(&mp_type_ValueError, "UART(%d) does not exist", self->uart_id));
    }

    // set timeout
    self->timeout = args.timeout.u_int;

    // set timeout_char
    // make sure it is at least as long as a whole character (13 bits to be safe)
    // minimum value is 2ms because sys-tick has a resolution of only 1ms
    self->timeout_char = args.timeout_char.u_int;
    uint32_t min_timeout_char = 13000 / init->BaudRate + 2;
    if (self->timeout_char < min_timeout_char) {
        self->timeout_char = min_timeout_char;
    }

    // setup the read buffer
    m_del(byte, self->read_buf, self->read_buf_len << self->char_width);
    if (init->WordLength == UART_WORDLENGTH_9B && init->Parity == UART_PARITY_NONE) {
        self->char_mask = 0x1ff;
        self->char_width = CHAR_WIDTH_9BIT;
    } else {
        if (init->WordLength == UART_WORDLENGTH_9B || init->Parity == UART_PARITY_NONE) {
            self->char_mask = 0xff;
        } else {
            self->char_mask = 0x7f;
        }
        self->char_width = CHAR_WIDTH_8BIT;
    }
    self->read_buf_head = 0;
    self->read_buf_tail = 0;
    if (args.read_buf_len.u_int <= 0) {
        // no read buffer
        self->read_buf_len = 0;
        self->read_buf = NULL;
        HAL_NVIC_DisableIRQ(self->irqn);
        __HAL_UART_DISABLE_IT(&self->uart, UART_IT_RXNE);
    } else {
        // read buffer using interrupts
        self->read_buf_len = args.read_buf_len.u_int + 1; // +1 to adjust for usable length of buffer
        self->read_buf = m_new(byte, self->read_buf_len << self->char_width);
        __HAL_UART_ENABLE_IT(&self->uart, UART_IT_RXNE);
        HAL_NVIC_SetPriority(self->irqn, IRQ_PRI_UART, IRQ_SUBPRI_UART);
        HAL_NVIC_EnableIRQ(self->irqn);
    }

    // compute actual baudrate that was configured
    // (this formula assumes UART_OVERSAMPLING_16)
    uint32_t actual_baudrate = 0;
    #if defined(MCU_SERIES_F7)
    UART_ClockSourceTypeDef clocksource = UART_CLOCKSOURCE_UNDEFINED;
    UART_GETCLOCKSOURCE(&self->uart, clocksource);
    switch (clocksource) {
        case UART_CLOCKSOURCE_PCLK1:  actual_baudrate = HAL_RCC_GetPCLK1Freq(); break;
        case UART_CLOCKSOURCE_PCLK2:  actual_baudrate = HAL_RCC_GetPCLK2Freq(); break;
        case UART_CLOCKSOURCE_HSI:    actual_baudrate = HSI_VALUE; break;
        case UART_CLOCKSOURCE_SYSCLK: actual_baudrate = HAL_RCC_GetSysClockFreq(); break;
        case UART_CLOCKSOURCE_LSE:    actual_baudrate = LSE_VALUE; break;
        case UART_CLOCKSOURCE_UNDEFINED: break;
    }
    #else
    if (self->uart.Instance == USART1
        #if defined(USART6)
        || self->uart.Instance == USART6
        #endif
        ) {
        actual_baudrate = HAL_RCC_GetPCLK2Freq();
    } else {
        actual_baudrate = HAL_RCC_GetPCLK1Freq();
    }
    #endif
    actual_baudrate /= self->uart.Instance->BRR;

    // check we could set the baudrate within 5%
    uint32_t baudrate_diff;
    if (actual_baudrate > init->BaudRate) {
        baudrate_diff = actual_baudrate - init->BaudRate;
    } else {
        baudrate_diff = init->BaudRate - actual_baudrate;
    }
    init->BaudRate = actual_baudrate; // remember actual baudrate for printing
    if (20 * baudrate_diff > init->BaudRate) {
        nlr_raise(mp_obj_new_exception_msg_varg(&mp_type_ValueError, "set baudrate %d is not within 5%% of desired value", actual_baudrate));
    }

    return mp_const_none;
}

/// \classmethod \constructor(bus, ...)
///
/// Construct a UART object on the given bus.  `bus` can be 1-6, or 'XA', 'XB', 'YA', or 'YB'.
/// With no additional parameters, the UART object is created but not
/// initialised (it has the settings from the last initialisation of
/// the bus, if any).  If extra arguments are given, the bus is initialised.
/// See `init` for parameters of initialisation.
///
/// The physical pins of the UART busses are:
///
///   - `UART(4)` is on `XA`: `(TX, RX) = (X1, X2) = (PA0, PA1)`
///   - `UART(1)` is on `XB`: `(TX, RX) = (X9, X10) = (PB6, PB7)`
///   - `UART(6)` is on `YA`: `(TX, RX) = (Y1, Y2) = (PC6, PC7)`
///   - `UART(3)` is on `YB`: `(TX, RX) = (Y9, Y10) = (PB10, PB11)`
///   - `UART(2)` is on: `(TX, RX) = (X3, X4) = (PA2, PA3)`
STATIC mp_obj_t pyb_uart_make_new(const mp_obj_type_t *type, size_t n_args, size_t n_kw, const mp_obj_t *args) {
    // check arguments
    mp_arg_check_num(n_args, n_kw, 1, MP_OBJ_FUN_ARGS_MAX, true);

    // work out port
    int uart_id = 0;
    if (MP_OBJ_IS_STR(args[0])) {
        const char *port = mp_obj_str_get_str(args[0]);
        if (0) {
        #ifdef MICROPY_HW_UART1_NAME
        } else if (strcmp(port, MICROPY_HW_UART1_NAME) == 0) {
            uart_id = PYB_UART_1;
        #endif
        #ifdef MICROPY_HW_UART2_NAME
        } else if (strcmp(port, MICROPY_HW_UART2_NAME) == 0) {
            uart_id = PYB_UART_2;
        #endif
        #ifdef MICROPY_HW_UART3_NAME
        } else if (strcmp(port, MICROPY_HW_UART3_NAME) == 0) {
            uart_id = PYB_UART_3;
        #endif
        #ifdef MICROPY_HW_UART4_NAME
        } else if (strcmp(port, MICROPY_HW_UART4_NAME) == 0) {
            uart_id = PYB_UART_4;
        #endif
        #ifdef MICROPY_HW_UART5_NAME
        } else if (strcmp(port, MICROPY_HW_UART5_NAME) == 0) {
            uart_id = PYB_UART_5;
        #endif
        #ifdef MICROPY_HW_UART6_NAME
        } else if (strcmp(port, MICROPY_HW_UART6_NAME) == 0) {
            uart_id = PYB_UART_6;
        #endif
        } else {
            nlr_raise(mp_obj_new_exception_msg_varg(&mp_type_ValueError, "UART(%s) does not exist", port));
        }
    } else {
        uart_id = mp_obj_get_int(args[0]);
        if (!uart_exists(uart_id)) {
            nlr_raise(mp_obj_new_exception_msg_varg(&mp_type_ValueError, "UART(%d) does not exist", uart_id));
        }
    }

    pyb_uart_obj_t *self;
    if (MP_STATE_PORT(pyb_uart_obj_all)[uart_id - 1] == NULL) {
        // create new UART object
        self = m_new0(pyb_uart_obj_t, 1);
        self->base.type = &pyb_uart_type;
        self->uart_id = uart_id;
        MP_STATE_PORT(pyb_uart_obj_all)[uart_id - 1] = self;
    } else {
        // reference existing UART object
        self = MP_STATE_PORT(pyb_uart_obj_all)[uart_id - 1];
    }

    if (n_args > 1 || n_kw > 0) {
        // start the peripheral
        mp_map_t kw_args;
        mp_map_init_fixed_table(&kw_args, n_kw, args + n_args);
        pyb_uart_init_helper(self, n_args - 1, args + 1, &kw_args);
    }

    return self;
}

STATIC mp_obj_t pyb_uart_init(mp_uint_t n_args, const mp_obj_t *args, mp_map_t *kw_args) {
    return pyb_uart_init_helper(args[0], n_args - 1, args + 1, kw_args);
}
STATIC MP_DEFINE_CONST_FUN_OBJ_KW(pyb_uart_init_obj, 1, pyb_uart_init);

/// \method deinit()
/// Turn off the UART bus.
STATIC mp_obj_t pyb_uart_deinit(mp_obj_t self_in) {
    pyb_uart_obj_t *self = self_in;
    self->is_enabled = false;
    UART_HandleTypeDef *uart = &self->uart;
    HAL_UART_DeInit(uart);
    if (uart->Instance == USART1) {
        HAL_NVIC_DisableIRQ(USART1_IRQn);
        __USART1_FORCE_RESET();
        __USART1_RELEASE_RESET();
        __USART1_CLK_DISABLE();
    } else if (uart->Instance == USART2) {
        HAL_NVIC_DisableIRQ(USART2_IRQn);
        __USART2_FORCE_RESET();
        __USART2_RELEASE_RESET();
        __USART2_CLK_DISABLE();
    #if defined(USART3)
    } else if (uart->Instance == USART3) {
        HAL_NVIC_DisableIRQ(USART3_IRQn);
        __USART3_FORCE_RESET();
        __USART3_RELEASE_RESET();
        __USART3_CLK_DISABLE();
    #endif
    #if defined(UART4)
    } else if (uart->Instance == UART4) {
        HAL_NVIC_DisableIRQ(UART4_IRQn);
        __UART4_FORCE_RESET();
        __UART4_RELEASE_RESET();
        __UART4_CLK_DISABLE();
    #endif
    #if defined(UART5)
    } else if (uart->Instance == UART5) {
        HAL_NVIC_DisableIRQ(UART5_IRQn);
        __UART5_FORCE_RESET();
        __UART5_RELEASE_RESET();
        __UART5_CLK_DISABLE();
    #endif
    #if defined(UART6)
    } else if (uart->Instance == USART6) {
        HAL_NVIC_DisableIRQ(USART6_IRQn);
        __USART6_FORCE_RESET();
        __USART6_RELEASE_RESET();
        __USART6_CLK_DISABLE();
    #endif
    #if defined(UART7)
    } else if (uart->Instance == UART7) {
        HAL_NVIC_DisableIRQ(UART7_IRQn);
        __UART7_FORCE_RESET();
        __UART7_RELEASE_RESET();
        __UART7_CLK_DISABLE();
    #endif
    #if defined(UART8)
    } else if (uart->Instance == UART8) {
        HAL_NVIC_DisableIRQ(UART8_IRQn);
        __UART8_FORCE_RESET();
        __UART8_RELEASE_RESET();
        __UART8_CLK_DISABLE();
    #endif
    }
    return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(pyb_uart_deinit_obj, pyb_uart_deinit);

/// \method any()
/// Return `True` if any characters waiting, else `False`.
STATIC mp_obj_t pyb_uart_any(mp_obj_t self_in) {
    pyb_uart_obj_t *self = self_in;
    return MP_OBJ_NEW_SMALL_INT(uart_rx_any(self));
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(pyb_uart_any_obj, pyb_uart_any);

/// \method writechar(char)
/// Write a single character on the bus.  `char` is an integer to write.
/// Return value: `None`.
STATIC mp_obj_t pyb_uart_writechar(mp_obj_t self_in, mp_obj_t char_in) {
    pyb_uart_obj_t *self = self_in;

    // get the character to write (might be 9 bits)
    uint16_t data = mp_obj_get_int(char_in);

    // write the character
    int errcode;
    if (uart_tx_wait(self, self->timeout)) {
        uart_tx_data(self, &data, 1, &errcode);
    } else {
        errcode = MP_ETIMEDOUT;
    }

    if (errcode != 0) {
        mp_raise_OSError(errcode);
    }

    return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_2(pyb_uart_writechar_obj, pyb_uart_writechar);

/// \method readchar()
/// Receive a single character on the bus.
/// Return value: The character read, as an integer.  Returns -1 on timeout.
STATIC mp_obj_t pyb_uart_readchar(mp_obj_t self_in) {
    pyb_uart_obj_t *self = self_in;
    if (uart_rx_wait(self, self->timeout)) {
        return MP_OBJ_NEW_SMALL_INT(uart_rx_char(self));
    } else {
        // return -1 on timeout
        return MP_OBJ_NEW_SMALL_INT(-1);
    }
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(pyb_uart_readchar_obj, pyb_uart_readchar);

// uart.sendbreak()
STATIC mp_obj_t pyb_uart_sendbreak(mp_obj_t self_in) {
    pyb_uart_obj_t *self = self_in;
    #if defined(MCU_SERIES_F7) || defined(MCU_SERIES_L4)
    self->uart.Instance->RQR = USART_RQR_SBKRQ; // write-only register
    #else
    self->uart.Instance->CR1 |= USART_CR1_SBK;
    #endif
    return mp_const_none;
}
STATIC MP_DEFINE_CONST_FUN_OBJ_1(pyb_uart_sendbreak_obj, pyb_uart_sendbreak);

STATIC const mp_map_elem_t pyb_uart_locals_dict_table[] = {
    // instance methods

    { MP_OBJ_NEW_QSTR(MP_QSTR_init), (mp_obj_t)&pyb_uart_init_obj },
    { MP_OBJ_NEW_QSTR(MP_QSTR_deinit), (mp_obj_t)&pyb_uart_deinit_obj },
    { MP_OBJ_NEW_QSTR(MP_QSTR_any), (mp_obj_t)&pyb_uart_any_obj },

    /// \method read([nbytes])
    { MP_OBJ_NEW_QSTR(MP_QSTR_read), (mp_obj_t)&mp_stream_read_obj },
    /// \method readline()
    { MP_OBJ_NEW_QSTR(MP_QSTR_readline), (mp_obj_t)&mp_stream_unbuffered_readline_obj},
    /// \method readinto(buf[, nbytes])
    { MP_OBJ_NEW_QSTR(MP_QSTR_readinto), (mp_obj_t)&mp_stream_readinto_obj },
    /// \method write(buf)
    { MP_OBJ_NEW_QSTR(MP_QSTR_write), (mp_obj_t)&mp_stream_write_obj },

    { MP_OBJ_NEW_QSTR(MP_QSTR_writechar), (mp_obj_t)&pyb_uart_writechar_obj },
    { MP_OBJ_NEW_QSTR(MP_QSTR_readchar), (mp_obj_t)&pyb_uart_readchar_obj },
    { MP_OBJ_NEW_QSTR(MP_QSTR_sendbreak), (mp_obj_t)&pyb_uart_sendbreak_obj },

    // class constants
    { MP_OBJ_NEW_QSTR(MP_QSTR_RTS), MP_OBJ_NEW_SMALL_INT(UART_HWCONTROL_RTS) },
    { MP_OBJ_NEW_QSTR(MP_QSTR_CTS), MP_OBJ_NEW_SMALL_INT(UART_HWCONTROL_CTS) },
};

STATIC MP_DEFINE_CONST_DICT(pyb_uart_locals_dict, pyb_uart_locals_dict_table);

STATIC mp_uint_t pyb_uart_read(mp_obj_t self_in, void *buf_in, mp_uint_t size, int *errcode) {
    pyb_uart_obj_t *self = self_in;
    byte *buf = buf_in;

    // check that size is a multiple of character width
    if (size & self->char_width) {
        *errcode = MP_EIO;
        return MP_STREAM_ERROR;
    }

    // convert byte size to char size
    size >>= self->char_width;

    // make sure we want at least 1 char
    if (size == 0) {
        return 0;
    }

    // wait for first char to become available
    if (!uart_rx_wait(self, self->timeout)) {
        // return EAGAIN error to indicate non-blocking (then read() method returns None)
        *errcode = MP_EAGAIN;
        return MP_STREAM_ERROR;
    }

    // read the data
    byte *orig_buf = buf;
    for (;;) {
        int data = uart_rx_char(self);
        if (self->char_width == CHAR_WIDTH_9BIT) {
            *(uint16_t*)buf = data;
            buf += 2;
        } else {
            *buf++ = data;
        }
        if (--size == 0 || !uart_rx_wait(self, self->timeout_char)) {
            // return number of bytes read
            return buf - orig_buf;
        }
    }
}

STATIC mp_uint_t pyb_uart_write(mp_obj_t self_in, const void *buf_in, mp_uint_t size, int *errcode) {
    pyb_uart_obj_t *self = self_in;
    const byte *buf = buf_in;

    // check that size is a multiple of character width
    if (size & self->char_width) {
        *errcode = MP_EIO;
        return MP_STREAM_ERROR;
    }

    // wait to be able to write the first character. EAGAIN causes write to return None
    if (!uart_tx_wait(self, self->timeout)) {
        *errcode = MP_EAGAIN;
        return MP_STREAM_ERROR;
    }

    // write the data
    size_t num_tx = uart_tx_data(self, buf, size >> self->char_width, errcode);

    if (*errcode == 0 || *errcode == MP_ETIMEDOUT) {
        // return number of bytes written, even if there was a timeout
        return num_tx << self->char_width;
    } else {
        return MP_STREAM_ERROR;
    }
}

STATIC mp_uint_t pyb_uart_ioctl(mp_obj_t self_in, mp_uint_t request, mp_uint_t arg, int *errcode) {
    pyb_uart_obj_t *self = self_in;
    mp_uint_t ret;
    if (request == MP_STREAM_POLL) {
        mp_uint_t flags = arg;
        ret = 0;
        if ((flags & MP_STREAM_POLL_RD) && uart_rx_any(self)) {
            ret |= MP_STREAM_POLL_RD;
        }
        if ((flags & MP_STREAM_POLL_WR) && __HAL_UART_GET_FLAG(&self->uart, UART_FLAG_TXE)) {
            ret |= MP_STREAM_POLL_WR;
        }
    } else {
        *errcode = MP_EINVAL;
        ret = MP_STREAM_ERROR;
    }
    return ret;
}

STATIC const mp_stream_p_t uart_stream_p = {
    .read = pyb_uart_read,
    .write = pyb_uart_write,
    .ioctl = pyb_uart_ioctl,
    .is_text = false,
};

const mp_obj_type_t pyb_uart_type = {
    { &mp_type_type },
    .name = MP_QSTR_UART,
    .print = pyb_uart_print,
    .make_new = pyb_uart_make_new,
    .getiter = mp_identity_getiter,
    .iternext = mp_stream_unbuffered_iter,
    .protocol = &uart_stream_p,
    .locals_dict = (mp_obj_t)&pyb_uart_locals_dict,
};


text:
/*
Package orismologer translates non-OpenConfig telemetry sources (eg: SNMP OIDs) to OpenConfig paths.
*/
package orismologer

import (
	"fmt"
	"strings"

	"github.com/golang/glog"
	"github.com/google/orismologer/functions"
	"github.com/google/orismologer/octree"
	"github.com/google/orismologer/oparse"
	"github.com/google/orismologer/utils"

	pb "github.com/google/orismologer/proto_out/proto"
)

type transformationMap map[string]*pb.Transformation
type nocPathResolver func(*pb.NocPath, string) (interface{}, error)
type functionLibrary interface {
	Contains(funcName string) bool
	Call(funcName string, args ...interface{}) (interface{}, error)
}

// Orismologer translates non-OpenConfig telemetry sources (eg: SNMP OIDs) to OpenConfig paths.
type Orismologer struct {
	mappings        octree.OcTree
	transformations transformationMap
	vendorInfo      *pb.VendorOids
	nocPathResolver nocPathResolver
	functions       functionLibrary
}

/*
NewOrismologer builds an Orismologer instance from the text protos in the given files.
mappingsFile should contain a Mappings proto.
transformationFile should contain a Transformations proto.
vendorOidsFile should contain a VendorOids proto.
*/
func NewOrismologer(mappingsFile, transformationsFile, vendorOidsFile string) (*Orismologer, error) {
	mappings, err := utils.LoadMappings(mappingsFile)
	if err != nil {
		return nil, err
	}
	transformations, err := utils.LoadTransformations(transformationsFile)
	if err != nil {
		return nil, err
	}
	vendorOids, err := utils.LoadVendorOids(vendorOidsFile)
	if err != nil {
		return nil, err
	}
	return newOrismologer(mappings, transformations, vendorOids)
}

func newOrismologer(mappings *pb.Mappings, transformations *pb.Transformations, vendorInfo *pb.VendorOids) (*Orismologer, error) {
	t, err := octree.NewTree(mappings)
	if err != nil {
		return nil, err
	}
	transformationMap, err := makeTransformationMap(transformations)
	if err != nil {
		return nil, err
	}
	return &Orismologer{
		mappings:        t,
		transformations: transformationMap,
		vendorInfo:      vendorInfo,
		nocPathResolver: resolve,
		functions:       functions.NewLibrary(),
	}, nil
}

func makeTransformationMap(transformations *pb.Transformations) (transformationMap, error) {
	transformationMap := transformationMap{}
	for _, transformation := range transformations.GetTransformations() {
		name := transformation.GetBind()
		if _, ok := transformationMap[name]; ok {
			return nil, fmt.Errorf("more than one transformation bound to identifier %q", name)
		}
		transformationMap[name] = transformation
	}
	return transformationMap, nil
}

// PrintOcPaths pretty prints the tree of OpenConfig paths defined for this Orismologer instance.
func (o *Orismologer) PrintOcPaths(root string) error {
	return o.mappings.Print(root)
}

/*
Eval retrieves the current value of a given OpenConfig path for a target which does not natively
support OpenConfig.
The vendor name is used to identify dependencies for the target (eg: which OIDs it supports).
*/
// TODO: Support a dry run, to validate mappings and transformations protos.
func (o *Orismologer) Eval(openConfigPath, target, vendor string) (interface{}, error) {
	transformationName, err := o.mappings.GetTransformationIdentifier(openConfigPath)
	if err != nil {
		return nil, fmt.Errorf("failed to identify a transformation for path %q: %v", openConfigPath, err)
	}
	transformation, ok := o.transformations[transformationName]
	if !ok {
		return nil, fmt.Errorf("could not locate transformation %q for path %q", transformationName, openConfigPath)
	}
	glog.Infof("found transformation %q for path %q", transformationName, openConfigPath)
	return o.eval(transformation, target, vendor)
}

/*
eval parses and evaluates a Transformation proto's Expressions field, resolving any variables used
in expressions to their associated Transformations and recursively evaluating those until a final
value is obtained by resolving a NocPath. If a transformation defines multiple expressions then the
output of the first one that successfully evaluates is returned.

NocPaths are resolved using the function given to the Orismologer instance at instantiation.
*/
// TODO: Eval paths with keys, eg: thing/name[name=value]
// TODO: Safeguard against really long paths, and circular references.
func (o *Orismologer) eval(transformation *pb.Transformation, target string, vendor string) (interface{}, error) {
	transformationName := transformation.GetBind()
	glog.Infof("evaluating transformation %q for target %q of vendor %q", transformationName, target, vendor)
	nocPaths := o.getNocPaths(transformation)
	// Try to eval each expression defined for this transformation, taking the first that works.
	for _, expressionString := range transformation.GetExpressions() {
		glog.Infof("evaluating expression `%v`", expressionString)
		expression, variables, _, err := o.parseAndValidateExpression(expressionString)
		if err != nil {
			glog.Errorf("%v", err)
			continue
		}
		values, err := o.evalVariables(variables, nocPaths, target, vendor)
		if err != nil {
			if unresolvableNocPathError, ok := err.(unresolvableNocPathError); ok {
				glog.Info(unresolvableNocPathError.msg) // This is not an error we need to surface to the user.
			} else {
				glog.Errorf("%v", err)
			}
			glog.Infof("could not evaluate all variables for expression `%v`, continuing to next expression", expressionString)
			continue
		}

		// Evaluate the expression, passing in the values of the variables it uses.
		transformationResult, err := oparse.Eval(expression, values, o.functions.Call)
		if err != nil {
			return nil, err
		}
		return transformationResult, nil
	}
	return nil, fmt.Errorf("none of the expressions of transformation %q could be evaluated (see logs for details)", transformationName)
}

// getNocPaths returns a map of all the NocPaths defined in the given transformation.
func (o *Orismologer) getNocPaths(transformation *pb.Transformation) map[string]*pb.NocPath {
	transformationName := transformation.GetBind()
	paths := map[string]*pb.NocPath{}
	for _, nocPath := range transformation.GetNocPaths() {
		pathName := nocPath.GetBind()
		if len(pathName) == 0 {
			glog.Errorf("Transformation %q contains a NocPath without an identifier", transformationName)
		} else {
			glog.Infof("storing NocPath %q of transformation %q", pathName, transformationName)
			paths[pathName] = nocPath
		}
	}
	return paths
}

/*
Returns the expression parsed from the given string and any variables and function names used in it.
*/
func (o *Orismologer) parseAndValidateExpression(expressionString string) (*oparse.Expression, []string, []string, error) {
	expression, err := oparse.Parse(expressionString)
	if err != nil {
		glog.Errorf("could not parse expression `%v`", expressionString)
		return nil, nil, nil, err
	}
	variables, functionNames := expression.Identifiers()
	for _, functionName := range functionNames {
		if !o.functions.Contains(functionName) {
			return nil, nil, nil, fmt.Errorf("function %q is not defined", functionName)
		}
	}
	return expression, variables, functionNames, nil
}

/*
Evaluates each of the given variables, returning an error if one or more cannot be evaluated.
*/
func (o *Orismologer) evalVariables(variables []string, nocPaths map[string]*pb.NocPath, target string, vendor string) (map[string]interface{}, error) {
	values := oparse.Context{}
	for _, variable := range variables {
		glog.Infof("evaluating variable %q", variable)
		var value interface{}
		var err error
		nocPath := nocPaths[variable]
		transformation := o.transformations[variable]
		switch {
		case nocPath != nil:
			value, err = o.handleNocPath(nocPath, target, vendor)
			if err != nil {
				return nil, err
			}
		case transformation != nil:
			value, err = o.eval(transformation, target, vendor)
			if err != nil {
				return nil, fmt.Errorf("could not evaluate sub-transformation %q: %v", variable, err)
			}
		default:
			return nil, fmt.Errorf("NocPath or sub-transformation %q is undefined", variable)
		}
		glog.Infof("evaluated variable %q = %v", variable, value)
		values[variable] = value
	}
	return values, nil
}

// Gets a value for the given NocPath for the given target.
func (o *Orismologer) handleNocPath(nocPath *pb.NocPath, target string, vendor string) (interface{}, error) {
	pathName := nocPath.GetBind()
	if !o.canResolve(nocPath, vendor) {
		return nil, unresolvableNocPathError{
			fmt.Sprintf("ignoring NocPath %q as it cannot be resolved for vendor %q", pathName, vendor),
		}
	}
	value, err := o.nocPathResolver(nocPath, target)
	if err != nil {
		return nil, fmt.Errorf("failed to resolve NocPath %q for target %q (this NocPath should normally be resolvable for this target): %v", pathName, target, err)
	}
	return value, nil
}

type unresolvableNocPathError struct {
	msg string
}

func (f unresolvableNocPathError) Error() string {
	return f.msg
}

// canResolve returns true if the given target supports the given NocPath.
func (o *Orismologer) canResolve(nocPath *pb.NocPath, vendor string) bool {
	// NB: Currently assumes NocPaths are OIDs only.
	vendorRoot := o.vendorInfo.GetVendorRoot()
	for _, oid := range nocPath.GetOids() {
		if !strings.HasPrefix(oid, vendorRoot) {
			return true
		}
		vendorOid, ok := o.vendorInfo.GetVendors()[vendor]
		if !ok {
			return false
		}
		if strings.HasPrefix(oid, vendorRoot+"."+vendorOid) {
			return true
		}
	}
	return false
}

/*
resolve retrieves the value for a given NocPath from a given target.
This may involve sending an SNMP request, running a CLI command and parsing the output, etc.
*/
func resolve(nocPath *pb.NocPath, target string) (interface{}, error) {
	// TODO: Implement.
	glog.Infof("Requesting NocPath %q from target %q", nocPath.GetBind(), target)
	samples := nocPath.GetSamples()
	if len(samples) > 0 {
		return samples[0], nil
	}
	return "dummy", nil
}


text:
package collection_test

import (
	"testing"

	. "github.com/onsi/gomega"
	_ "google.golang.org/protobuf/types/known/emptypb"
	_ "google.golang.org/protobuf/types/known/structpb"

	"istio.io/istio/pkg/config"
	"istio.io/istio/pkg/config/schema/collection"
	"istio.io/istio/pkg/config/schema/resource"
)

var (
	emptyResource = resource.Builder{
		Kind:         "Empty",
		Plural:       "empties",
		ProtoPackage: "google.golang.org/protobuf/types/known/emptypb",
		Proto:        "google.protobuf.Empty",
	}.MustBuild()

	structResource = resource.Builder{
		Kind:         "Struct",
		Plural:       "structs",
		ProtoPackage: "google.golang.org/protobuf/types/known/structpb",
		Proto:        "google.protobuf.Struct",
	}.MustBuild()
)

func TestSchemas_Basic(t *testing.T) {
	g := NewWithT(t)

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()

	schemas := collection.SchemasFor(s)
	g.Expect(schemas.All()).To(HaveLen(1))
	g.Expect(schemas.All()[0]).To(Equal(s))
}

func TestSchemas_MustAdd(t *testing.T) {
	g := NewWithT(t)
	defer func() {
		r := recover()
		g.Expect(r).To(BeNil())
	}()
	b := collection.NewSchemasBuilder()

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()
	b.MustAdd(s)
}

func TestSchemas_MustRegister_Panic(t *testing.T) {
	g := NewWithT(t)
	defer func() {
		r := recover()
		g.Expect(r).NotTo(BeNil())
	}()
	b := collection.NewSchemasBuilder()

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()
	b.MustAdd(s)
	b.MustAdd(s)
}

func TestSchemas_Find(t *testing.T) {
	g := NewWithT(t)

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()

	schemas := collection.SchemasFor(s)

	s2, found := schemas.Find("foo")
	g.Expect(found).To(BeTrue())
	g.Expect(s2).To(Equal(s))

	_, found = schemas.Find("zoo")
	g.Expect(found).To(BeFalse())
}

func TestSchemas_MustFind(t *testing.T) {
	g := NewWithT(t)
	defer func() {
		r := recover()
		g.Expect(r).To(BeNil())
	}()

	b := collection.NewSchemasBuilder()

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()

	b.MustAdd(s)
	schemas := b.Build()

	s2 := schemas.MustFind("foo")
	g.Expect(s2).To(Equal(s))
}

func TestSchemas_MustFind_Panic(t *testing.T) {
	g := NewWithT(t)
	defer func() {
		r := recover()
		g.Expect(r).NotTo(BeNil())
	}()

	b := collection.NewSchemasBuilder()

	s := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()

	b.MustAdd(s)
	schemas := b.Build()

	_ = schemas.MustFind("zoo")
}

func TestSchema_FindByGroupVersionKind(t *testing.T) {
	g := NewWithT(t)

	s := collection.Builder{
		Name: "foo",
		Resource: resource.Builder{
			ProtoPackage: "github.com/gogo/protobuf/types",
			Proto:        "google.protobuf.Empty",
			Group:        "mygroup",
			Kind:         "Empty",
			Plural:       "empties",
			Version:      "v1",
		}.MustBuild(),
	}.MustBuild()

	schemas := collection.SchemasFor(s)

	s2, found := schemas.FindByGroupVersionKind(config.GroupVersionKind{
		Group:   "mygroup",
		Version: "v1",
		Kind:    "Empty",
	})
	g.Expect(found).To(BeTrue())
	g.Expect(s2).To(Equal(s))

	_, found = schemas.FindByGroupVersionKind(config.GroupVersionKind{
		Group:   "fake",
		Version: "v1",
		Kind:    "Empty",
	})
	g.Expect(found).To(BeFalse())
}

func TestSchema_MustFindByGroupVersionKind(t *testing.T) {
	g := NewWithT(t)
	b := collection.NewSchemasBuilder()

	s := collection.Builder{
		Name: "foo",
		Resource: resource.Builder{
			ProtoPackage: "github.com/gogo/protobuf/types",
			Proto:        "google.protobuf.Empty",
			Group:        "mygroup",
			Kind:         "Empty",
			Plural:       "empties",
			Version:      "v1",
		}.MustBuild(),
	}.MustBuild()

	b.MustAdd(s)
	schemas := b.Build()

	got := schemas.MustFindByGroupVersionKind(config.GroupVersionKind{
		Group:   "mygroup",
		Version: "v1",
		Kind:    "Empty",
	})
	g.Expect(s).To(Equal(got))
}

func TestSchema_MustFindByGroupVersionKind_Panic(t *testing.T) {
	g := NewWithT(t)

	defer func() {
		r := recover()
		g.Expect(r).NotTo(BeNil())
	}()

	schemas := collection.NewSchemasBuilder().Build()
	_ = schemas.MustFindByGroupVersionKind(config.GroupVersionKind{
		Group:   "mygroup",
		Version: "v1",
		Kind:    "Empty",
	})
}

func TestSchemas_CollectionNames(t *testing.T) {
	g := NewWithT(t)
	b := collection.NewSchemasBuilder()

	s1 := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()
	s2 := collection.Builder{
		Name:     "bar",
		Resource: emptyResource,
	}.MustBuild()
	b.MustAdd(s1)
	b.MustAdd(s2)

	names := b.Build().CollectionNames()
	expected := collection.Names{collection.NewName("bar"), collection.NewName("foo")}
	g.Expect(names).To(Equal(expected))
}

func TestSchemas_Kinds(t *testing.T) {
	g := NewWithT(t)

	s := collection.SchemasFor(
		collection.Builder{
			Name:     "foo",
			Resource: emptyResource,
		}.MustBuild(),
		collection.Builder{
			Name:     "bar",
			Resource: emptyResource,
		}.MustBuild(),
		collection.Builder{
			Name:     "baz",
			Resource: structResource,
		}.MustBuild())

	actual := s.Kinds()
	expected := []string{emptyResource.Kind(), structResource.Kind()}
	g.Expect(actual).To(Equal(expected))
}

func TestSchemas_Validate(t *testing.T) {
	cases := []struct {
		name        string
		schemas     []collection.Schema
		expectError bool
	}{
		{
			name: "valid",
			schemas: []collection.Schema{
				collection.Builder{
					Name: "foo",
					Resource: resource.Builder{
						Kind:   "Empty1",
						Plural: "Empty1s",
						Proto:  "google.protobuf.Empty",
					}.MustBuild(),
				}.MustBuild(),
				collection.Builder{
					Name: "bar",
					Resource: resource.Builder{
						Kind:   "Empty2",
						Plural: "Empty2s",
						Proto:  "google.protobuf.Empty",
					}.MustBuild(),
				}.MustBuild(),
			},
			expectError: false,
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			g := NewWithT(t)
			b := collection.NewSchemasBuilder()
			for _, s := range c.schemas {
				b.MustAdd(s)
			}
			err := b.Build().Validate()
			if c.expectError {
				g.Expect(err).ToNot(BeNil())
			} else {
				g.Expect(err).To(BeNil())
			}
		})
	}
}

func TestSchemas_Validate_Error(t *testing.T) {
	g := NewWithT(t)
	b := collection.NewSchemasBuilder()

	s1 := collection.Builder{
		Name: "foo",
		Resource: resource.Builder{
			Kind:         "Zoo",
			ProtoPackage: "github.com/gogo/protobuf/types",
			Proto:        "zoo",
		}.BuildNoValidate(),
	}.MustBuild()
	b.MustAdd(s1)

	err := b.Build().Validate()
	g.Expect(err).NotTo(BeNil())
}

func TestSchemas_ForEach(t *testing.T) {
	schemas := collection.SchemasFor(
		collection.Builder{
			Name:     "foo",
			Resource: emptyResource,
		}.MustBuild(),
		collection.Builder{
			Name:     "bar",
			Resource: emptyResource,
		}.MustBuild(),
	)

	cases := []struct {
		name     string
		expected []string
		actual   func() []string
	}{
		{
			name:     "all",
			expected: []string{"foo", "bar"},
			actual: func() []string {
				a := make([]string, 0)
				schemas.ForEach(func(s collection.Schema) bool {
					a = append(a, s.Name().String())
					return false
				})
				return a
			},
		},
		{
			name:     "exit early",
			expected: []string{"foo"},
			actual: func() []string {
				a := make([]string, 0)
				schemas.ForEach(func(s collection.Schema) bool {
					a = append(a, s.Name().String())
					return true
				})
				return a
			},
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			g := NewWithT(t)
			actual := c.actual()
			g.Expect(actual).To(Equal(c.expected))
		})
	}
}

func TestSchemas_Remove(t *testing.T) {
	g := NewWithT(t)

	foo := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()
	bar := collection.Builder{
		Name:     "bar",
		Resource: emptyResource,
	}.MustBuild()
	baz := collection.Builder{
		Name:     "baz",
		Resource: emptyResource,
	}.MustBuild()

	schemas := collection.SchemasFor(foo, bar)
	g.Expect(schemas.Remove(bar)).To(Equal(collection.SchemasFor(foo)))
	g.Expect(schemas.Remove(foo, bar, baz)).To(Equal(collection.SchemasFor()))
	g.Expect(schemas).To(Equal(collection.SchemasFor(foo, bar)))
}

func TestSchemas_Add(t *testing.T) {
	g := NewWithT(t)

	foo := collection.Builder{
		Name:     "foo",
		Resource: emptyResource,
	}.MustBuild()
	bar := collection.Builder{
		Name:     "bar",
		Resource: emptyResource,
	}.MustBuild()
	baz := collection.Builder{
		Name:     "baz",
		Resource: emptyResource,
	}.MustBuild()

	schemas := collection.SchemasFor(foo, bar)
	g.Expect(schemas.Add(baz)).To(Equal(collection.SchemasFor(foo, bar, baz)))
	g.Expect(schemas).To(Equal(collection.SchemasFor(foo, bar)))
}


text:
namespace content {

blink::WebConnectionType NetConnectionTypeToWebConnectionType(
    net::NetworkChangeNotifier::ConnectionType type);

}  // namespace content

#endif  // CONTENT_RENDERER_NET_INFO_HELPER_H_


text:
"""
General tests for all estimators in sklearn.
"""

# Authors: Andreas Mueller <amueller@ais.uni-bonn.de>
#          Gael Varoquaux gael.varoquaux@normalesup.org
# License: BSD Style.
import os
import warnings
import sys
import traceback
import inspect

import numpy as np
from scipy import sparse

from sklearn.utils.testing import assert_raises
from sklearn.utils.testing import assert_equal
from sklearn.utils.testing import assert_true
from sklearn.utils.testing import assert_false
from sklearn.utils.testing import assert_array_equal
from sklearn.utils.testing import assert_array_almost_equal
from sklearn.utils.testing import all_estimators
from sklearn.utils.testing import meta_estimators
from sklearn.utils.testing import set_random_state
from sklearn.utils.testing import assert_greater

import sklearn
from sklearn.base import (clone, ClassifierMixin, RegressorMixin,
                          TransformerMixin, ClusterMixin)
from sklearn.utils import shuffle
from sklearn.preprocessing import StandardScaler, Scaler
from sklearn.datasets import (load_iris, load_boston, make_blobs,
                              make_classification)
from sklearn.metrics import accuracy_score, adjusted_rand_score, f1_score

from sklearn.lda import LDA
from sklearn.svm.base import BaseLibSVM

# import "special" estimators
from sklearn.decomposition import SparseCoder
from sklearn.pls import _PLS, PLSCanonical, PLSRegression, CCA, PLSSVD
from sklearn.ensemble import RandomTreesEmbedding
from sklearn.feature_selection import SelectKBest
from sklearn.dummy import DummyClassifier, DummyRegressor
from sklearn.naive_bayes import MultinomialNB, BernoulliNB
from sklearn.covariance import EllipticEnvelope, EllipticEnvelop
from sklearn.feature_extraction import DictVectorizer, FeatureHasher
from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.kernel_approximation import AdditiveChi2Sampler
from sklearn.preprocessing import (LabelBinarizer, LabelEncoder, Binarizer,
                                   Normalizer, OneHotEncoder)
from sklearn.cluster import (WardAgglomeration, AffinityPropagation,
                             SpectralClustering)
from sklearn.isotonic import IsotonicRegression
from sklearn.random_projection import (GaussianRandomProjection,
                                       SparseRandomProjection)

from sklearn.cross_validation import train_test_split

dont_test = [SparseCoder, EllipticEnvelope, EllipticEnvelop, DictVectorizer,
             LabelBinarizer, LabelEncoder, TfidfTransformer,
             IsotonicRegression, OneHotEncoder, RandomTreesEmbedding,
             FeatureHasher, DummyClassifier, DummyRegressor]

def test_all_estimators():
    # Test that estimators are default-constructible, clonable
    # and have working repr.
    estimators = all_estimators(include_meta_estimators=True)
    clf = LDA()

    for name, E in estimators:
        # some can just not be sensibly default constructed
        if E in dont_test:
            continue
        # test default-constructibility
        # get rid of deprecation warnings
        with warnings.catch_warnings(record=True):
            if name in meta_estimators:
                e = E(clf)
            else:
                e = E()
            # test cloning
            clone(e)
            # test __repr__
            repr(e)
            # test that set_params returns self
            assert_true(isinstance(e.set_params(), E))

            # test if init does nothing but set parameters
            # this is important for grid_search etc.
            # We get the default parameters from init and then
            # compare these against the actual values of the attributes.

            # this comes from getattr. Gets rid of deprecation decorator.
            init = getattr(e.__init__, 'deprecated_original', e.__init__)
            try:
                args, varargs, kws, defaults = inspect.getargspec(init)
            except TypeError:
                # init is not a python function.
                # true for mixins
                continue
            params = e.get_params()
            if name in meta_estimators:
                # they need a non-default argument
                args = args[2:]
            else:
                args = args[1:]
            if args:
                # non-empty list
                assert_equal(len(args), len(defaults))
            else:
                continue
            for arg, default in zip(args, defaults):
                if arg not in params.keys():
                    # deprecated parameter, not in get_params
                    assert_true(default is None)
                    continue

                if isinstance(params[arg], np.ndarray):
                    assert_array_equal(params[arg], default)
                else:
                    assert_equal(params[arg], default)

def test_estimators_sparse_data():
    # All estimators should either deal with sparse data, or raise an
    # intelligible error message
    rng = np.random.RandomState(0)
    X = rng.rand(40, 10)
    X[X < .8] = 0
    X = sparse.csr_matrix(X)
    y = (4 * rng.rand(40)).astype(np.int)
    estimators = all_estimators()
    estimators = [(name, E) for name, E in estimators
                  if issubclass(E, (ClassifierMixin, RegressorMixin))]
    for name, Clf in estimators:
        if Clf in dont_test:
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            clf = Clf()
        # fit
        try:
            clf.fit(X, y)
        except TypeError, e:
            if not 'sparse' in repr(e):
                print ("Estimator %s doesn't seem to fail gracefully on "
                       "sparse data" % name)
                traceback.print_exc(file=sys.stdout)
                raise e
        except Exception, exc:
            print ("Estimator %s doesn't seem to fail gracefully on "
                   "sparse data" % name)
            traceback.print_exc(file=sys.stdout)
            raise exc

def test_transformers():
    # test if transformers do something sensible on training set
    # also test all shapes / shape errors
    transformers = all_estimators(type_filter='transformer')
    X, y = make_blobs(n_samples=30, centers=[[0, 0, 0], [1, 1, 1]],
                      random_state=0, n_features=2, cluster_std=0.1)
    n_samples, n_features = X.shape
    X = StandardScaler().fit_transform(X)
    X -= X.min()

    succeeded = True

    for name, Trans in transformers:
        trans = None

        if Trans in dont_test:
            continue
        # these don't actually fit the data:
        if Trans in [AdditiveChi2Sampler, Binarizer, Normalizer]:
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            trans = Trans()
        set_random_state(trans)
        if hasattr(trans, 'compute_importances'):
            trans.compute_importances = True

        if Trans is SelectKBest:
            # SelectKBest has a default of k=10
            # which is more feature than we have.
            trans.k = 1
        elif Trans in [GaussianRandomProjection,
                       SparseRandomProjection]:
            # Due to the jl lemma and very few samples, the number
            # of components of the random matrix projection will be greater
            # than the number of features.
            # So we impose a smaller number (avoid "auto" mode)
            trans.n_components = 1

        # fit

        if Trans in (_PLS, PLSCanonical, PLSRegression, CCA, PLSSVD):
            random_state = np.random.RandomState(seed=12345)
            y_ = np.vstack([y, 2 * y + random_state.randint(2, size=len(y))])
            y_ = y_.T
        else:
            y_ = y

        try:
            trans.fit(X, y_)
            X_pred = trans.fit_transform(X, y=y_)
            if isinstance(X_pred, tuple):
                for x_pred in X_pred:
                    assert_equal(x_pred.shape[0], n_samples)
            else:
                assert_equal(X_pred.shape[0], n_samples)
        except Exception as e:
            print trans
            print e
            print
            succeeded = False
            continue

        if hasattr(trans, 'transform'):
            if Trans in (_PLS, PLSCanonical, PLSRegression, CCA, PLSSVD):
                X_pred2 = trans.transform(X, y_)
            else:
                X_pred2 = trans.transform(X)
            if isinstance(X_pred, tuple) and isinstance(X_pred2, tuple):
                for x_pred, x_pred2 in zip(X_pred, X_pred2):
                    assert_array_almost_equal(
                        x_pred, x_pred2, 2,
                        "fit_transform not correct in %s" % Trans)
            else:
                assert_array_almost_equal(
                    X_pred, X_pred2, 2,
                    "fit_transform not correct in %s" % Trans)

            # raises error on malformed input for transform
            assert_raises(ValueError, trans.transform, X.T)
    assert_true(succeeded)

def test_transformers_sparse_data():
    # All estimators should either deal with sparse data, or raise an
    # intelligible error message
    rng = np.random.RandomState(0)
    X = rng.rand(40, 10)
    X[X < .8] = 0
    X = sparse.csr_matrix(X)
    y = (4 * rng.rand(40)).astype(np.int)
    estimators = all_estimators(type_filter='transformer')
    for name, Trans in estimators:
        if Trans in dont_test:
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            if Trans in [Scaler, StandardScaler]:
                trans = Trans(with_mean=False)
            elif Trans in [GaussianRandomProjection,
                           SparseRandomProjection]:
                # Due to the jl lemma and very few samples, the number
                # of components of the random matrix projection will be greater
                # than the number of features.
                # So we impose a smaller number (avoid "auto" mode)
                trans = Trans(n_components=np.int(X.shape[1] / 4))
            else:
                trans = Trans()
        # fit
        try:
            trans.fit(X, y)
        except TypeError, e:
            if not 'sparse' in repr(e):
                print ("Estimator %s doesn't seem to fail gracefully on "
                       "sparse data" % name)
                traceback.print_exc(file=sys.stdout)
                raise e
        except Exception, exc:
            print ("Estimator %s doesn't seem to fail gracefully on "
                   "sparse data" % name)
            traceback.print_exc(file=sys.stdout)
            raise exc

def test_estimators_nan_inf():
    # Test that all estimators check their input for NaN's and infs
    rnd = np.random.RandomState(0)
    X_train_finite = rnd.uniform(size=(10, 3))
    X_train_nan = rnd.uniform(size=(10, 3))
    X_train_nan[0, 0] = np.nan
    X_train_inf = rnd.uniform(size=(10, 3))
    X_train_inf[0, 0] = np.inf
    y = np.ones(10)
    y[:5] = 0
    estimators = all_estimators()
    estimators = [(name, E) for name, E in estimators
                  if (issubclass(E, ClassifierMixin) or
                      issubclass(E, RegressorMixin) or
                      issubclass(E, TransformerMixin) or
                      issubclass(E, ClusterMixin))]
    error_string_fit = "Estimator doesn't check for NaN and inf in fit."
    error_string_predict = ("Estimator doesn't check for NaN and inf in"
                            " predict.")
    error_string_transform = ("Estimator doesn't check for NaN and inf in"
                              " transform.")
    for X_train in [X_train_nan, X_train_inf]:
        for name, Est in estimators:
            if Est in dont_test:
                continue
            if Est in (_PLS, PLSCanonical, PLSRegression, CCA, PLSSVD):
                continue

            # catch deprecation warnings
            with warnings.catch_warnings(record=True):
                est = Est()
                if Est in [GaussianRandomProjection,
                           SparseRandomProjection]:
                    # Due to the jl lemma and very few samples, the number
                    # of components of the random matrix projection will be
                    # greater
                    # than the number of features.
                    # So we impose a smaller number (avoid "auto" mode)
                    est = Est(n_components=1)

                set_random_state(est, 1)
                # try to fit
                try:
                    if issubclass(Est, ClusterMixin):
                        est.fit(X_train)
                    else:
                        est.fit(X_train, y)
                except ValueError, e:
                    if not 'inf' in repr(e) and not 'NaN' in repr(e):
                        print(error_string_fit, Est, e)
                        traceback.print_exc(file=sys.stdout)
                        raise e
                except Exception, exc:
                        print(error_string_fit, Est, exc)
                        traceback.print_exc(file=sys.stdout)
                        raise exc
                else:
                    raise AssertionError(error_string_fit, Est)
                # actually fit
                if issubclass(Est, ClusterMixin):
                    # All estimators except clustering algorithm
                    # support fitting with (optional) y
                    est.fit(X_train_finite)
                else:
                    est.fit(X_train_finite, y)

                # predict
                if hasattr(est, "predict"):
                    try:
                        est.predict(X_train)
                    except ValueError, e:
                        if not 'inf' in repr(e) and not 'NaN' in repr(e):
                            print(error_string_predict, Est, e)
                            traceback.print_exc(file=sys.stdout)
                            raise e
                    except Exception, exc:
                        print(error_string_predict, Est, exc)
                        traceback.print_exc(file=sys.stdout)
                    else:
                        raise AssertionError(error_string_predict, Est)

                # transform
                if hasattr(est, "transform"):
                    try:
                        est.transform(X_train)
                    except ValueError, e:
                        if not 'inf' in repr(e) and not 'NaN' in repr(e):
                            print(error_string_transform, Est, e)
                            traceback.print_exc(file=sys.stdout)
                            raise e
                    except Exception, exc:
                        print(error_string_transform, Est, exc)
                        traceback.print_exc(file=sys.stdout)
                    else:
                        raise AssertionError(error_string_transform, Est)

def test_classifiers_one_label():
    # test classifiers trained on a single label always return this label
    # or raise an sensible error message
    rnd = np.random.RandomState(0)
    X_train = rnd.uniform(size=(10, 3))
    X_test = rnd.uniform(size=(10, 3))
    y = np.ones(10)
    classifiers = all_estimators(type_filter='classifier')
    error_string_fit = "Classifier can't train when only one class is present."
    error_string_predict = ("Classifier can't predict when only one class is "
                            "present.")
    for name, Clf in classifiers:
        if Clf in dont_test:
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            clf = Clf()
            # try to fit
            try:
                clf.fit(X_train, y)
            except ValueError, e:
                if not 'class' in repr(e):
                    print(error_string_fit, Clf, e)
                    traceback.print_exc(file=sys.stdout)
                    raise e
                else:
                    continue
            except Exception, exc:
                    print(error_string_fit, Clf, exc)
                    traceback.print_exc(file=sys.stdout)
                    raise exc
            # predict
            try:
                assert_array_equal(clf.predict(X_test), y)
            except Exception, exc:
                print(error_string_predict, Clf, exc)
                traceback.print_exc(file=sys.stdout)

def test_clustering():
    # test if clustering algorithms do something sensible
    # also test all shapes / shape errors
    clustering = all_estimators(type_filter='cluster')
    iris = load_iris()
    X, y = iris.data, iris.target
    X, y = shuffle(X, y, random_state=7)
    n_samples, n_features = X.shape
    X = StandardScaler().fit_transform(X)
    for name, Alg in clustering:
        if Alg is WardAgglomeration:
            # this is clustering on the features
            # let's not test that here.
            continue
        # catch deprecation and neighbors warnings
        with warnings.catch_warnings(record=True):
            alg = Alg()
            if hasattr(alg, "n_clusters"):
                alg.set_params(n_clusters=3)
            set_random_state(alg)
            if Alg is AffinityPropagation:
                alg.set_params(preference=-100)
            # fit
            alg.fit(X)

        assert_equal(alg.labels_.shape, (n_samples,))
        pred = alg.labels_
        assert_greater(adjusted_rand_score(pred, y), 0.4)
        # fit another time with ``fit_predict`` and compare results
        if Alg is SpectralClustering:
            # there is no way to make Spectral clustering deterministic :(
            continue
        set_random_state(alg)
        with warnings.catch_warnings(record=True):
            pred2 = alg.fit_predict(X)
        assert_array_equal(pred, pred2)

def test_classifiers_train():
    # test if classifiers do something sensible on training set
    # also test all shapes / shape errors
    classifiers = all_estimators(type_filter='classifier')
    X_m, y_m = make_blobs(random_state=0)
    X_m, y_m = shuffle(X_m, y_m, random_state=7)
    X_m = StandardScaler().fit_transform(X_m)
    # generate binary problem from multi-class one
    y_b = y_m[y_m != 2]
    X_b = X_m[y_m != 2]
    for (X, y) in [(X_m, y_m), (X_b, y_b)]:
        # do it once with binary, once with multiclass
        classes = np.unique(y)
        n_classes = len(classes)
        n_samples, n_features = X.shape
        for name, Clf in classifiers:
            if Clf in dont_test:
                continue
            if Clf in [MultinomialNB, BernoulliNB]:
                # TODO also test these!
                continue
            # catch deprecation warnings
            with warnings.catch_warnings(record=True):
                clf = Clf()
            # raises error on malformed input for fit
            assert_raises(ValueError, clf.fit, X, y[:-1])

            # fit
            clf.fit(X, y)
            assert_true(hasattr(clf, "classes_"))
            y_pred = clf.predict(X)
            assert_equal(y_pred.shape, (n_samples,))
            # training set performance
            assert_greater(accuracy_score(y, y_pred), 0.85)

            # raises error on malformed input for predict
            assert_raises(ValueError, clf.predict, X.T)
            if hasattr(clf, "decision_function"):
                try:
                    # decision_function agrees with predict:
                    decision = clf.decision_function(X)
                    if n_classes is 2:
                        assert_equal(decision.ravel().shape, (n_samples,))
                        dec_pred = (decision.ravel() > 0).astype(np.int)
                        assert_array_equal(dec_pred, y_pred)
                    if n_classes is 3 and not isinstance(clf, BaseLibSVM):
                        # 1on1 of LibSVM works differently
                        assert_equal(decision.shape, (n_samples, n_classes))
                        assert_array_equal(np.argmax(decision, axis=1), y_pred)

                    # raises error on malformed input
                    assert_raises(ValueError, clf.decision_function, X.T)
                    # raises error on malformed input for decision_function
                    assert_raises(ValueError, clf.decision_function, X.T)
                except NotImplementedError:
                    pass
            if hasattr(clf, "predict_proba"):
                try:
                    # predict_proba agrees with predict:
                    y_prob = clf.predict_proba(X)
                    assert_equal(y_prob.shape, (n_samples, n_classes))
                    assert_array_equal(np.argmax(y_prob, axis=1), y_pred)
                    # check that probas for all classes sum to one
                    assert_array_almost_equal(
                        np.sum(y_prob, axis=1), np.ones(n_samples))
                    # raises error on malformed input
                    assert_raises(ValueError, clf.predict_proba, X.T)
                    # raises error on malformed input for predict_proba
                    assert_raises(ValueError, clf.predict_proba, X.T)
                except NotImplementedError:
                    pass

def test_classifiers_classes():
    # test if classifiers can cope with non-consecutive classes
    classifiers = all_estimators(type_filter='classifier')
    X, y = make_blobs(random_state=12345)
    X, y = shuffle(X, y, random_state=7)
    X = StandardScaler().fit_transform(X)
    y = 2 * y + 1
    classes = np.unique(y)
    # TODO: make work with next line :)
    #y = y.astype(np.str)
    for name, Clf in classifiers:
        if Clf in dont_test:
            continue
        if Clf in [MultinomialNB, BernoulliNB]:
            # TODO also test these!
            continue

        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            clf = Clf()
        # fit
        clf.fit(X, y)
        y_pred = clf.predict(X)
        # training set performance
        assert_array_equal(np.unique(y), np.unique(y_pred))
        assert_greater(accuracy_score(y, y_pred), 0.78,
                       "accuracy of %s not greater than 0.78" % str(Clf))
        assert_array_equal(
            clf.classes_, classes,
            "Unexpected classes_ attribute for %r" % clf)

def test_regressors_int():
    # test if regressors can cope with integer labels (by converting them to
    # float)
    regressors = all_estimators(type_filter='regressor')
    boston = load_boston()
    X, y = boston.data, boston.target
    X, y = shuffle(X, y, random_state=0)
    X = StandardScaler().fit_transform(X)
    y = np.random.randint(2, size=X.shape[0])
    for name, Reg in regressors:
        if Reg in dont_test or Reg in (CCA,):
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            # separate estimators to control random seeds
            reg1 = Reg()
            reg2 = Reg()
        set_random_state(reg1)
        set_random_state(reg2)

        if Reg in (_PLS, PLSCanonical, PLSRegression):
            y_ = np.vstack([y, 2 * y + np.random.randint(2, size=len(y))])
            y_ = y_.T
        else:
            y_ = y

        # fit
        reg1.fit(X, y_)
        pred1 = reg1.predict(X)
        reg2.fit(X, y_.astype(np.float))
        pred2 = reg2.predict(X)
        assert_array_almost_equal(pred1, pred2, 2, name)

def test_regressors_train():
    regressors = all_estimators(type_filter='regressor')
    boston = load_boston()
    X, y = boston.data, boston.target
    X, y = shuffle(X, y, random_state=0)
    # TODO: test with intercept
    # TODO: test with multiple responses
    X = StandardScaler().fit_transform(X)
    y = StandardScaler().fit_transform(y)
    succeeded = True
    for name, Reg in regressors:
        if Reg in dont_test:
            continue
        # catch deprecation warnings
        with warnings.catch_warnings(record=True):
            reg = Reg()
        if not hasattr(reg, 'alphas') and hasattr(reg, 'alpha'):
            # linear regressors need to set alpha, but not generalized CV ones
            reg.alpha = 0.01

        # raises error on malformed input for fit
        assert_raises(ValueError, reg.fit, X, y[:-1])
        # fit
        try:
            if Reg in (_PLS, PLSCanonical, PLSRegression, CCA):
                y_ = np.vstack([y, 2 * y + np.random.randint(2, size=len(y))])
                y_ = y_.T
            else:
                y_ = y
            reg.fit(X, y_)
            reg.predict(X)

            if Reg not in (PLSCanonical, CCA):  # TODO: find out why
                assert_greater(reg.score(X, y_), 0.5)
        except Exception as e:
            print(reg)
            print e
            print
            succeeded = False

    assert_true(succeeded)

def test_configure():
    # Smoke test the 'configure' step of setup, this tests all the
    # 'configure' functions in the setup.pys in the scikit
    cwd = os.getcwd()
    setup_path = os.path.abspath(os.path.join(sklearn.__path__[0], '..'))
    setup_filename = os.path.join(setup_path, 'setup.py')
    if not os.path.exists(setup_filename):
        return
    try:
        os.chdir(setup_path)
        old_argv = sys.argv
        sys.argv = ['setup.py', 'config']
        with warnings.catch_warnings():
            # The configuration spits out warnings when not finding
            # Blas/Atlas development headers
            warnings.simplefilter('ignore',  UserWarning)
            execfile('setup.py', dict(__name__='__main__'))
    finally:
        sys.argv = old_argv
        os.chdir(cwd)

def test_class_weight_classifiers():
    # test that class_weight works and that the semantics are consistent
    classifiers = all_estimators(type_filter='classifier')

    with warnings.catch_warnings(record=True):
        classifiers = [c for c in classifiers
                       if 'class_weight' in c[1]().get_params().keys()]

    for n_centers in [2, 3]:
        # create a very noisy dataset
        X, y = make_blobs(centers=n_centers, random_state=0, cluster_std=20)
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=.5,
                                                            random_state=0)
        for name, Clf in classifiers:
            if name == "NuSVC":
                # the sparse version has a parameter that doesn't do anything
                continue
            if name.endswith("NB"):
                # NaiveBayes classifiers have a somewhat different interface.
                # FIXME SOON!
                continue
            if n_centers == 2:
                class_weight = {0: 1000, 1: 0.0001}
            else:
                class_weight = {0: 1000, 1: 0.0001, 2: 0.0001}

            with warnings.catch_warnings(record=True):
                clf = Clf(class_weight=class_weight)
            if hasattr(clf, "n_iter"):
                clf.set_params(n_iter=100)

            set_random_state(clf)
            clf.fit(X_train, y_train)
            y_pred = clf.predict(X_test)
            assert_greater(np.mean(y_pred == 0), 0.9)

def test_class_weight_auto_classifies():
    # test that class_weight="auto" improves f1-score
    classifiers = all_estimators(type_filter='classifier')

    with warnings.catch_warnings(record=True):
        classifiers = [c for c in classifiers
                       if 'class_weight' in c[1]().get_params().keys()]

    for n_classes, weights in zip([2, 3], [[.8, .2], [.8, .1, .1]]):
        # create unbalanced dataset
        X, y = make_classification(n_classes=n_classes, n_samples=200,
                                   n_features=10, weights=weights,
                                   random_state=0, n_informative=n_classes)
        X = StandardScaler().fit_transform(X)
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=.5,
                                                            random_state=0)
        for name, Clf in classifiers:
            if name == "NuSVC":
                # the sparse version has a parameter that doesn't do anything
                continue

            if name.startswith("RidgeClassifier"):
                # RidgeClassifier behaves unexpected
                # FIXME!
                continue

            if name.endswith("NB"):
                # NaiveBayes classifiers have a somewhat different interface.
                # FIXME SOON!
                continue

            with warnings.catch_warnings(record=True):
                clf = Clf()
            if hasattr(clf, "n_iter"):
                clf.set_params(n_iter=100)

            set_random_state(clf)
            clf.fit(X_train, y_train)
            y_pred = clf.predict(X_test)

            clf.set_params(class_weight='auto')
            clf.fit(X_train, y_train)
            y_pred_auto = clf.predict(X_test)
            assert_greater(f1_score(y_test, y_pred_auto),
                           f1_score(y_test, y_pred))

def test_estimators_overwrite_params():
    # test whether any classifier overwrites his init parameters during fit
    for est_type in ["classifier", "regressor", "transformer"]:
        estimators = all_estimators(type_filter="classifier")
        X, y = make_blobs(random_state=0, n_samples=6)
        # some want non-negative input
        X -= X.min()
        for name, Est in estimators:
            with warnings.catch_warnings(record=True):
                # catch deprecation warnings
                est = Est()
            params = est.get_params()
            est.fit(X, y)
            new_params = est.get_params()
            for k, v in params.items():
                assert_false(np.any(new_params[k] != v),
                             "Estimator %s changes its parameter %s"
                             "from %s to %s during fit."
                             % (name, k, v, new_params[k]))

def test_cluster_overwrite_params():
    # test whether any classifier overwrites his init parameters during fit
    clusterers = all_estimators(type_filter="cluster")
    X, y = make_blobs(random_state=0, n_samples=9)
    # some want non-negative input
    X
    for name, Clt in clusterers:
        with warnings.catch_warnings(record=True):
            # catch deprecation warnings
            clt = Clt()
        params = clt.get_params()
        clt.fit(X)
        new_params = clt.get_params()
        for k, v in params.items():
            assert_false(np.any(new_params[k] != v),
                         "Estimator %s changes its parameter %s"
                         "from %s to %s during fit."
                         % (name, k, v, new_params[k]))


text:
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.Text;
using System.Threading.Tasks;

namespace Webapp.Models
{
    [JsonObject]
    public class ApiModel<TValue> where TValue : class
    {
        public ApiResult Result { get; set; }

        [JsonProperty(DefaultValueHandling = DefaultValueHandling.Ignore)]
        public TValue Value { get; private set; }

        public ApiModel(TValue val)
        {
            this.Value = val;
            this.Result = ApiResult.SuccessResult;
        }
        public ApiModel(TValue val, ApiResult result)
        {
            this.Result = result;
            this.Value = val;
        }
        public ApiModel(Exception e, bool includeExceptions)
        {
            this.Result = ApiResult.FromException(e, includeExceptions);
        }

        // Helpers
        public static ApiModel<TValue> AsError(string errorField, string errorMessage)
        {
            return new ApiModel<TValue>(null, ApiResult.AsError(errorField, errorMessage));
        }
        public static ApiModel<TValue> AsError(string errorMessage, int errorCode = 0)
        {
            return new ApiModel<TValue>(null, ApiResult.AsError(errorMessage, errorCode));
        }
        public static ApiModel<TValue> AsError(string errorMessage, TValue val)
        {
            return new ApiModel<TValue>(val, ApiResult.AsError(errorMessage));
        }
        public static ApiModel<TValue> AsSuccess(TValue val, string message = null)
        {
            return new ApiModel<TValue>(val, ApiResult.AsSuccess(message));
        }

        public static ApiModel<TValue> FromException(Exception exception, bool includeExceptions = false)
        {
            return new ApiModel<TValue>(null, ApiResult.FromException(exception, includeExceptions));
        }
    }

    public partial class ApiModel
    {
        public static ApiModel<TValue> AsSuccess<TValue>(TValue val, string message = null) where TValue : class
        {
            return new ApiModel<TValue>(val, ApiResult.AsSuccess(message));
        }
        public static ApiModel<TValue> AsError<TValue>(TValue val, string message = null) where TValue : class
        {
            return new ApiModel<TValue>(val, ApiResult.AsError(message));
        }
        public static ApiModel<TValue> FromException<TValue>(TValue val, Exception e, bool includeExceptions = false) where TValue : class
        {
            return new ApiModel<TValue>(val, ApiResult.FromException(e, includeExceptions));
        }
    }
}