text:
# -*- coding: utf-8 -*-
"""
Tests the speed of image updates for an ImageItem and RawImageWidget.
The speed will generally depend on the type of data being shown, whether
it is being scaled and/or converted by lookup table, and whether OpenGL
is used by the view widget
"""

## Add path to library (just for examples; you do not need this)
import initExample

import argparse
import sys

import numpy as np

import pyqtgraph as pg
import pyqtgraph.ptime as ptime
from pyqtgraph.Qt import QtGui, QtCore, QT_LIB

pg.setConfigOption('imageAxisOrder', 'row-major')

import importlib
ui_template = importlib.import_module(f'VideoTemplate_{QT_LIB.lower()}')

try:
    import cupy as cp
    pg.setConfigOption("useCupy", True)
    _has_cupy = True
except ImportError:
    cp = None
    _has_cupy = False

try:
    import numba
    _has_numba = True
except ImportError:
    numba = None
    _has_numba = False

try:
    from pyqtgraph.widgets.RawImageWidget import RawImageGLWidget
except ImportError:
    RawImageGLWidget = None

parser = argparse.ArgumentParser(description="Benchmark for testing video performance")
parser.add_argument('--cuda', default=False, action='store_true', help="Use CUDA to process on the GPU", dest="cuda")
parser.add_argument('--dtype', default='uint8', choices=['uint8', 'uint16', 'float'], help="Image dtype (uint8, uint16, or float)")
parser.add_argument('--frames', default=3, type=int, help="Number of image frames to generate (default=3)")
parser.add_argument('--image-mode', default='mono', choices=['mono', 'rgb'], help="Image data mode (mono or rgb)", dest='image_mode')
parser.add_argument('--levels', default=None, type=lambda s: tuple([float(x) for x in s.split(',')]), help="min,max levels to scale monochromatic image dynamic range, or rmin,rmax,gmin,gmax,bmin,bmax to scale rgb")
parser.add_argument('--lut', default=False, action='store_true', help="Use color lookup table")
parser.add_argument('--lut-alpha', default=False, action='store_true', help="Use alpha color lookup table", dest='lut_alpha')
parser.add_argument('--size', default='512x512', type=lambda s: tuple([int(x) for x in s.split('x')]), help="WxH image dimensions default='512x512'")
args = parser.parse_args(sys.argv[1:])

if RawImageGLWidget is not None:
    # don't limit frame rate to vsync
    sfmt = QtGui.QSurfaceFormat()
    sfmt.setSwapInterval(0)
    QtGui.QSurfaceFormat.setDefaultFormat(sfmt)

app = pg.mkQApp("Video Speed Test Example")

win = QtGui.QMainWindow()
win.setWindowTitle('pyqtgraph example: VideoSpeedTest')
ui = ui_template.Ui_MainWindow()
ui.setupUi(win)
win.show()

if RawImageGLWidget is None:
    ui.rawGLRadio.setEnabled(False)
    ui.rawGLRadio.setText(ui.rawGLRadio.text() + " (OpenGL not available)")
else:
    ui.rawGLImg = RawImageGLWidget()
    ui.stack.addWidget(ui.rawGLImg)

# read in CLI args
ui.cudaCheck.setChecked(args.cuda and _has_cupy)
ui.cudaCheck.setEnabled(_has_cupy)
ui.numbaCheck.setChecked(_has_numba and pg.getConfigOption("useNumba"))
ui.numbaCheck.setEnabled(_has_numba)
ui.framesSpin.setValue(args.frames)
ui.widthSpin.setValue(args.size[0])
ui.heightSpin.setValue(args.size[1])
ui.dtypeCombo.setCurrentText(args.dtype)
ui.rgbCheck.setChecked(args.image_mode=='rgb')
ui.maxSpin1.setOpts(value=255, step=1)
ui.minSpin1.setOpts(value=0, step=1)
levelSpins = [ui.minSpin1, ui.maxSpin1, ui.minSpin2, ui.maxSpin2, ui.minSpin3, ui.maxSpin3]
if args.cuda and _has_cupy:
    xp = cp
else:
    xp = np
if args.levels is None:
    ui.scaleCheck.setChecked(False)
    ui.rgbLevelsCheck.setChecked(False)
else:
    ui.scaleCheck.setChecked(True)
    if len(args.levels) == 2:
        ui.rgbLevelsCheck.setChecked(False)
        ui.minSpin1.setValue(args.levels[0])
        ui.maxSpin1.setValue(args.levels[1])
    elif len(args.levels) == 6:
        ui.rgbLevelsCheck.setChecked(True)
        for spin,val in zip(levelSpins, args.levels):
            spin.setValue(val)
    else:
        raise ValueError("levels argument must be 2 or 6 comma-separated values (got %r)" % (args.levels,))
ui.lutCheck.setChecked(args.lut)
ui.alphaCheck.setChecked(args.lut_alpha)

#ui.graphicsView.useOpenGL()  ## buggy, but you can try it if you need extra speed.

vb = pg.ViewBox()
ui.graphicsView.setCentralItem(vb)
vb.setAspectLocked()
img = pg.ImageItem()
vb.addItem(img)

LUT = None
def updateLUT():
    global LUT, ui
    dtype = ui.dtypeCombo.currentText()
    if dtype == 'uint8':
        n = 256
    else:
        n = 4096
    LUT = ui.gradient.getLookupTable(n, alpha=ui.alphaCheck.isChecked())
    if _has_cupy and xp == cp:
        LUT = cp.asarray(LUT)
ui.gradient.sigGradientChanged.connect(updateLUT)
updateLUT()

ui.alphaCheck.toggled.connect(updateLUT)

def updateScale():
    global ui, levelSpins
    if ui.rgbLevelsCheck.isChecked():
        for s in levelSpins[2:]:
            s.setEnabled(True)
    else:
        for s in levelSpins[2:]:
            s.setEnabled(False)

updateScale()

ui.rgbLevelsCheck.toggled.connect(updateScale)

cache = {}
def mkData():
    with pg.BusyCursor():
        global data, cache, ui, xp
        frames = ui.framesSpin.value()
        width = ui.widthSpin.value()
        height = ui.heightSpin.value()
        cacheKey = (ui.dtypeCombo.currentText(), ui.rgbCheck.isChecked(), frames, width, height)
        if cacheKey not in cache:
            if cacheKey[0] == 'uint8':
                dt = xp.uint8
                loc = 128
                scale = 64
                mx = 255
            elif cacheKey[0] == 'uint16':
                dt = xp.uint16
                loc = 4096
                scale = 1024
                mx = 2**16 - 1
            elif cacheKey[0] == 'float':
                dt = xp.float32
                loc = 1.0
                scale = 0.1
                mx = 1.0
            else:
                raise ValueError(f"unable to handle dtype: {cacheKey[0]}")

            chan_shape = (height, width)
            if ui.rgbCheck.isChecked():
                frame_shape = chan_shape + (3,)
            else:
                frame_shape = chan_shape
            data = xp.empty((frames,) + frame_shape, dtype=dt)
            view = data.reshape((-1,) + chan_shape)
            for idx in range(view.shape[0]):
                subdata = xp.random.normal(loc=loc, scale=scale, size=chan_shape)
                # note: gaussian filtering has been removed as it slows down array
                #       creation greatly.
                if cacheKey[0] != 'float':
                    xp.clip(subdata, 0, mx, out=subdata)
                view[idx] = subdata

            data[:, 10:50, 10] = mx
            data[:, 48, 9:12] = mx
            data[:, 47, 8:13] = mx
            cache = {cacheKey: data} # clear to save memory (but keep one to prevent unnecessary regeneration)

        data = cache[cacheKey]
        updateLUT()
        updateSize()

def updateSize():
    global ui, vb
    frames = ui.framesSpin.value()
    width = ui.widthSpin.value()
    height = ui.heightSpin.value()
    dtype = xp.dtype(str(ui.dtypeCombo.currentText()))
    rgb = 3 if ui.rgbCheck.isChecked() else 1
    ui.sizeLabel.setText('%d MB' % (frames * width * height * rgb * dtype.itemsize / 1e6))
    vb.setRange(QtCore.QRectF(0, 0, width, height))

def noticeCudaCheck():
    global xp, cache
    cache = {}
    if ui.cudaCheck.isChecked():
        if _has_cupy:
            xp = cp
        else:
            xp = np
            ui.cudaCheck.setChecked(False)
    else:
        xp = np
    mkData()

def noticeNumbaCheck():
    pg.setConfigOption('useNumba', _has_numba and ui.numbaCheck.isChecked())

mkData()

ui.dtypeCombo.currentIndexChanged.connect(mkData)
ui.rgbCheck.toggled.connect(mkData)
ui.widthSpin.editingFinished.connect(mkData)
ui.heightSpin.editingFinished.connect(mkData)
ui.framesSpin.editingFinished.connect(mkData)

ui.widthSpin.valueChanged.connect(updateSize)
ui.heightSpin.valueChanged.connect(updateSize)
ui.framesSpin.valueChanged.connect(updateSize)
ui.cudaCheck.toggled.connect(noticeCudaCheck)
ui.numbaCheck.toggled.connect(noticeNumbaCheck)

ptr = 0
lastTime = ptime.time()
fps = None
def update():
    global ui, ptr, lastTime, fps, LUT, img
    if ui.lutCheck.isChecked():
        useLut = LUT
    else:
        useLut = None

    downsample = ui.downsampleCheck.isChecked()

    if ui.scaleCheck.isChecked():
        if ui.rgbLevelsCheck.isChecked():
            useScale = [
                [ui.minSpin1.value(), ui.maxSpin1.value()],
                [ui.minSpin2.value(), ui.maxSpin2.value()],
                [ui.minSpin3.value(), ui.maxSpin3.value()]]
        else:
            useScale = [ui.minSpin1.value(), ui.maxSpin1.value()]
    else:
        useScale = None

    if ui.rawRadio.isChecked():
        ui.rawImg.setImage(data[ptr%data.shape[0]], lut=useLut, levels=useScale)
        ui.stack.setCurrentIndex(1)
    elif ui.rawGLRadio.isChecked():
        ui.rawGLImg.setImage(data[ptr%data.shape[0]], lut=useLut, levels=useScale)
        ui.stack.setCurrentIndex(2)
    else:
        img.setImage(data[ptr%data.shape[0]], autoLevels=False, levels=useScale, lut=useLut, autoDownsample=downsample)
        ui.stack.setCurrentIndex(0)
        #img.setImage(data[ptr%data.shape[0]], autoRange=False)

    ptr += 1
    now = ptime.time()
    dt = now - lastTime
    lastTime = now
    if fps is None:
        fps = 1.0/dt
    else:
        s = np.clip(dt*3., 0, 1)
        fps = fps * (1-s) + (1.0/dt) * s
    ui.fpsLabel.setText('%0.2f fps' % fps)
    app.processEvents()  ## force complete redraw for every plot
timer = QtCore.QTimer()
timer.timeout.connect(update)
timer.start(0)

if __name__ == '__main__':
    pg.exec()


text:
from django import forms

from sme_uniforme_apps.proponentes.models import Anexo

class AnexoForm(forms.ModelForm):

    def __init__(self, *args, **kwargs):
        super(AnexoForm, self).__init__(*args, **kwargs)

        self.fields['tipo_documento'].required = True

    class Meta:
        model = Anexo
        fields = '__all__'

domain: python


text:
# Generated by Django 4.0.3 on 2022-04-06 17:40

from django.db import migrations

class Migration(migrations.Migration):

    dependencies = [
        ('records', '0006_alter_records_phasesday'),
    ]

    operations = [
        migrations.RenameField(
            model_name='records',
            old_name='date',
            new_name='created_date',
        ),
    ]


text:
# Licensed under the MIT license
# http://opensource.org/licenses/mit-license.php

# Copyright 2006, Frank Scholz <coherence@beebits.net>

class RenderingControlClient:

    def __init__(self, service):
        self.service = service
        self.namespace = service.get_type()
        self.url = service.get_control_url()
        self.service.subscribe()
        self.service.client = self
        #print "RenderingControlClient __init__", self.url

    #def __del__(self):
    #    #print "RenderingControlClient deleted"
    #    pass

    def remove(self):
        self.service.remove()
        self.service = None
        self.namespace = None
        self.url = None
        del self

    def subscribe_for_variable(self, var_name, callback,signal=False):
        self.service.subscribe_for_variable(var_name, instance=0, callback=callback,signal=signal)

    def list_presets(self, instance_id=0):
        action = self.service.get_action('ListPresets')
        return action.call(InstanceID=instance_id)

    def select_presets(self, instance_id=0, preset_name=''):
        action = self.service.get_action('SelectPresets')
        return action.call( InstanceID=instance_id,
                            PresetName=preset_name)

    def get_mute(self, instance_id=0, channel='Master'):
        action = self.service.get_action('GetMute')
        return action.call( InstanceID=instance_id,
                            Channel=channel)

    def set_mute(self, instance_id=0, channel='Master', desired_mute=0):
        action = self.service.get_action('SetMute')
        return action.call( InstanceID=instance_id,
                            Channel=channel,
                            DesiredMute=desired_mute)

    def get_volume(self, instance_id=0, channel='Master'):
        action = self.service.get_action('GetVolume')
        return action.call( InstanceID=instance_id,
                            Channel=channel)

    def set_volume(self, instance_id=0, channel='Master', desired_volume=0):
        action = self.service.get_action('SetVolume')
        return action.call( InstanceID=instance_id,
                            Channel=channel,
                            DesiredVolume=desired_volume)

    def get_volume_db(self, instance_id=0, channel='Master'):
        action = self.service.get_action('GetVolumeDB')
        return action.call( InstanceID=instance_id,
                            Channel=channel)

    def set_volume_db(self, instance_id=0, channel='Master', desired_volume=0):
        action = self.service.get_action('SetVolumeDB')
        return action.call( InstanceID=instance_id,
                            Channel=channel,
                            DesiredVolume=desired_volume)

    def get_volume_db_range(self, instance_id=0, channel='Master'):
        action = self.service.get_action('GetVolumeDBRange')
        return action.call( InstanceID=instance_id,
                            Channel=channel)

    def get_loudness(self, instance_id=0, channel='Master'):
        action = self.service.get_action('GetLoudness')
        return action.call( InstanceID=instance_id,
                            Channel=channel)

    def set_loudness(self, instance_id=0, channel='Master', desired_loudness=0):
        action = self.service.get_action('SetLoudness')
        return action.call( InstanceID=instance_id,
                            Channel=channel,
                            DesiredLoudness=desired_loudness)


text:
#!/usr/bin/env python2.7
# William Lam
# wwww.virtuallyghetto.com
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import pyVim.connect as connect
import getpass
import requests
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
# Snippet borrowed from Michael Rice
# https://gist.github.com/michaelrice/a6794a017e349fc65d01
requests.packages.urllib3.disable_warnings()

try:
    _create_unverified_https_context = ssl._create_unverified_context
except AttributeError:
    # Legacy Python that doesn't verify HTTPS certificates by default
    pass
else:
    # Handle target environment that doesn't support HTTPS verification
    ssl._create_default_https_context = _create_unverified_https_context

# Demonstrates configuring the Message of the Day (MOTD) on vCenter Server

# Example output:
# > logged in to vcsa
# > Setting vCenter Server MOTD to "Hello from virtuallyGhetto"
# > logout

parser = argparse.ArgumentParser()
parser.add_argument('-s', '--host',
                    required=True,
                    action='store',
                    help='Remote host to connect to')

parser.add_argument('-u', '--user',
                    required=True,
                    action='store',
                    help='User name to use when connecting to host')

parser.add_argument('-p', '--password',
                    required=False,
                    action='store',
                    help='Password to use when connecting to host')

parser.add_argument('-o', '--port',
                    required=False,
                    action='store',
                    help="port to use, default 443", default=443)

parser.add_argument('-m', '--message',
                    required=True,
                    action='store',
                    help='Message to be used for VC MOTD')

args = parser.parse_args()
if args.password:
    password = args.password
else:
    password = getpass.getpass(
        prompt='Enter password for host %s and user %s: ' %
               (args.host, args.user))

si = connect.SmartConnect(host=args.host,
                          user=args.user,
                          pwd=password,
                          port=int(args.port))

print "logged in to %s" % args.host

print "Setting vCenter Server MOTD to \"%s\"" % args.message
si.content.sessionManager.UpdateServiceMessage(message=args.message)

print "logout"
si.content.sessionManager.Logout()


text:
# -*- coding: utf-8 -*-
from __future__ import print_function, division, absolute_import, unicode_literals
from parser_base import RegexParser
import model

class RegexSemantics(object):
    def __init__(self):
        super(RegexSemantics, self).__init__()
        self._count = 0

    def START(self, ast):
        return model.Regex(ast)

    def CHOICE(self, ast):
        return model.Choice(ast.opts)

    def SEQUENCE(self, ast):
        if not ast.terms:
            return model.Empty()
        elif len(ast.terms) < 2:
            return ast.terms[0]
        else:
            return model.Sequence(ast.terms)

    def CLOSURE(self, ast):
        return model.Closure(ast)

    def SUBEXP(self, ast):
        return ast

    def LITERAL(self, ast):
        return model.Literal(ast)

def translate(regex, trace=False):
    parser = RegexParser(trace=trace, semantics=RegexSemantics())
    model = parser.parse(regex, 'START')
    model.set_rule_numbers()
    return model.render()


text:
"""
Client for Yandex.Disk.
"""
__version__ = '0.0.1'


text:
#!/usr/bin/env python
# -*- coding: utf-8 -*-

# refer to `https://bitbucket.org/akorn/wheezy.captcha`

import random
import string
import os.path
from io import BytesIO

from PIL import Image
from PIL import ImageFilter
from PIL.ImageDraw import Draw
from PIL.ImageFont import truetype

class Bezier:
    def __init__(self):
        self.tsequence = tuple([t / 20.0 for t in range(21)])
        self.beziers = {}

    def pascal_row(self, n):
        """ Returns n-th row of Pascal's triangle
        """
        result = [1]
        x, numerator = 1, n
        for denominator in range(1, n // 2 + 1):
            x *= numerator
            x /= denominator
            result.append(x)
            numerator -= 1
        if n & 1 == 0:
            result.extend(reversed(result[:-1]))
        else:
            result.extend(reversed(result))
        return result

    def make_bezier(self, n):
        """ Bezier curves:
            http://en.wikipedia.org/wiki/B%C3%A9zier_curve#Generalization
        """
        try:
            return self.beziers[n]
        except KeyError:
            combinations = self.pascal_row(n - 1)
            result = []
            for t in self.tsequence:
                tpowers = (t ** i for i in range(n))
                upowers = ((1 - t) ** i for i in range(n - 1, -1, -1))
                coefs = [c * a * b for c, a, b in zip(combinations,
                                                      tpowers, upowers)]
                result.append(coefs)
            self.beziers[n] = result
            return result

class Captcha(object):
    def __init__(self):
        self._bezier = Bezier()
        self._dir = os.path.dirname(__file__)
        # self._captcha_path = os.path.join(self._dir, '..', 'static', 'captcha')

    @staticmethod
    def instance():
        if not hasattr(Captcha, "_instance"):
            Captcha._instance = Captcha()
        return Captcha._instance

    def initialize(self, width=200, height=75, color=None, text=None, fonts=None):
        # self.image = Image.new('RGB', (width, height), (255, 255, 255))

        # 4位验证随机码
        self._text = text if text else random.sample(string.ascii_uppercase + string.ascii_uppercase + '3456789', 4)
        # print(self._text)

        self.fonts = fonts if fonts else \
            [os.path.join(self._dir, 'fonts', font) for font in ['Arial.ttf', 'Georgia.ttf', 'actionj.ttf']]
        self.width = width
        self.height = height
        self._color = color if color else self.random_color(0, 200, random.randint(220, 255))

    @staticmethod
    def random_color(start, end, opacity=None):
        red = random.randint(start, end)
        green = random.randint(start, end)
        blue = random.randint(start, end)
        if opacity is None:
            return red, green, blue
        return red, green, blue, opacity

    # draw image

    def background(self, image):
        Draw(image).rectangle([(0, 0), image.size], fill=self.random_color(238, 255))
        return image

    @staticmethod
    def smooth(image):
        return image.filter(ImageFilter.SMOOTH)

    def curve(self, image, width=4, number=6, color=None):
        dx, height = image.size
        dx /= number
        path = [(dx * i, random.randint(0, height))
                for i in range(1, number)]
        bcoefs = self._bezier.make_bezier(number - 1)
        points = []
        for coefs in bcoefs:
            points.append(tuple(sum([coef * p for coef, p in zip(coefs, ps)])
                                for ps in zip(*path)))
        Draw(image).line(points, fill=color if color else self._color, width=width)
        return image

    def noise(self, image, number=50, level=2, color=None):
        width, height = image.size
        dx = width / 10
        width -= dx
        dy = height / 10
        height -= dy
        draw = Draw(image)
        for i in range(number):
            x = int(random.uniform(dx, width))
            y = int(random.uniform(dy, height))
            draw.line(((x, y), (x + level, y)), fill=color if color else self._color, width=level)
        return image

    def text(self, image, fonts, font_sizes=None, drawings=None, squeeze_factor=0.75, color=None):
        color = color if color else self._color
        fonts = tuple([truetype(name, size)
                       for name in fonts
                       for size in font_sizes or (65, 70, 75)])
        draw = Draw(image)
        char_images = []
        for c in self._text:
            font = random.choice(fonts)
            c_width, c_height = draw.textsize(c, font=font)
            char_image = Image.new('RGB', (c_width, c_height), (0, 0, 0))
            char_draw = Draw(char_image)
            char_draw.text((0, 0), c, font=font, fill=color)
            char_image = char_image.crop(char_image.getbbox())
            for drawing in drawings:
                d = getattr(self, drawing)
                char_image = d(char_image)
            char_images.append(char_image)
        width, height = image.size
        offset = int((width - sum(int(i.size[0] * squeeze_factor)
                                  for i in char_images[:-1]) -
                      char_images[-1].size[0]) / 2)
        for char_image in char_images:
            c_width, c_height = char_image.size
            mask = char_image.convert('L').point(lambda i: i * 1.97)
            image.paste(char_image,
                        (offset, int((height - c_height) / 2)),
                        mask)
            offset += int(c_width * squeeze_factor)
        return image

    # draw text
    @staticmethod
    def warp(image, dx_factor=0.27, dy_factor=0.21):
        width, height = image.size
        dx = width * dx_factor
        dy = height * dy_factor
        x1 = int(random.uniform(-dx, dx))
        y1 = int(random.uniform(-dy, dy))
        x2 = int(random.uniform(-dx, dx))
        y2 = int(random.uniform(-dy, dy))
        image2 = Image.new('RGB',
                           (width + abs(x1) + abs(x2),
                            height + abs(y1) + abs(y2)))
        image2.paste(image, (abs(x1), abs(y1)))
        width2, height2 = image2.size
        return image2.transform(
            (width, height), Image.QUAD,
            (x1, y1,
             -x1, height2 - y2,
             width2 + x2, height2 + y2,
             width2 - x2, -y1))

    @staticmethod
    def offset(image, dx_factor=0.1, dy_factor=0.2):
        width, height = image.size
        dx = int(random.random() * width * dx_factor)
        dy = int(random.random() * height * dy_factor)
        image2 = Image.new('RGB', (width + dx, height + dy))
        image2.paste(image, (dx, dy))
        return image2

    @staticmethod
    def rotate(image, angle=25):
        return image.rotate(
            random.uniform(-angle, angle), Image.BILINEAR, expand=1)

    def captcha(self, path=None, fmt='JPEG'):
        """Create a captcha.

        Args:
            path: save path, default None.
            fmt: image format, PNG / JPEG.
        Returns:
            A tuple, (name, text, StringIO.value).
            For example:
                ('fXZJN4AFxHGoU5mIlcsdOypa', 'JGW9', '\x89PNG\r\n\x1a\n\x00\x00\x00\r...')

        """
        image = Image.new('RGB', (self.width, self.height), (255, 255, 255))
        image = self.background(image)
        image = self.text(image, self.fonts, drawings=['warp', 'rotate', 'offset'])
        image = self.curve(image)
        image = self.noise(image)
        image = self.smooth(image)
        name = "".join(random.sample(string.ascii_lowercase + string.ascii_uppercase + '3456789', 24))
        text = "".join(self._text)
        out = BytesIO()
        image.save(out, format=fmt)
        if path:
            image.save(os.path.join(path, name), fmt)
        return name, text, out.getvalue()

    def generate_captcha(self):
        self.initialize()
        return self.captcha("")

# 全局变量。使用就调用generate_captcha()即可
captcha = Captcha.instance()

if __name__ == '__main__':
    x = captcha.generate_captcha()
    y = "%s.jpg" % x[1]
    print(x)
    with open(y,"wb") as f:
        f.write(x[2])


text:
import numpy as np
import scipy.interpolate
from numpy import polyint, polymul, polyval
from scipy.interpolate import BSpline as SciBSpline, PPoly

from ..._utils import _domain_range
from ._basis import Basis

class BSpline(Basis):
    r"""BSpline basis.

    BSpline basis elements are defined recursively as:

    .. math::
        B_{i, 1}(x) = 1 \quad \text{if } t_i \le x < t_{i+1},
        \quad 0 \text{ otherwise}

    .. math::
        B_{i, k}(x) = \frac{x - t_i}{t_{i+k} - t_i} B_{i, k-1}(x)
        + \frac{t_{i+k+1} - x}{t_{i+k+1} - t_{i+1}} B_{i+1, k-1}(x)

    Where k indicates the order of the spline.

    Implementation details: In order to allow a discontinuous behaviour at
    the boundaries of the domain it is necessary to placing m knots at the
    boundaries [RS05]_. This is automatically done so that the user only has to
    specify a single knot at the boundaries.

    Attributes:
        domain_range (tuple): A tuple of length 2 containing the initial and
            end values of the interval over which the basis can be evaluated.
        n_basis (int): Number of functions in the basis.
        order (int): Order of the splines. One greather than their degree.
        knots (list): List of knots of the spline functions.

    Examples:
        Constructs specifying number of basis and order.

        >>> bss = BSpline(n_basis=8, order=4)

        If no order is specified defaults to 4 because cubic splines are
        the most used. So the previous example is the same as:

        >>> bss = BSpline(n_basis=8)

        It is also possible to create a BSpline basis specifying the knots.

        >>> bss = BSpline(knots=[0, 0.2, 0.4, 0.6, 0.8, 1])

        Once we create a basis we can evaluate each of its functions at a
        set of points.

        >>> bss = BSpline(n_basis=3, order=3)
        >>> bss([0, 0.5, 1])
        array([[[ 1.  ],
                [ 0.25],
                [ 0.  ]],
               [[ 0.  ],
                [ 0.5 ],
                [ 0.  ]],
               [[ 0.  ],
                [ 0.25],
                [ 1.  ]]])

        And evaluates first derivative

        >>> deriv = bss.derivative()
        >>> deriv([0, 0.5, 1])
        array([[[-2.],
                [-1.],
                [ 0.]],
               [[ 2.],
                [ 0.],
                [-2.]],
               [[ 0.],
                [ 1.],
                [ 2.]]])

    References:
        .. [RS05] Ramsay, J., Silverman, B. W. (2005). *Functional Data
            Analysis*. Springer. 50-51.

    """

    def __init__(self, domain_range=None, n_basis=None, order=4, knots=None):
        """Bspline basis constructor.

        Args:
            domain_range (tuple, optional): Definition of the interval where
                the basis defines a space. Defaults to (0,1) if knots are not
                specified. If knots are specified defaults to the first and
                last element of the knots.
            n_basis (int, optional): Number of splines that form the basis.
            order (int, optional): Order of the splines. One greater that
                their degree. Defaults to 4 which mean cubic splines.
            knots (array_like): List of knots of the splines. If domain_range
                is specified the first and last elements of the knots have to
                match with it.

        """

        if domain_range is not None:
            domain_range = _domain_range(domain_range)

            if len(domain_range) != 1:
                raise ValueError("Domain range should be unidimensional.")

            domain_range = domain_range[0]

        # Knots default to equally space points in the domain_range
        if knots is None:
            if n_basis is None:
                raise ValueError("Must provide either a list of knots or the"
                                 "number of basis.")
        else:
            knots = tuple(knots)
            knots = sorted(knots)
            if domain_range is None:
                domain_range = (knots[0], knots[-1])
            else:
                if domain_range[0] != knots[0] or domain_range[1] != knots[-1]:
                    raise ValueError("The ends of the knots must be the same "
                                     "as the domain_range.")

        # n_basis default to number of knots + order of the splines - 2
        if n_basis is None:
            n_basis = len(knots) + order - 2

        if (n_basis - order + 2) < 2:
            raise ValueError(f"The number of basis ({n_basis}) minus the "
                             f"order of the bspline ({order}) should be "
                             f"greater than 3.")

        self._order = order
        self._knots = None if knots is None else tuple(knots)
        super().__init__(domain_range=domain_range, n_basis=n_basis)

        # Checks
        if self.n_basis != self.order + len(self.knots) - 2:
            raise ValueError(f"The number of basis ({self.n_basis}) has to "
                             f"equal the order ({self.order}) plus the "
                             f"number of knots ({len(self.knots)}) minus 2.")

    @property
    def knots(self):
        if self._knots is None:
            return tuple(np.linspace(*self.domain_range[0],
                                     self.n_basis - self.order + 2))
        else:
            return self._knots

    @property
    def order(self):
        return self._order

    def _evaluation_knots(self):
        """
        Get the knots adding m knots to the boundary in order to allow a
        discontinuous behaviour at the boundaries of the domain [RS05]_.

        References:
            .. [RS05] Ramsay, J., Silverman, B. W. (2005). *Functional Data
                Analysis*. Springer. 50-51.
        """
        return np.array((self.knots[0],) * (self.order - 1) + self.knots +
                        (self.knots[-1],) * (self.order - 1))

    def _evaluate(self, eval_points):

        # Input is scalar
        eval_points = eval_points[..., 0]

        # Places m knots at the boundaries
        knots = self._evaluation_knots()

        # c is used the select which spline the function splev below computes
        c = np.zeros(len(knots))

        # Initialise empty matrix
        mat = np.empty((self.n_basis, len(eval_points)))

        # For each basis computes its value for each evaluation point
        for i in range(self.n_basis):
            # write a 1 in c in the position of the spline calculated in each
            # iteration
            c[i] = 1
            # compute the spline
            mat[i] = scipy.interpolate.splev(eval_points,
                                             (knots, c, self.order - 1))
            c[i] = 0

        return mat

    def _derivative_basis_and_coefs(self, coefs, order=1):
        if order >= self.order:
            return (
                BSpline(n_basis=1, domain_range=self.domain_range, order=1),
                np.zeros((len(coefs), 1)))

        deriv_splines = [self._to_scipy_BSpline(coefs[i]).derivative(order)
                         for i in range(coefs.shape[0])]

        deriv_coefs = [BSpline._from_scipy_BSpline(spline)[1]
                       for spline in deriv_splines]

        deriv_basis = BSpline._from_scipy_BSpline(deriv_splines[0])[0]

        return deriv_basis, np.array(deriv_coefs)[:, 0:deriv_basis.n_basis]

    def rescale(self, domain_range=None):
        r"""Return a copy of the basis with a new domain range, with the
            corresponding values rescaled to the new bounds.
            The knots of the BSpline will be rescaled in the new interval.

            Args:
                domain_range (tuple, optional): Definition of the interval
                    where the basis defines a space. Defaults uses the same as
                    the original basis.
        """

        knots = np.array(self.knots, dtype=np.dtype('float'))

        if domain_range is not None:  # Rescales the knots
            knots -= knots[0]
            knots *= ((domain_range[1] - domain_range[0]
                       ) / (self.knots[-1] - self.knots[0]))
            knots += domain_range[0]

            # Fix possible round error
            knots[0] = domain_range[0]
            knots[-1] = domain_range[1]

        else:
            # TODO: Allow multiple dimensions
            domain_range = self.domain_range[0]

        return BSpline(domain_range, self.n_basis, self.order, knots)

    def __repr__(self):
        """Representation of a BSpline basis."""
        return (f"{self.__class__.__name__}(domain_range={self.domain_range}, "
                f"n_basis={self.n_basis}, order={self.order}, "
                f"knots={self.knots})")

    def _gram_matrix(self):
        # Places m knots at the boundaries
        knots = self._evaluation_knots()

        # c is used the select which spline the function
        # PPoly.from_spline below computes
        c = np.zeros(len(knots))

        # Initialise empty list to store the piecewise polynomials
        ppoly_lst = []

        no_0_intervals = np.where(np.diff(knots) > 0)[0]

        # For each basis gets its piecewise polynomial representation
        for i in range(self.n_basis):

            # Write a 1 in c in the position of the spline
            # transformed in each iteration
            c[i] = 1

            # Gets the piecewise polynomial representation and gets
            # only the positions for no zero length intervals
            # This polynomial are defined relatively to the knots
            # meaning that the column i corresponds to the ith knot.
            # Let the ith knot be a
            # Then f(x) = pp(x - a)
            pp = PPoly.from_spline((knots, c, self.order - 1))
            pp_coefs = pp.c[:, no_0_intervals]

            # We have the coefficients for each interval in coordinates
            # (x - a), so we will need to subtract a when computing the
            # definite integral
            ppoly_lst.append(pp_coefs)
            c[i] = 0

        # Now for each pair of basis computes the inner product after
        # applying the linear differential operator
        matrix = np.zeros((self.n_basis, self.n_basis))

        for interval in range(len(no_0_intervals)):
            for i in range(self.n_basis):
                poly_i = np.trim_zeros(ppoly_lst[i][:,
                                                    interval], 'f')
                # Indefinite integral
                square = polymul(poly_i, poly_i)
                integral = polyint(square)

                # Definite integral
                matrix[i, i] += np.diff(polyval(
                    integral, self.knots[interval: interval + 2]
                    - self.knots[interval]))[0]

                # The Gram matrix is banded, so not all intervals are used
                for j in range(i + 1, min(i + self.order, self.n_basis)):
                    poly_j = np.trim_zeros(ppoly_lst[j][:, interval], 'f')

                    # Indefinite integral
                    integral = polyint(polymul(poly_i, poly_j))

                    # Definite integral
                    matrix[i, j] += np.diff(polyval(
                        integral, self.knots[interval: interval + 2]
                        - self.knots[interval])
                    )[0]

                    # The matrix is symmetric
                    matrix[j, i] = matrix[i, j]

        return matrix

    def _to_scipy_BSpline(self, coefs):

        knots = np.concatenate((
            np.repeat(self.knots[0], self.order - 1),
            self.knots,
            np.repeat(self.knots[-1], self.order - 1)))

        return SciBSpline(knots, coefs, self.order - 1)

    @staticmethod
    def _from_scipy_BSpline(bspline):
        order = bspline.k
        knots = bspline.t

        # Remove additional knots at the borders
        if order != 0:
            knots = knots[order: -order]

        coefs = bspline.c
        domain_range = [knots[0], knots[-1]]

        return BSpline(domain_range, order=order + 1, knots=knots), coefs

    @property
    def inknots(self):
        """Return number of basis."""
        return self.knots[1:len(self.knots) - 1]

    def __eq__(self, other):
        return (super().__eq__(other)
                and self.order == other.order
                and self.knots == other.knots)

    def __hash__(self):
        return hash((super().__hash__(), self.order, self.knots))


text:
# Copyright 2015: Mirantis Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from rally import osclients
from rally.task import scenario

# NOTE(boris-42): Shortcut to remove import of both rally.task.scenario and
#                 rally.plugins.openstack.scenario
configure = scenario.configure

class OpenStackScenario(scenario.Scenario):
    """Base class for all OpenStack scenarios."""

    def __init__(self, context=None, admin_clients=None, clients=None):
        super(OpenStackScenario, self).__init__(context)
        if context:
            api_info = {}
            if "api_versions" in context.get("config", {}):
                api_versions = context["config"]["api_versions"]
                for service in api_versions:
                    api_info[service] = {
                        "version": api_versions[service].get("version"),
                        "service_type": api_versions[service].get(
                            "service_type")}
            if "admin" in context:
                self._admin_clients = osclients.Clients(
                    context["admin"]["credential"], api_info)
            if "user" in context:
                self._clients = osclients.Clients(
                    context["user"]["credential"], api_info)

        if admin_clients:
            if hasattr(self, "_admin_clients"):
                raise ValueError(
                    "Only one of context[\"admin\"] or admin_clients"
                    " must be supplied")
            self._admin_clients = admin_clients
        if clients:
            if hasattr(self, "_clients"):
                raise ValueError(
                    "Only one of context[\"user\"] or clients"
                    " must be supplied")
            self._clients = clients

    def clients(self, client_type, version=None):
        """Returns a python openstack client of the requested type.

        The client will be that for one of the temporary non-administrator
        users created before the benchmark launch.

        :param client_type: Client type ("nova"/"glance" etc.)
        :param version: client version ("1"/"2" etc.)

        :returns: Standard python OpenStack client instance
        """
        client = getattr(self._clients, client_type)

        return client(version) if version is not None else client()

    def admin_clients(self, client_type, version=None):
        """Returns a python admin openstack client of the requested type.

        :param client_type: Client type ("nova"/"glance" etc.)
        :param version: client version ("1"/"2" etc.)

        :returns: Python openstack client object
        """
        client = getattr(self._admin_clients, client_type)

        return client(version) if version is not None else client()


text:
import socket
import struct
import os

PACKET_SIZE = 1024
TIME_OUT = 5
SUCCESS = b'File Has Been Transferred'

def getPayload(fileName):
    try:
        with open(file=fileName, mode="r+b") as readFile:
            payload = readFile.read()
            if len(payload) == 0:
                print("That is a blank file.\nProgram now exiting ...")
                exit()

            return payload

    except FileNotFoundError:
        print("\nNo payload file.\nProgram now exiting ...")
        exit()

def main():
    # fileName = "test.txt"
    # serverIP = "127.0.0.1"
    # serverPort = 5005

    fileName = input("Enter path of the file to be sent to the server:\n")
    payload = getPayload(fileName=fileName)
    print("File Found ...")

    serverIP = input("\nEnter the IP Address of the server:\n")
    if serverIP is None:
        print("Cannot leave server IP address blank.\nProgram now exiting ...")
        exit()

    try:
        serverPort = int(input("\nEnter the Port of the server:\n"))
    except ValueError as ve:
        print("Please provide a valid port number. Should only contain character 0-9.\nProgram now exiting ...")
        exit()

    if serverPort is None:
        print("Cannot leave server port blank.\nProgram now exiting ...")
        exit()

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(TIME_OUT)

            sock.settimeout(TIME_OUT)
            sock.connect((serverIP, serverPort))

            print("\nTransferring File ...")

            name = fileName.split("/")[-1]
            nameBytes = name.encode("utf-8")
            nameLength = len(nameBytes)

            nameSizeBytes = struct.pack("I", nameLength)
            payloadLength = len(payload) + 8 + nameLength

            numPackets = (payloadLength // PACKET_SIZE)
            if (payloadLength / PACKET_SIZE) > numPackets:
                numPackets += 1

            packedNumBytes = struct.pack('I', numPackets)

            header = packedNumBytes + nameSizeBytes + nameBytes

            payload = header + payload

            sock.sendall(payload)
            data = sock.recv(PACKET_SIZE)

        print("\nStatus:")
        print(data.decode("utf-8"))
        print("\nProgram done ...")

    except ConnectionRefusedError or ConnectionResetError as e:
        print(f"\n{e} Error Occurred. Check for correct server IP Address and Ports. Check server status.\nProgram now exiting ...")

    except Exception as e:
        print(f"\n{e} error has broken things.")

if __name__ == '__main__':
    os.system("clear")
    main()


text:
from setuptools import find_packages, setup

setup(
    author="Zackary Troop",
    name="wave-reader",
    version="0.0.10",
    url="https://github.com/ztroop/wave-reader-utils",
    license="MIT",
    description="Unofficial package for Airthings Wave communication.",
    long_description=open("README.md").read(),
    long_description_content_type="text/markdown",
    packages=find_packages(exclude=["examples", "tests"]),
    install_requires=[
        "Authlib>=0.15.4",
        "bleak>=0.10.0",
        "httpx>=0.18.2",
    ],
    python_requires=">=3.7.*",
    classifiers=[
        "Intended Audience :: Developers",
        "License :: OSI Approved :: MIT License",
        "Operating System :: OS Independent",
        "Programming Language :: Python :: 3.7",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
    ],
)


text:
import inspect
from collections import OrderedDict
from json.decoder import JSONDecodeError
from typing import Optional, Tuple, Union
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup

from recipe_scrapers.settings import settings

from ._schemaorg import SchemaOrg

# some sites close their content for 'bots', so user-agent must be supplied
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.0.7) Gecko/2009021910 Firefox/3.0.7"
}

class AbstractScraper:
    def __init__(
        self,
        url,
        proxies: Optional[str] = None,  # allows us to specify optional proxy server
        timeout: Optional[
            Union[float, Tuple, None]
        ] = None,  # allows us to specify optional timeout for request
    ):
        if settings.TEST_MODE:  # when testing, we load a file
            page_data = url.read()
            url = "https://test.example.com/"
        else:
            page_data = requests.get(
                url, headers=HEADERS, proxies=proxies, timeout=timeout
            ).content

        self.soup = BeautifulSoup(page_data, "html.parser")
        self.url = url

        # Attempt to read Schema.org data. Gracefully fail if it raises an exception parsing the JSON.
        # The scraper subclass can use BeautifulSoup to extract the information.
        try:
            self.schema = SchemaOrg(page_data)
        except JSONDecodeError:
            pass

        # attach the plugins as instructed in settings.PLUGINS
        for name, func in inspect.getmembers(self, inspect.ismethod):
            current_method = getattr(self.__class__, name)
            for plugin in reversed(settings.PLUGINS):
                if plugin.should_run(self.host(), name):
                    current_method = plugin.run(current_method)
            setattr(self.__class__, name, current_method)

    @classmethod
    def host(cls) -> str:
        """ get the host of the url, so we can use the correct scraper """
        raise NotImplementedError("This should be implemented.")

    def canonical_url(self):
        canonical_link = self.soup.find("link", {"rel": "canonical", "href": True})
        if canonical_link:
            return urljoin(self.url, canonical_link["href"])
        return self.url

    def title(self):
        raise NotImplementedError("This should be implemented.")

    def total_time(self):
        """ total time it takes to preparate the recipe in minutes """
        raise NotImplementedError("This should be implemented.")

    def yields(self):
        """ The number of servings or items in the recipe """
        raise NotImplementedError("This should be implemented.")

    def image(self):
        raise NotImplementedError("This should be implemented.")

    def nutrients(self):
        raise NotImplementedError("This should be implemented.")

    def language(self):
        """
        Human language the recipe is written in.

        May be overridden by individual scrapers.
        """
        candidate_languages = OrderedDict()
        html = self.soup.find("html", {"lang": True})
        candidate_languages[html.get("lang")] = True

        # Deprecated: check for a meta http-equiv header
        # See: https://www.w3.org/International/questions/qa-http-and-lang
        meta_language = (
            self.soup.find(
                "meta",
                {
                    "http-equiv": lambda x: x and x.lower() == "content-language",
                    "content": True,
                },
            )
            if settings.META_HTTP_EQUIV
            else None
        )
        if meta_language:
            language = meta_language.get("content").split(",", 1)[0]
            if language:
                candidate_languages[language] = True

        # If other langs exist, remove 'en' commonly generated by HTML editors
        if len(candidate_languages) > 1:
            candidate_languages.pop("en", None)

        # Return the first candidate language
        return candidate_languages.popitem(last=False)[0]

    def ingredients(self):
        raise NotImplementedError("This should be implemented.")

    def instructions(self):
        raise NotImplementedError("This should be implemented.")

    def ratings(self):
        raise NotImplementedError("This should be implemented.")

    def author(self):
        raise NotImplementedError("This should be implemented.")

    def reviews(self):
        raise NotImplementedError("This should be implemented.")

    def links(self):
        invalid_href = {"#", ""}
        links_html = self.soup.findAll("a", href=True)

        return [link.attrs for link in links_html if link["href"] not in invalid_href]

    def site_name(self):
        meta = self.soup.find("meta", property="og:site_name")
        return meta.get("content") if meta else None


text:
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
@package EFIT2D_Classes

Support Library: efit2d-pyopencl

Manuscript Title: Optimized OpenCL implementation of the Elastodynamic Finite Integration Technique for viscoelastic media

Authors: M Molero, U Iturraran-Viveros, S Aparicio, M.G. Hernández

Program title: EFIT2D-PyOpenCL

Journal reference: Comput. Phys. Commun.

Programming language: Python.

External routines:  numpy, scipy, matplotlib, glumpy, pyopencl

Computer: computers having GPU or Multicore CPU with OpenCL drivers.

	All classes here defined are used to define:
		- The scenario,
		- Material objects,
		- Input sources,
		- Inspection setup,
		- Simulation parameters

"""

import		  numpy				  as      np
from		  math				  import  sin, cos, sqrt, pi, exp
import		  random
import		  time
from		  scipy				  import  signal
from		  scipy.fftpack		  import  fftshift
from skimage.transform import rotate

try:
	from	Image import		Image
except:
	from    PIL import Image

from		  matplotlib		  import  cm
import		  matplotlib.pyplot	  as      plt

def imresize(arr, size, **kwargs):
    from PIL import Image
    size_list = [int(arr.shape[0] * size), int(arr.shape[1] * size)]
    return np.array(Image.fromarray(arr).resize(size_list))

def imrotate(arr, angle, **kwargs):
    return rotate(arr, angle=angle)

def RaisedCosinePulse(t, Freq, Amplitude):
	"""
	Raised-Cosine Pulse

	@param t   time vector
	@param Freq  Frequency in Hz
	@param Amplitude  Real Value of Amplitude

	@return  Output signal vector
	@retval  P vector of length equals to the time vector t

	"""
	N  = np.size(t,0)
	P  = np.zeros((N,),dtype=np.float32)
	for m in range(0,N):
		if t[m] <= 2.0/Freq:
			P[m] = Amplitude *(1-cos(pi*Freq*t[m]))*cos(2*pi*Freq*t[m])

	return P

def	ricker(t,ts,fsavg):
	"""
	Ricker Pulse

	@param t      time vector
	@param ts     temporal delay
	@param fsavg  pulse width parameter

	@return  Output signal vector
	"""

	a  = fsavg*pi*(t-ts)
	a2 = a*a
	return ((1.0-2.0*a2)*np.exp(-a2))

##

class NewImage:
	"""
	Class NewImage: Definition of the Main Geometric Scenario.
	"""

	def __init__(self, Width=40, Height=40,Pixel_mm=10,label=0,SPML=False):
		"""
		Constructor of the Class NewImage

		@param Width      Width of the Scenario
		@param Height     Height of the Scenario
		@param Pixel_mm   Ratio Pixel per mm
		@param label      Label
		@param SPML       Flag used to indicate the boundary conditions

		"""
		## Width of the Scenario
		self.Width		 = Width
		## Height of the Scenario
		self.Height		 = Height
		## Ratio Pixel per mm
		self.Pixel_mm	 = Pixel_mm
		## Label
		self.Label   	 = label

		## Flag used to indicate the boundary conditions
		self.SPML		 = SPML

		## Dimension 1 of the Scenario Matrix
		self.M			 = int(self.Height * self.Pixel_mm)
		## Dimension 2 od the Scenario Matrix
		self.N			 = int(self.Width  * self.Pixel_mm)

		## Scenarion Matrix (MxN)
		self.I			 = np.ones((self.M,self.N),dtype=np.uint8)*label
		self.Itemp		 = 0

		## Size of the Boundary Layer
		self.Tap         = 0

		## Configure if boundary layers will be treated as absorbing layers or air layers.
		#
		#   False: Absorbing layers
		#
		#   True : Air boundaries
		self.AirBoundary = False

	def createLayer(self, centerW, centerH, Width, Height, label, Theta=0):
		"""
		Create a Layer

		@param  centerW   center in width-axis of the Layer
		@param  centerH   center in height-axis of the Layer
		@param  Width     Width of the Layer
		@param  Height    Height of the Layer
		@param  label     Label of the layer
		@param  Theta     Rotation Angle
		"""

		a  = int(Height*self.Pixel_mm/2.0)
		b  = int(Width*self.Pixel_mm/2.0)
		for	 x in  range(-a,a):
			for y in range(-b,b):
				tempX = round (x + centerH*self.Pixel_mm)
				tempY = round (y + centerW*self.Pixel_mm)
				self.I[tempX,tempY] = label

		if Theta != 0:
			self.I = imrotate(self.I,Theta,interp='nearest')

	def createABS(self,Tap):
		"""
		Create the boundary layers depending on the boundary conditions required

		@param Tap  Layer Size

		"""

		self.Tap		 = Tap
		self.SPML		 = True

		self.AirBoundary = False

		self.M, self.N   = np.shape(self.I)

		TP		     = round(Tap* self.Pixel_mm )
		M_pml		 = int( self.M	 + 2*TP )
		N_pml		 = int( self.N	 + 2*TP )

		self.Itemp		 = 255.0*np.ones((M_pml,N_pml),dtype=np.uint8)
		self.Itemp[TP : M_pml-TP, TP : N_pml-TP] = np.copy(self.I)

class Material:
	"""
	Class Material: Definition of a material

	@param name Material Name
	@param rho  Density (kg/m3)
	@param c11  C11 (Pa)
	@param c12  C12 (Pa)
	@param c22  C22 (Pa)
	@param c44  C44 (Pa)
	@param eta_v Bulk  Viscosity Constant (Pa s)
	@param eta_s Shear Viscosity Constant (Pa s)
	@param label Material Label

	"""
	def __init__(self, name="Water",rho=1000,c11=2.19e9,c12=0.0,c22=0.0,c44=0.0,eta_v=0, eta_s=0,label=0):

		"""
		Constructor of the Material object
		"""
		## Material Name
		self.name	=  name

		##Density (kg/m3)
		self.rho	=  rho

		## C11 (Pa)
		self.c11	=  c11

		## C12 (Pa)
		self.c12	=  c12

		## C22 (Pa)
		self.c22	=  c22

		## C44 (Pa)
		self.c44	=  c44

		## Longitudinal Velocity (m/s)
		self.VL		=  sqrt( c11/rho )

		## Shear Velocity (m/s)
		self.VT		=  sqrt( c44/rho )

		## Bulk  Viscosity Constant (Pa s)
		self.eta_v  =  eta_v

		## Shear Viscosity Constant (Pa s)
		self.eta_s  =  eta_s

		## Material Label
		self.Label	=  label

	def __str__(self):
			return "Material:"

	def __repr__(self):
			return "Material:"

class Source:
	"""
	Class Source: Define the Inspection Type

	@param TypeLaunch   Type of Inspection: Transmission or PulseEcho

	"""
	def __init__(self,TypeLaunch = 'Transmission'):

		##  Type of Inspection: Transmission or PulseEcho
		self.TypeLaunch		    = TypeLaunch

		## Define the location of the transducers in function of the type of the Inspection
		self.Theta			    = 0

		if	 self.TypeLaunch == 'PulseEcho':
			self.pulseEcho()

		elif self.TypeLaunch == 'Transmission':
			self.transmission()

	def __str__(self):
		return "Source: "

	def __repr__(self):
		return "Source: "

	def pulseEcho(self):
		"""
		Define Theta for PulseEcho Inspection. PulseEcho Inspection uses the same transducer acting as emitter and as receiver
		"""
		self.Theta = [270*pi/180, 270*pi/180]

	def transmission(self):
		"""
		Define Theta for Transmission Inspection. Transmision uses two transducers, one used as emitter and another as receiver
		"""
		self.Theta = [270*pi/180, 90*pi/180]

class Transducer:
	"""
	Class Transducer:  Definition of the Transducer Object

	@param Size          Transducer Size
	@param Offset        Offset position of the Transducer. By default is set to zero
	@param BorderOffset  Border offset position of the Transducer. By default is set to zero
	@param Location      Location is set to zero that indicates Up location
	@param name          Transducer Name

	"""
	def __init__(self, Size = 10, Offset=0, BorderOffset=0, Location=0, name = 'emisor'):
		"""
		Constructor of the Class Transducer
		"""

		# Location = 0 => Top

		## Transducer Size
		self.Size		  = Size

		## Offset position of the Transducer. By default is set to zero
		#
		# This offset is measured taking into account the center of the Scenario in the width-axis
		#
		# Positive Values indicate offsets toward the right
		#
		# Negative values indicate offsets toward the left
		self.Offset		  = Offset

		## Border offset position of the Transducer. By default is set to zero
		#
		# This border offset takes into account the center od the Scenario in the width axis
		# but this offset is measured in direction of the height-axis
		#
		# Only Positive values must be defined.
		self.BorderOffset = BorderOffset

		##Size of the trasnducer in Pixels
		self.SizePixel	  = 0

		## Location-> 0: Top. This version only works when the location=0
		self.Location	  = Location

		## Name of the transducer
		self.name         = name

	def __str__(self):
		return "Transducer: "

	def __repr__(self):
		return "Transducer: "

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

class Signal:
	"""
	Class Signal: Signal Definition (Source Input for the Simulation)

	@param Amplitude   Signal Amplitude
	@param Frequency   Frequency Amplitude
	@param Name        Name of the Signal:  RaisedCosinePulse or RickerPulse
	@param ts          Time Delay: used only for RickerPulse

	"""
	def __init__(self, Amplitude=1, Frequency=1e6, name ="RaisedCosinePulse", ts=1):

		## Signal Amplitude
		self.Amplitude = Amplitude

		## Frequency Amplitude
		self.Frequency = Frequency

		## Name of the Signal:  RaisedCosinePulse or RickerPulse
		self.name      = name

		## Time Delay: used only for RickerPulse
		if ts == 1:
			self.ts        = 3.0/Frequency;

	def __str__(self):
		return "Signal: "

	def __repr__(self):
		return "Signal: "

	def generate(self,t):
		"""
		Generate the signal waveform

		@param t  vector time
		@return signal vector with the same length as the vector time

		"""

		if self.name == "RaisedCosinePulse":
			return RaisedCosinePulse(t, self.Frequency, self.Amplitude)
		elif self.name == "RickerPulse":
			return ricker(t, self.ts, self.Frequency)

	def saveSignal(self,t):
		"""
		Save the signal waveform into the object
		@param t  vector time

		"""
		self.time_signal  = self.generate(t)

######################################
class Inspection:
	"""
	Class Inspection:  used for the configuration of the inspections to be emulated
	"""

	def __init__(self):
		"""
		Constructor of the Class Inspection
		"""

		## Position of the Transducer (Angle)
		self.Theta	= 0

		## Vector x-axis Position of the Transducer
		self.XL		= 0

		## Vector y-axis Position of the Transducer
		self.YL		= 0

		##
		self.IR		= 0

	def __str__(self):
		return "Inspection: "

	def __repr__(self):
		return "Inspection: "

	def setTransmisor(self, source, transducer, x2, y2, X0, Y0):

		self.Theta	= source.Theta

		Ntheta		= np.size(self.Theta,0)
		NXL			= int(2*transducer.SizePixel)

		xL			= np.zeros((NXL,),dtype=np.float32)
		yL			= np.zeros((NXL,),dtype=np.float32)

		for m in range(0,Ntheta):

			if np.abs(np.cos(self.Theta[m])) < 1e-5:
				yL = np.linspace(y2[m]-transducer.SizePixel,y2[m]+transducer.SizePixel,num=NXL, endpoint=True)
				xL[:] = x2[m]*np.ones((NXL,),dtype=np.float32)

			elif np.abs(np.cos(self.Theta[m])) == 1:
				xL[:] = np.linspace(x2[m]-transducer.SizePixel, x2[m]+transducer.SizePixel,num=NXL, endpoint=True)
				yL[:] = y2[m] - ( (x2[m]-X0 )/( y2[m]-Y0 ) )*( xL[:]-x2[m] )

			else:
				xL[:] = np.linspace(x2[m]-(transducer.SizePixel*np.abs(np.cos(self.Theta[m]))),x2[m]+(transducer.SizePixel*np.abs(np.cos(self.Theta[m]))), num=NXL, endpoint=True )
				yL[:] = y2[m] - ( (x2[m]-X0 )/( y2[m]-Y0 )	)*( xL[:]-x2[m] )

			if m==0:
				self.XL		= np.zeros((np.size(xL,0),Ntheta),dtype=np.float32)
				self.YL		= np.zeros((np.size(xL,0),Ntheta),dtype=np.float32)

			self.XL[:,m]  = (np.around(xL[:]))
			self.YL[:,m]  = (np.around(yL[:]))

	def addOffset(self, image, transducer, NRI):
		"""
		Handle Offset

		"""
		NXL	   = np.size(self.XL,0)
		Ntheta = np.size(self.Theta,0)

		M_pml, N_pml = np.shape(image.Itemp)

		self.YL +=	 (np.around(transducer.Offset * image.Pixel_mm * NRI / float(N_pml)))

		self.IR		 = np.zeros((Ntheta,Ntheta),dtype=np.float32)
		B			 = list(range(0,Ntheta))
		self.IR[:,0] = np.int32(B[:])

		for i in range(1,Ntheta):
			B  = np.roll(B,-1)
			self.IR[:,i] = np.int32(B)

	def addBorderOffset(self, image, transducer, MRI):
		"""
		Handle  Border Offset

		"""

		M_pml, N_pml = np.shape(image.Itemp)
		ratio = float(MRI) / float(M_pml)

		self.XL[:,0] += (np.around(transducer.BorderOffset * image.Pixel_mm * ratio) )
		self.XL[:,1] -= (np.around(transducer.BorderOffset * image.Pixel_mm * ratio) )

	def flip(self):
		self.XL       = np.fliplr(self.XL)

	def SetReception(self,T):

		ReceptorX = (self.XL)
		ReceptorY = (self.YL)
		M,N		  = np.shape(ReceptorX)
		temp  = np.zeros((M,N-1),dtype=np.float32)

		for	 mm	 in range(0,M):
			for ir in  range(0,N-1):

				temp[mm,ir]	  =	 T[ int(ReceptorX[ mm,int(self.IR[0,ir+1]) ] ) , int(ReceptorY[ mm,int(self.IR[0,ir+1]) ]) ]

		if self.Field:
			return temp.transpose()
		else:
			return np.mean(temp,0)

	def SetReceptionVector(self, T, x, y):
		M		  = np.size(x)
		temp      = np.zeros((M,),dtype=np.float32)
		for	 mm	in range(0,M):
			temp[mm] = T[(int(x[mm])),(int(y[mm]))]

		return temp

class SimulationModel:
	"""
	Class  Simulation: setup the parameters for the  numerical simulation

	Usage:
		- First Define an Instance of the SimulationModel Object
		- Execute the method class: jobParameters using as input the materials list
		- Execute the method class: createNumerical Model using as input the scenario
		- Execute the method class: initReceivers to initialize the receivers
		- Execute the mtehod class: save signal using as input the attribute simModel.t
		- Save the Device into the simModel.Device attribute

	@param TimeScale    Scale Time Factor
	@param MaxFreq      Maximum Frequency
	@param PointCycle   Points per Cycle
	@param SimTime      Time Simuation
	@param SpatialScale Spatial Scale: 1 -> meters, 1e-3 -> millimeters

	"""
	def __init__(self,TimeScale=1, MaxFreq=2e6, PointCycle=10, SimTime=50e6, SpatialScale=1e-3):

		## Scale Time Factor
		self.TimeScale	= TimeScale

		## Maximum Frequency
		self.MaxFreq	= MaxFreq	  # MHz

		## Points per Cycle
		self.PointCycle = PointCycle

		## Time Simuation
		self.SimTime	= SimTime	  # microseconds

		## Spatial Scale: 1 -> meters, 1e-3 -> millimeters
		self.SpatialScale = SpatialScale

		## Spatial Discretization
		self.dx         = 0

		## Temporal Discretization
		self.dt         = 0

		self.Rgrid      = 0
		self.TapG       = 0
		self.t          = 0
		self.Ntiempo    = 0

		self.MRI,self.NRI = (0,0)

		self.receiver_signals   = 0
		self.Device           = 'CPU'

		self.XL         = 0
		self.YL         = 0

	def __str__(self):
		return "Simulation Model: "

	def __repr__(self):
		return "Simulation Model: "

	def jobParameters(self,materiales):
		"""
		Define Main Simulation Parameters

		@parm materiales  Materials List

		"""
		indVL = [mat.VL for mat in materiales if mat.VL > 400]
		indVT = [mat.VT for mat in materiales if mat.VT > 400]

		VL	  = np.array(indVL)
		VT	  = np.array(indVT)
		V	  = np.hstack( (VL, VT) )

		self.dx = np.float32( np.min([V]) / (self.PointCycle*self.MaxFreq) )
		self.dt = self.TimeScale * np.float32( 0.7071 * self.dx / (	 np.max([V]) ) )

		self.Ntiempo = int(round(self.SimTime/self.dt))
		self.t	= self.dt*np.arange(0,self.Ntiempo)

	def createNumericalModel(self, image):
		"""
		Create the Numerical Model

		@param image  The Scenario Object
		"""

		#Spatial Scale
		Mp			      =	 np.shape(image.Itemp)[0]*self.SpatialScale/image.Pixel_mm/self.dx
		self.Rgrid	      =	 Mp/np.shape(image.Itemp)[0]

		self.TapG	      =	 np.around(image.Tap * self.Rgrid * image.Pixel_mm)
		self.Im		      =	 imresize(image.Itemp, self.Rgrid, interp='nearest')
		self.MRI,self.NRI =	 np.shape(self.Im)

		print("dt: " + str(self.dt) + " dx: "  + str(self.dx) + " Grid: " +  str(self.MRI) + " x " + str(self.NRI))

	def initReceivers(self):
		"""
		Initialize the receivers

		"""
		self.receiver_signals = 0

	def setDevice(self,Device):
		"""
		Set the Computation Device

		@param Device Device to be used

		Define the device used to compute the simulations:
			 - "CPU"        : uses the global memory in th CPU
			 - "GPU_Global" : uses the global memory in the GPU
			 - "GPU_Local"  : uses the local memory in the GPU

		"""

		if Device == 0:
			self.Device = 'CPU'
		elif Device ==1:
			self.Device = 'GPU_Global'
		elif Device ==2:
			self.Device = 'GPU_Local'


text:
# coding: utf-8
import math
import random
import time

import asyncclick as click

@click.group()
def cli():
    """This script showcases different terminal UI helpers in Click."""
    pass

@cli.command()
def colordemo():
    """Demonstrates ANSI color support."""
    for color in "red", "green", "blue":
        click.echo(click.style("I am colored {}".format(color), fg=color))
        click.echo(click.style("I am background colored {}".format(color), bg=color))

@cli.command()
def pager():
    """Demonstrates using the pager."""
    lines = []
    for x in range(200):
        lines.append("{}. Hello World!".format(click.style(str(x), fg="green")))
    click.echo_via_pager("\n".join(lines))

@cli.command()
@click.option(
    "--count",
    default=8000,
    type=click.IntRange(1, 100000),
    help="The number of items to process.",
)
def progress(count):
    """Demonstrates the progress bar."""
    items = range(count)

    def process_slowly(item):
        time.sleep(0.002 * random.random())

    def filter(items):
        for item in items:
            if random.random() > 0.3:
                yield item

    with click.progressbar(
        items, label="Processing accounts", fill_char=click.style("#", fg="green")
    ) as bar:
        for item in bar:
            process_slowly(item)

    def show_item(item):
        if item is not None:
            return "Item #{}".format(item)

    with click.progressbar(
        filter(items),
        label="Committing transaction",
        fill_char=click.style("#", fg="yellow"),
        item_show_func=show_item,
    ) as bar:
        for item in bar:
            process_slowly(item)

    with click.progressbar(
        length=count,
        label="Counting",
        bar_template="%(label)s  %(bar)s | %(info)s",
        fill_char=click.style(u"█", fg="cyan"),
        empty_char=" ",
    ) as bar:
        for item in bar:
            process_slowly(item)

    with click.progressbar(
        length=count,
        width=0,
        show_percent=False,
        show_eta=False,
        fill_char=click.style("#", fg="magenta"),
    ) as bar:
        for item in bar:
            process_slowly(item)

    # 'Non-linear progress bar'
    steps = [math.exp(x * 1.0 / 20) - 1 for x in range(20)]
    count = int(sum(steps))
    with click.progressbar(
        length=count,
        show_percent=False,
        label="Slowing progress bar",
        fill_char=click.style(u"█", fg="green"),
    ) as bar:
        for item in steps:
            time.sleep(item)
            bar.update(item)

@cli.command()
@click.argument("url")
def open(url):
    """Opens a file or URL In the default application."""
    click.launch(url)

@cli.command()
@click.argument("url")
def locate(url):
    """Opens a file or URL In the default application."""
    click.launch(url, locate=True)

@cli.command()
def edit():
    """Opens an editor with some text in it."""
    MARKER = "# Everything below is ignored\n"
    message = click.edit("\n\n{}".format(MARKER))
    if message is not None:
        msg = message.split(MARKER, 1)[0].rstrip("\n")
        if not msg:
            click.echo("Empty message!")
        else:
            click.echo("Message:\n{}".format(msg))
    else:
        click.echo("You did not enter anything!")

@cli.command()
def clear():
    """Clears the entire screen."""
    click.clear()

@cli.command()
def pause():
    """Waits for the user to press a button."""
    click.pause()

@cli.command()
def menu():
    """Shows a simple menu."""
    menu = "main"
    while 1:
        if menu == "main":
            click.echo("Main menu:")
            click.echo("  d: debug menu")
            click.echo("  q: quit")
            char = click.getchar()
            if char == "d":
                menu = "debug"
            elif char == "q":
                menu = "quit"
            else:
                click.echo("Invalid input")
        elif menu == "debug":
            click.echo("Debug menu")
            click.echo("  b: back")
            char = click.getchar()
            if char == "b":
                menu = "main"
            else:
                click.echo("Invalid input")
        elif menu == "quit":
            return


text:
import argparse
import multiprocessing
import os
import threading
import time
from typing import Any, Tuple, Union, Dict, Optional

from .helper import _get_event, ConditionalEvent
from ... import __stop_msg__, __ready_msg__, __default_host__
from ...enums import PeaRoleType, RuntimeBackendType, SocketType
from ...excepts import RuntimeFailToStart, RuntimeRunForeverEarlyError
from ...helper import typename
from ...logging.logger import JinaLogger

__all__ = ['BasePea']

def run(
    args: 'argparse.Namespace',
    name: str,
    runtime_cls,
    envs: Dict[str, str],
    is_started: Union['multiprocessing.Event', 'threading.Event'],
    is_shutdown: Union['multiprocessing.Event', 'threading.Event'],
    is_ready: Union['multiprocessing.Event', 'threading.Event'],
    cancel_event: Union['multiprocessing.Event', 'threading.Event'],
):
    """Method representing the :class:`BaseRuntime` activity.

    This method is the target for the Pea's `thread` or `process`

    .. note::
        :meth:`run` is running in subprocess/thread, the exception can not be propagated to the main process.
        Hence, please do not raise any exception here.

    .. note::
        Please note that env variables are process-specific. Subprocess inherits envs from
        the main process. But Subprocess's envs do NOT affect the main process. It does NOT
        mess up user local system envs.

    .. warning::
        If you are using ``thread`` as backend, envs setting will likely be overidden by others

    :param args: namespace args from the Pea
    :param name: name of the Pea to have proper logging
    :param runtime_cls: the runtime class to instantiate
    :param envs: a dictionary of environment variables to be set in the new Process
    :param is_started: concurrency event to communicate runtime is properly started. Used for better logging
    :param is_shutdown: concurrency event to communicate runtime is terminated
    :param is_ready: concurrency event to communicate runtime is ready to receive messages
    :param cancel_event: concurrency event to receive cancelling signal from the Pea. Needed by some runtimes
    """
    logger = JinaLogger(name, **vars(args))

    def _unset_envs():
        if envs and args.runtime_backend != RuntimeBackendType.THREAD:
            for k in envs.keys():
                os.unsetenv(k)

    def _set_envs():
        if args.env:
            if args.runtime_backend == RuntimeBackendType.THREAD:
                logger.warning(
                    'environment variables should not be set when runtime="thread".'
                )
            else:
                os.environ.update({k: str(v) for k, v in envs.items()})

    try:
        _set_envs()
        runtime = runtime_cls(
            args=args,
            cancel_event=cancel_event,
        )
    except Exception as ex:
        logger.error(
            f'{ex!r} during {runtime_cls!r} initialization'
            + f'\n add "--quiet-error" to suppress the exception details'
            if not args.quiet_error
            else '',
            exc_info=not args.quiet_error,
        )
    else:
        is_started.set()
        with runtime:
            is_ready.set()
            runtime.run_forever()
    finally:
        _unset_envs()
        is_shutdown.set()

class BasePea:
    """
    :class:`BasePea` is a thread/process- container of :class:`BaseRuntime`. It leverages :class:`threading.Thread`
    or :class:`multiprocessing.Process` to manage the lifecycle of a :class:`BaseRuntime` object in a robust way.

    A :class:`BasePea` must be equipped with a proper :class:`Runtime` class to work.
    """

    def __init__(self, args: 'argparse.Namespace'):
        super().__init__()  #: required here to call process/thread __init__
        self.args = args
        self.name = self.args.name or self.__class__.__name__

        self.logger = JinaLogger(self.name, **vars(self.args))

        if self.args.runtime_backend == RuntimeBackendType.THREAD:
            self.logger.warning(
                f' Using Thread as runtime backend is not recommended for production purposes. It is '
                f'just supposed to be used for easier debugging. Besides the performance considerations, it is'
                f'specially dangerous to mix `Executors` running in different types of `RuntimeBackends`.'
            )

        self._envs = {'JINA_POD_NAME': self.name, 'JINA_LOG_ID': self.args.identity}
        if self.args.quiet:
            self._envs['JINA_LOG_CONFIG'] = 'QUIET'
        if self.args.env:
            self._envs.update(self.args.env)

        # arguments needed to create `runtime` and communicate with it in the `run` in the stack of the new process
        # or thread. Control address from Zmqlet has some randomness and therefore we need to make sure Pea knows
        # control address of runtime
        self.runtime_cls = self._get_runtime_cls()
        self._timeout_ctrl = self.args.timeout_ctrl
        self._set_ctrl_adrr()
        test_worker = {
            RuntimeBackendType.THREAD: threading.Thread,
            RuntimeBackendType.PROCESS: multiprocessing.Process,
        }.get(getattr(args, 'runtime_backend', RuntimeBackendType.THREAD))()
        self.is_ready = _get_event(test_worker)
        self.is_shutdown = _get_event(test_worker)
        self.cancel_event = _get_event(test_worker)
        self.is_started = _get_event(test_worker)
        self.ready_or_shutdown = ConditionalEvent(
            getattr(args, 'runtime_backend', RuntimeBackendType.THREAD),
            events_list=[self.is_ready, self.is_shutdown],
        )
        self.worker = {
            RuntimeBackendType.THREAD: threading.Thread,
            RuntimeBackendType.PROCESS: multiprocessing.Process,
        }.get(getattr(args, 'runtime_backend', RuntimeBackendType.THREAD))(
            target=run,
            kwargs={
                'args': args,
                'name': self.name,
                'envs': self._envs,
                'is_started': self.is_started,
                'is_shutdown': self.is_shutdown,
                'is_ready': self.is_ready,
                'cancel_event': self.cancel_event,
                'runtime_cls': self.runtime_cls,
            },
        )
        self.daemon = self.args.daemon  #: required here to set process/thread daemon

    def _set_ctrl_adrr(self):
        """Sets control address for different runtimes"""
        self.runtime_ctrl_address = self.runtime_cls.get_control_address(
            host=self.args.host,
            port=self.args.port_ctrl,
            docker_kwargs=getattr(self.args, 'docker_kwargs', None),
        )

        if not self.runtime_ctrl_address:
            self.runtime_ctrl_address = f'{self.args.host}:{self.args.port_in}'

    def start(self):
        """Start the Pea.
        This method calls :meth:`start` in :class:`threading.Thread` or :class:`multiprocesssing.Process`.
        .. #noqa: DAR201
        """
        self.worker.start()
        if not self.args.noblock_on_start:
            self.wait_start_success()
        return self

    def join(self, *args, **kwargs):
        """Joins the Pea.
        This method calls :meth:`join` in :class:`threading.Thread` or :class:`multiprocesssing.Process`.

        :param args: extra positional arguments to pass to join
        :param kwargs: extra keyword arguments to pass to join
        """
        self.worker.join(*args, **kwargs)

    def terminate(self):
        """Terminate the Pea.
        This method calls :meth:`terminate` in :class:`threading.Thread` or :class:`multiprocesssing.Process`.
        """
        if hasattr(self.worker, 'terminate'):
            self.worker.terminate()

    def _retry_control_message(self, command: str, num_retry: int = 3):
        from ..zmq import send_ctrl_message

        for retry in range(1, num_retry + 1):
            self.logger.debug(f'Sending {command} command for the {retry}th time')
            try:
                send_ctrl_message(
                    self.runtime_ctrl_address,
                    command,
                    timeout=self._timeout_ctrl,
                    raise_exception=True,
                )
                break
            except Exception as ex:
                self.logger.warning(f'{ex!r}')
                if retry == num_retry:
                    raise ex

    def activate_runtime(self):
        """ Send activate control message. """
        self.runtime_cls.activate(
            logger=self.logger,
            socket_in_type=self.args.socket_in,
            control_address=self.runtime_ctrl_address,
            timeout_ctrl=self._timeout_ctrl,
        )

    def _cancel_runtime(self, skip_deactivate: bool = False):
        """
        Send terminate control message.

        :param skip_deactivate: Mark that the DEACTIVATE signal may be missed if set to True
        """
        self.runtime_cls.cancel(
            cancel_event=self.cancel_event,
            logger=self.logger,
            socket_in_type=self.args.socket_in,
            control_address=self.runtime_ctrl_address,
            timeout_ctrl=self._timeout_ctrl,
            skip_deactivate=skip_deactivate,
        )

    def _wait_for_ready_or_shutdown(self, timeout: Optional[float]):
        """
        Waits for the process to be ready or to know it has failed.

        :param timeout: The time to wait before readiness or failure is determined
            .. # noqa: DAR201
        """
        return self.runtime_cls.wait_for_ready_or_shutdown(
            timeout=timeout,
            ready_or_shutdown_event=self.ready_or_shutdown.event,
            ctrl_address=self.runtime_ctrl_address,
            timeout_ctrl=self._timeout_ctrl,
            shutdown_event=self.is_shutdown,
        )

    def wait_start_success(self):
        """Block until all peas starts successfully.

        If not success, it will raise an error hoping the outer function to catch it
        """
        _timeout = self.args.timeout_ready
        if _timeout <= 0:
            _timeout = None
        else:
            _timeout /= 1e3

        if self._wait_for_ready_or_shutdown(_timeout):
            if self.is_shutdown.is_set():
                # return too early and the shutdown is set, means something fails!!
                if not self.is_started.is_set():
                    raise RuntimeFailToStart
                else:
                    raise RuntimeRunForeverEarlyError
            else:
                self.logger.success(__ready_msg__)
        else:
            _timeout = _timeout or -1
            self.logger.warning(
                f'{self.runtime_cls!r} timeout after waiting for {self.args.timeout_ready}ms, '
                f'if your executor takes time to load, you may increase --timeout-ready'
            )
            self.close()
            raise TimeoutError(
                f'{typename(self)}:{self.name} can not be initialized after {_timeout * 1e3}ms'
            )

    @property
    def _is_dealer(self):
        """Return true if this `Pea` must act as a Dealer responding to a Router
        .. # noqa: DAR201
        """
        return self.args.socket_in == SocketType.DEALER_CONNECT

    def close(self) -> None:
        """Close the Pea

        This method makes sure that the `Process/thread` is properly finished and its resources properly released
        """
        # if that 1s is not enough, it means the process/thread is still in forever loop, cancel it
        self.logger.debug('waiting for ready or shutdown signal from runtime')
        if self.is_ready.is_set() and not self.is_shutdown.is_set():
            try:
                self._cancel_runtime()
                if not self.is_shutdown.wait(timeout=self._timeout_ctrl):
                    self.terminate()
                    time.sleep(0.1)
                    raise Exception(
                        f'Shutdown signal was not received for {self._timeout_ctrl}'
                    )
            except Exception as ex:
                self.logger.error(
                    f'{ex!r} during {self.close!r}'
                    + f'\n add "--quiet-error" to suppress the exception details'
                    if not self.args.quiet_error
                    else '',
                    exc_info=not self.args.quiet_error,
                )

            # if it is not daemon, block until the process/thread finish work
            if not self.args.daemon:
                self.join()
        elif self.is_shutdown.is_set():
            # here shutdown has been set already, therefore `run` will gracefully finish
            pass
        else:
            # sometimes, we arrive to the close logic before the `is_ready` is even set.
            # Observed with `gateway` when Pods fail to start
            self.logger.warning(
                'Pea is being closed before being ready. Most likely some other Pea in the Flow or Pod '
                'failed to start'
            )
            _timeout = self.args.timeout_ready
            if _timeout <= 0:
                _timeout = None
            else:
                _timeout /= 1e3
            self.logger.debug('waiting for ready or shutdown signal from runtime')
            if self._wait_for_ready_or_shutdown(_timeout):
                if not self.is_shutdown.is_set():
                    self._cancel_runtime(skip_deactivate=True)
                    if not self.is_shutdown.wait(timeout=self._timeout_ctrl):
                        self.terminate()
                        time.sleep(0.1)
                        raise Exception(
                            f'Shutdown signal was not received for {self._timeout_ctrl}'
                        )
            else:
                self.logger.warning(
                    'Terminating process after waiting for readiness signal for graceful shutdown'
                )
                # Just last resource, terminate it
                self.terminate()
                time.sleep(0.1)
        self.logger.debug(__stop_msg__)
        self.logger.close()

    def __enter__(self):
        return self.start()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def _get_runtime_cls(self) -> Tuple[Any, bool]:
        from .helper import update_runtime_cls
        from ..runtimes import get_runtime

        update_runtime_cls(self.args)
        return get_runtime(self.args.runtime_cls)

    @property
    def role(self) -> 'PeaRoleType':
        """Get the role of this pea in a pod

        .. #noqa: DAR201"""
        return self.args.pea_role

    @property
    def _is_inner_pea(self) -> bool:
        """Determine whether this is a inner pea or a head/tail

        .. #noqa: DAR201"""
        return self.role is PeaRoleType.SINGLETON or self.role is PeaRoleType.PARALLEL