# DS18x20 temperature sensor driver for MicroPython.
# MIT license; Copyright (c) 2016 Damien P. George

from micropython import const

_CONVERT = const(0x44)
_RD_SCRATCH = const(0xBE)
_WR_SCRATCH = const(0x4E)


            

Reported by Pylint.

              # DS18x20 temperature sensor driver for MicroPython.
# MIT license; Copyright (c) 2016 Damien P. George

from micropython import const

_CONVERT = const(0x44)
_RD_SCRATCH = const(0xBE)
_WR_SCRATCH = const(0x4E)


            

Reported by Pylint.

              _WR_SCRATCH = const(0x4E)


class DS18X20:
    def __init__(self, onewire):
        self.ow = onewire
        self.buf = bytearray(9)

    def scan(self):

            

Reported by Pylint.

              
class DS18X20:
    def __init__(self, onewire):
        self.ow = onewire
        self.buf = bytearray(9)

    def scan(self):
        return [rom for rom in self.ow.scan() if rom[0] in (0x10, 0x22, 0x28)]


            

Reported by Pylint.

                      self.ow = onewire
        self.buf = bytearray(9)

    def scan(self):
        return [rom for rom in self.ow.scan() if rom[0] in (0x10, 0x22, 0x28)]

    def convert_temp(self):
        self.ow.reset(True)
        self.ow.writebyte(self.ow.SKIP_ROM)

            

Reported by Pylint.

                  def scan(self):
        return [rom for rom in self.ow.scan() if rom[0] in (0x10, 0x22, 0x28)]

    def convert_temp(self):
        self.ow.reset(True)
        self.ow.writebyte(self.ow.SKIP_ROM)
        self.ow.writebyte(_CONVERT)

    def read_scratch(self, rom):

            

Reported by Pylint.

                      self.ow.writebyte(self.ow.SKIP_ROM)
        self.ow.writebyte(_CONVERT)

    def read_scratch(self, rom):
        self.ow.reset(True)
        self.ow.select_rom(rom)
        self.ow.writebyte(_RD_SCRATCH)
        self.ow.readinto(self.buf)
        if self.ow.crc8(self.buf):

            

Reported by Pylint.

                          raise Exception("CRC error")
        return self.buf

    def write_scratch(self, rom, buf):
        self.ow.reset(True)
        self.ow.select_rom(rom)
        self.ow.writebyte(_WR_SCRATCH)
        self.ow.write(buf)


            

Reported by Pylint.

                      self.ow.writebyte(_WR_SCRATCH)
        self.ow.write(buf)

    def read_temp(self, rom):
        buf = self.read_scratch(rom)
        if rom[0] == 0x10:
            if buf[1]:
                t = buf[0] >> 1 | 0x80
                t = -((~t + 1) & 0xFF)

            

Reported by Pylint.

              
    def read_temp(self, rom):
        buf = self.read_scratch(rom)
        if rom[0] == 0x10:
            if buf[1]:
                t = buf[0] >> 1 | 0x80
                t = -((~t + 1) & 0xFF)
            else:
                t = buf[0] >> 1

            

Reported by Pylint.

              
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 4
n_finished = 0

            

Reported by Pylint.

              
# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se

            se.remove(i)

            

Reported by Pylint.

              # main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se

            se.remove(i)
            assert i not in se

            

Reported by Pylint.

                          assert i not in se

    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 4

            

Reported by Pylint.

              # test concurrent mutating access to a shared set object
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

import _thread

# the shared set
se = set([-1, -2, -3, -4])


            

Reported by Pylint.

              se = set([-1, -2, -3, -4])

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se


            

Reported by Pylint.

              se = set([-1, -2, -3, -4])

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se


            

Reported by Pylint.

              se = set([-1, -2, -3, -4])

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se


            

Reported by Pylint.

              se = set([-1, -2, -3, -4])

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se


            

Reported by Pylint.

              se = set([-1, -2, -3, -4])

# main thread function
def th(n, lo, hi):
    for repeat in range(n):
        for i in range(lo, hi):
            se.add(i)
            assert i in se


            

Reported by Pylint.

                  sys.settrace
except AttributeError:
    print("SKIP")
    raise SystemExit


def print_stacktrace(frame, level=0):
    print(
        "%2d: %s@%s:%s => %s:%d"

            

Reported by Pylint.

              trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler

            

Reported by Pylint.

              

def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler


            

Reported by Pylint.

              # test sys.settrace with generators

import sys

try:
    sys.settrace
except AttributeError:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

                  raise SystemExit


def print_stacktrace(frame, level=0):
    print(
        "%2d: %s@%s:%s => %s:%d"
        % (
            level,
            "  ",

            

Reported by Pylint.

                      print_stacktrace(frame.f_back, level + 1)


trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)

            

Reported by Pylint.

              trace_count = 0


def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler

            

Reported by Pylint.

              

def trace_tick_handler(frame, event, arg):
    global trace_count
    print("### trace_handler::main event:", event)
    trace_count += 1
    print_stacktrace(frame)
    return trace_tick_handler


            

Reported by Pylint.

                  return trace_tick_handler


def test_generator():
    def make_gen():
        yield 1 << 0
        yield 1 << 1
        yield 1 << 2
        return 1 << 3

            

Reported by Pylint.

                      return 1 << 3

    gen = make_gen()
    r = 0
    try:

        r += gen.send(None)

        while True:

            

Reported by Pylint.

              
def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)


            

Reported by Pylint.

                          print(2)

        def set(self, value):
            self.value = value

        def print(self):
            print(self.value)

    C.f()

            

Reported by Pylint.

              # basic class

def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)

            

Reported by Pylint.

              # basic class

def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)

            

Reported by Pylint.

              # basic class

def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)

            

Reported by Pylint.

              # basic class

def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)

            

Reported by Pylint.

              # basic class

def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)

            

Reported by Pylint.

              
def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)


            

Reported by Pylint.

              
def go():
    class C:
        def f():
            print(1)

        def g(self):
            print(2)


            

Reported by Pylint.

                      def f():
            print(1)

        def g(self):
            print(2)

        def set(self, value):
            self.value = value


            

Reported by Pylint.

              #!/usr/bin/env python3
#
# This file is part of the MicroPython project, http://micropython.org/
#
# The MIT License (MIT)
#
# Copyright (c) 2020 Damien P. George
# Copyright (c) 2020 Jim Mussared
#

            

Reported by Pylint.

              import itertools
import os
import re
import subprocess

# Relative to top-level repo dir.
PATHS = [
    # C
    "extmod/*.[ch]",

            

Reported by Bandit.

              PY_EXTS = (".py",)


def list_files(paths, exclusions=None, prefix=""):
    files = set()
    for pattern in paths:
        files.update(glob.glob(os.path.join(prefix, pattern), recursive=True))
    for pattern in exclusions or []:
        files.difference_update(glob.fnmatch.filter(files, os.path.join(prefix, pattern)))

            

Reported by Pylint.

                  return sorted(files)


def fixup_c(filename):
    # Read file.
    with open(filename) as f:
        lines = f.readlines()

    # Write out file with fixups.

            

Reported by Pylint.

              
def fixup_c(filename):
    # Read file.
    with open(filename) as f:
        lines = f.readlines()

    # Write out file with fixups.
    with open(filename, "w", newline="") as f:
        dedent_stack = []

            

Reported by Pylint.

                      lines = f.readlines()

    # Write out file with fixups.
    with open(filename, "w", newline="") as f:
        dedent_stack = []
        while lines:
            # Get next line.
            l = lines.pop(0)


            

Reported by Pylint.

                      dedent_stack = []
        while lines:
            # Get next line.
            l = lines.pop(0)

            # Dedent #'s to match indent of following line (not previous line).
            m = re.match(r"( +)#(if |ifdef |ifndef |elif |else|endif)", l)
            if m:
                indent = len(m.group(1))

            

Reported by Pylint.

                          l = lines.pop(0)

            # Dedent #'s to match indent of following line (not previous line).
            m = re.match(r"( +)#(if |ifdef |ifndef |elif |else|endif)", l)
            if m:
                indent = len(m.group(1))
                directive = m.group(2)
                if directive in ("if ", "ifdef ", "ifndef "):
                    l_next = lines[0]

            

Reported by Pylint.

                                  indent_next = len(re.match(r"( *)", l_next).group(1))
                    if indent - 4 == indent_next and re.match(r" +(} else |case )", l_next):
                        # This #-line (and all associated ones) needs dedenting by 4 spaces.
                        l = l[4:]
                        dedent_stack.append(indent - 4)
                    else:
                        # This #-line does not need dedenting.
                        dedent_stack.append(-1)
                else:

            

Reported by Pylint.

                                  if dedent_stack[-1] >= 0:
                        # This associated #-line needs dedenting to match the #if.
                        indent_diff = indent - dedent_stack[-1]
                        assert indent_diff >= 0
                        l = l[indent_diff:]
                    if directive == "endif":
                        dedent_stack.pop()

            # Write out line.

            

Reported by Bandit.

              except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit



            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              
try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")

            

Reported by Pylint.

                  from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit


# test with heap locked to see it switch to number-only error message
def test(addr):
    alloc_emergency_exception_buf(256)

            

Reported by Pylint.

              

# test with heap locked to see it switch to number-only error message
def test(addr):
    alloc_emergency_exception_buf(256)
    s = socket.socket()
    s.connect(addr)
    try:
        s.setblocking(False)

            

Reported by Pylint.

                  s.connect(addr)
    try:
        s.setblocking(False)
        s = ssl.wrap_socket(s, do_handshake=False)
        heap_lock()
        print("heap is locked")
        while True:
            ret = s.write("foo")
            if ret:

            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              # test that modtls produces a numerical error message when out of heap

try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

              try:
    import usocket as socket, ussl as ssl, sys
except:
    import socket, ssl, sys
try:
    from micropython import alloc_emergency_exception_buf, heap_lock, heap_unlock
except:
    print("SKIP")
    raise SystemExit

            

Reported by Pylint.

              try:
    import usys as sys
except ImportError:
    import sys

try:
    import ubinascii as binascii
except:
    import binascii

            

Reported by Pylint.

              try:
    import usys as sys
except ImportError:
    import sys

try:
    import ubinascii as binascii
except:
    import binascii

            

Reported by Pylint.

              
try:
    import ubinascii as binascii
except:
    import binascii
try:
    import uhashlib as hashlib
except:
    import hashlib

            

Reported by Pylint.

                  import binascii
try:
    import uhashlib as hashlib
except:
    import hashlib

DEBUG = 0



            

Reported by Pylint.

              try:
    import usys as sys
except ImportError:
    import sys

try:
    import ubinascii as binascii
except:
    import binascii

            

Reported by Pylint.

              DEBUG = 0


def server_handshake(sock):
    clr = sock.makefile("rwb", 0)
    l = clr.readline()
    # sys.stdout.write(repr(l))

    webkey = None

            

Reported by Pylint.

              
def server_handshake(sock):
    clr = sock.makefile("rwb", 0)
    l = clr.readline()
    # sys.stdout.write(repr(l))

    webkey = None

    while 1:

            

Reported by Pylint.

                  webkey = None

    while 1:
        l = clr.readline()
        if not l:
            raise OSError("EOF in headers")
        if l == b"\r\n":
            break
        #    sys.stdout.write(l)

            

Reported by Pylint.

                      if l == b"\r\n":
            break
        #    sys.stdout.write(l)
        h, v = [x.strip() for x in l.split(b":", 1)]
        if DEBUG:
            print((h, v))
        if h == b"Sec-WebSocket-Key":
            webkey = v


            

Reported by Pylint.

                      if l == b"\r\n":
            break
        #    sys.stdout.write(l)
        h, v = [x.strip() for x in l.split(b":", 1)]
        if DEBUG:
            print((h, v))
        if h == b"Sec-WebSocket-Key":
            webkey = v


            

Reported by Pylint.

                      import asyncio
    except ImportError:
        print("SKIP")
        raise SystemExit


async def task(s, allow_cancel):
    try:
        print("task start")

            

Reported by Pylint.

              # Test cancelling a task

try:
    import uasyncio as asyncio
except ImportError:
    try:
        import asyncio
    except ImportError:
        print("SKIP")

            

Reported by Pylint.

                      raise SystemExit


async def task(s, allow_cancel):
    try:
        print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:

            

Reported by Pylint.

                      raise SystemExit


async def task(s, allow_cancel):
    try:
        print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:

            

Reported by Pylint.

                      print("task start")
        await asyncio.sleep(s)
        print("task done")
    except asyncio.CancelledError as er:
        print("task cancel")
        if allow_cancel:
            raise er



            

Reported by Pylint.

                          raise er


async def task2(allow_cancel):
    print("task 2")
    try:
        await asyncio.create_task(task(0.05, allow_cancel))
    except asyncio.CancelledError as er:
        print("task 2 cancel")

            

Reported by Pylint.

                  print("task 2")
    try:
        await asyncio.create_task(task(0.05, allow_cancel))
    except asyncio.CancelledError as er:
        print("task 2 cancel")
        raise er
    print("task 2 done")



            

Reported by Pylint.

                  print("task 2 done")


async def main():
    # Cancel task immediately
    t = asyncio.create_task(task(2, True))
    print(t.cancel())

    # Cancel task after it has started

            

Reported by Pylint.

              
async def main():
    # Cancel task immediately
    t = asyncio.create_task(task(2, True))
    print(t.cancel())

    # Cancel task after it has started
    t = asyncio.create_task(task(2, True))
    await asyncio.sleep(0.01)

            

Reported by Pylint.

                  print(t.cancel())

    # Cancel task after it has started
    t = asyncio.create_task(task(2, True))
    await asyncio.sleep(0.01)
    print(t.cancel())
    print("main sleep")
    await asyncio.sleep(0.01)


            

Reported by Pylint.

                      foo(i)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2
n_finished = 0

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

                  pass


def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1

            

Reported by Pylint.

                  pass


def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1

            

Reported by Pylint.

              

def thread_entry(n, tup):
    for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1


            

Reported by Pylint.

                  for i in tup:
        foo(i)
    with lock:
        global n_finished
        n_finished += 1


lock = _thread.allocate_lock()
n_thread = 2

            

Reported by Pylint.

              # test capability for threads to access a shared immutable data structure
#
# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd

import _thread


def foo(i):
    pass

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

              import _thread


def foo(i):
    pass


def thread_entry(n, tup):
    for i in tup:

            

Reported by Pylint.

              

class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags
        self.needs_mpy_cross = dir not in ("bare-arm", "minimal")

            

Reported by Pylint.

              
"""

import collections, sys, re, subprocess

MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:

            

Reported by Bandit.

              
"""

import collections, sys, re, subprocess

MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:

            

Reported by Pylint.

              MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags

            

Reported by Pylint.

              MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]


class PortData:
    def __init__(self, name, dir, output, make_flags=None):
        self.name = name
        self.dir = dir
        self.output = output
        self.make_flags = make_flags

            

Reported by Pylint.

              }


def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)

            

Reported by Pylint.

              
def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)
        elif a:
            a2.extend(a)

            

Reported by Pylint.

              def syscmd(*args):
    sys.stdout.flush()
    a2 = []
    for a in args:
        if isinstance(a, str):
            a2.append(a)
        elif a:
            a2.extend(a)
    subprocess.check_call(a2)

            

Reported by Pylint.

                          a2.append(a)
        elif a:
            a2.extend(a)
    subprocess.check_call(a2)


def parse_port_list(args):
    if not args:
        return list(port_data.values())

            

Reported by Bandit.

                  subprocess.check_call(a2)


def parse_port_list(args):
    if not args:
        return list(port_data.values())
    else:
        ports = []
        for arg in args:

            

Reported by Pylint.