pynamics/diagnostics/os_diagnostics.py

from pybricks.parameters import Port
from uerrno import EAGAIN, EBUSY, ECANCELED, EINVAL, EIO, ENODEV, EOPNOTSUPP, EPERM, ETIMEDOUT
import uio
import ujson
import umath
import uselect
import ustruct
import usys
import urandom
from urandom import random
from pybricks.tools import wait, multitask, run_task
from pybricks.hubs import PrimeHub
from pybricks.pupdevices import Motor
import pybricks as pybricksforvers
class FakeUART:
    def __init__(self, port, baudrate, timeout):
        self.timeout = timeout
        self._force_error = None
        print("Warning: No physical UART detected. Using simulator.")

    def set_error(self, errno):
        self._force_error = errno

    def read(self, length=1):
        if self._force_error is not None:
            err = self._force_error
            self._force_error = None
            raise OSError(err)
        if self.timeout is not None:
            wait(self.timeout)
            raise OSError(ETIMEDOUT)
        else:
            while True:
                wait(1000)

    def write(self, data):
        if self._force_error is not None:
            err = self._force_error
            self._force_error = None
            raise OSError(err)


def UARTDevice(port, baudrate=9600, timeout=None):
    return FakeUART(port, baudrate, timeout)


class OSDiagnostics:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    def testUErrno(self):
        uerrnotestobject = UerrnoTest(self.hub, self.motorclass)
        uerrnotestobject.testeagain()
        uerrnotestobject.testebusy()
        uerrnotestobject.testecanceled()
        uerrnotestobject.testeinval()
        uerrnotestobject.testeio()
        uerrnotestobject.testenodev()
        uerrnotestobject.testeopnotsupp()
        uerrnotestobject.testeperm()
        uerrnotestobject.testetimedout()
        uerrnotestobject.print_results()
        self.successfultests += uerrnotestobject.successfultests
        self.failedtests.update(uerrnotestobject.failedtests)
    def testUIO(self):
        uiotestobject = UIOTest(self.hub, self.motorclass)
        uiotestobject.print_results()
        self.successfultests += uiotestobject.successfultests
        self.failedtests.update(uiotestobject.failedtests)
    def testUJSON(self):
        ujsontestobject = UJSONTest(self.hub, self.motorclass)
        ujsontestobject.print_results()
        self.successfultests += ujsontestobject.successfultests
        self.failedtests.update(ujsontestobject.failedtests)
    def testUMath(self):
        umathtestobject = UMathTest(self.hub, self.motorclass)
        umathtestobject.print_results()
        self.successfultests += umathtestobject.successfultests
        self.failedtests.update(umathtestobject.failedtests)
    def testURandom(self):
        urandtestobject = URandomTest(self.hub, self.motorclass)
        urandtestobject.print_results()
        self.successfultests += urandtestobject.successfultests
        self.failedtests.update(urandtestobject.failedtests)

    def testUSelect(self):
        uselecttestobject = USelectTest(self.hub, self.motorclass)
        uselecttestobject.print_results()
        self.successfultests += uselecttestobject.successfultests
        self.failedtests.update(uselecttestobject.failedtests)

    def testUStruct(self):
        ustructtestobject = UStructTest(self.hub, self.motorclass)
        ustructtestobject.print_results()
        self.successfultests += ustructtestobject.successfultests
        self.failedtests.update(ustructtestobject.failedtests)

    def testUSys(self):
        usystestobject = USysTest(self.hub, self.motorclass)
        usystestobject.print_results()
        self.successfultests += usystestobject.successfultests
        self.failedtests.update(usystestobject.failedtests)
    def testAll(self):
        self.testUErrno()
        self.testUIO()
        self.testUJSON()
        self.testUMath()
        self.testURandom()
        self.testUSelect()
        self.testUStruct()
        self.testUSys()
        print(f"\n=== Results: {self.successfultests}/62 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
        else:
            print("No tests failed. Great job!")
    

class UerrnoTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    def testeagain(self):
        print("Starting Test 1/9: EAGAIN - Try Again Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(EAGAIN)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 1/9: EAGAIN - FAILED")
            self.failedtests["EAGAIN"] = "No error raised"
        except OSError as ex:
            if ex.errno == EAGAIN:
                print("EAGAIN can be thrown and caught.\nCompleted Test 1/9: EAGAIN - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 1/9: EAGAIN - FAILED")
                self.failedtests["EAGAIN"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 1/9: EAGAIN - FAILED")
                self.failedtests["EAGAIN"] = ex.errno

    def testebusy(self):
        # No reliable hardware trigger; use FakeUART
        print("Starting Test 2/9: EBUSY - Device Busy Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(EBUSY)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 2/9: EBUSY - FAILED")
            self.failedtests["EBUSY"] = "No error raised"
        except OSError as ex:
            if ex.errno == EBUSY:
                print("EBUSY can be thrown and caught.\nCompleted Test 2/9: EBUSY - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 2/9: EBUSY - FAILED")
                self.failedtests["EBUSY"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 2/9: EBUSY - FAILED")
                self.failedtests["EBUSY"] = ex.errno

    def testecanceled(self):
        # No reliable hardware trigger; use FakeUART
        print("Starting Test 3/9: ECANCELED - Operation Canceled Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(ECANCELED)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 3/9: ECANCELED - FAILED")
            self.failedtests["ECANCELED"] = "No error raised"
        except OSError as ex:
            if ex.errno == ECANCELED:
                print("ECANCELED can be thrown and caught.\nCompleted Test 3/9: ECANCELED - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 3/9: ECANCELED - FAILED")
                self.failedtests["ECANCELED"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 3/9: ECANCELED - FAILED")
                self.failedtests["ECANCELED"] = ex.errno

    def testeinval(self):
        # Triggered by passing an out-of-range value to motor.control.limits()
        print("Starting Test 4/9: EINVAL - Invalid Argument Error")
        try:
            usys.stderr.flush()
        except (OSError) as ex:
            if ex.errno == EINVAL:
                print("EINVAL can be thrown and caught.\nCompleted Test 4/9: EINVAL - SUCCESSFUL")
                self.successfultests += 1
            elif errno_val == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 4/9: EINVAL - FAILED")
                self.failedtests["EINVAL"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex}.\nCompleted Test 4/9: EINVAL - FAILED")
                self.failedtests["EINVAL"] = str(ex)
    def testeio(self):
        # No reliable scriptable trigger (requires physical unplug); use FakeUART
        print("Starting Test 5/9: EIO - I/O Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(EIO)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 5/9: EIO - FAILED")
            self.failedtests["EIO"] = "No error raised"
        except OSError as ex:
            if ex.errno == EIO:
                print("EIO can be thrown and caught.\nCompleted Test 5/9: EIO - SUCCESSFUL")
                self.successfultests += 1
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 5/9: EIO - FAILED")
                self.failedtests["EIO"] = ex.errno

    def testenodev(self):
        # Triggered by initializing a motor on an empty port
        print("Starting Test 6/9: ENODEV - Device Not Found Error")
        input("Make sure Port A doesn't have anything plugged in, then press Enter.")
        try:
            my_motor = self.motorclass(Port.A)
            print("OS detected a motor when there was nothing. You may have allowed missing motors. This is useful for debugging, but not recommended for production as it can cause issues with device control.")
            self.failedtests["ENODEV"] = "No error raised"
        except OSError as ex:
            if ex.errno == ENODEV:
                print("There is no motor on this port. ENODEV can be thrown and caught.\nCompleted Test 6/9: ENODEV - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 6/9: ENODEV - FAILED")
                self.failedtests["ENODEV"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 6/9: ENODEV - FAILED")
                self.failedtests["ENODEV"] = ex.errno

    def testeopnotsupp(self):
        # No reliable scriptable trigger without specific hardware; use FakeUART
        print("Starting Test 7/9: EOPNOTSUPP - Operation Not Supported Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(EOPNOTSUPP)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 7/9: EOPNOTSUPP - FAILED")
            self.failedtests["EOPNOTSUPP"] = "No error raised"
        except OSError as ex:
            if ex.errno == EOPNOTSUPP:
                print("EOPNOTSUPP can be thrown and caught.\nCompleted Test 7/9: EOPNOTSUPP - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 7/9: EOPNOTSUPP - FAILED")
                self.failedtests["EOPNOTSUPP"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 7/9: EOPNOTSUPP - FAILED")
                self.failedtests["EOPNOTSUPP"] = ex.errno

    def testeperm(self):
        print("Starting Test 8/9: EPERM - Operation Not Permitted Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        uart.set_error(EPERM)
        try:
            uart.read(1)
            print("No error raised.\nCompleted Test 8/9: EPERM - FAILED")
            self.failedtests["EPERM"] = "No error raised"
        except OSError as ex:
            if ex.errno == EPERM:
                print("EPERM can be thrown and caught.\nCompleted Test 8/9: EPERM - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 8/9: EPERM - FAILED")
                self.failedtests["EPERM"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 8/9: EPERM - FAILED")
                self.failedtests["EPERM"] = ex.errno
    def testetimedout(self):
        # Triggered by FakeUART (or real UART) timing out on read
        print("Starting Test 9/9: ETIMEDOUT - Timed Out Error")
        uart = UARTDevice(Port.A, baudrate=9600, timeout=1000)
        try:
            data = uart.read(10)
            print("No error raised.\nCompleted Test 9/9: ETIMEDOUT - FAILED")
            self.failedtests["ETIMEDOUT"] = "No error raised"
        except OSError as ex:
            if ex.errno == ETIMEDOUT:
                print("Timed out with synthetic UART device. ETIMEDOUT can be thrown and caught.\nCompleted Test 9/9: ETIMEDOUT - SUCCESSFUL")
                self.successfultests += 1
            elif ex.errno == EIO:
                print("An unspecified error occurred (EIO).\nCompleted Test 9/9: ETIMEDOUT - FAILED")
                self.failedtests["ETIMEDOUT"] = "EIO - Unspecified Error"
            else:
                print(f"Another error occurred with code: {ex.errno}.\nCompleted Test 9/9: ETIMEDOUT - FAILED")
                self.failedtests["ETIMEDOUT"] = ex.errno

    def print_results(self):
        print(f"\n=== Results: {self.successfultests}/9 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class UIOTest(OSDiagnostics):
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    # uio contains BytesIO, StringIO, and FileIO, but due to SPIKE Prime's lack of a filesystem, the test will omit FileIO and only include BytesIO and StringIO
    def testbytesio(self):
        try:
            buffer = uio.BytesIO()

            buffer.write(b'Hello, ')
            buffer.write(b'Pybricks byte stream!')

            current_content = buffer.getvalue()
            print(f"Buffer content (via getvalue()): {current_content}") 

            print(f"Current cursor position: {buffer.tell()}")
            # TODO: After testing that BytesIO actually works on this system, add checks to make sure that the outputs match what they should. 
            buffer.seek(0)

            read_data = buffer.read()
            print(f"Read data (via read()): {read_data}")

            print(f"Current cursor position after reading: {buffer.tell()}")

            buffer.close()
            print("Buffer was closed successfully.")
            print("Completed Test 1/2: BytesIO - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("An unexpected error occured.")
            print("Completed Test 1/2: BytesIO - FAILED")
            self.failedtests["BytesIO"] = ex.errno
    def teststringio(self):
        try:
            buffer = uio.StringIO()

            buffer.write('Hello, ')
            buffer.write('Pybricks string stream!')

            current_content = buffer.getvalue()
            print(f"Buffer content (via getvalue()): {current_content}") 

            print(f"Current cursor position: {buffer.tell()}")
            # TODO: After testing that StringIO actually works on this system, add checks to make sure that the outputs match what they should. 
            buffer.seek(0)

            read_data = buffer.read()
            print(f"Read data (via read()): {read_data}")

            print(f"Current cursor position after reading: {buffer.tell()}")

            buffer.close()
            print("Buffer was closed successfully.")
            print("Completed Test 2/2: StringIO - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("An unexpected error occured.")
            print("Completed Test 2/2: StringIO - FAILED")
            self.failedtests["StringIO"] = ex.errno

    def print_results(self):
        self.testbytesio()
        self.teststringio()
        print(f"\n=== Results: {self.successfultests}/2 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class UJSONTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}

    # Tests ujson.dumps() and ujson.loads()
    def testdumpsloads(self):
        try:
            original = {"robot": "Pybricks", "speed": 500, "active": True}

            json_str = ujson.dumps(original)
            print(f"Serialized (via dumps()): {json_str}")

            restored = ujson.loads(json_str)
            print(f"Deserialized (via loads()): {restored}")

            # TODO: After confirming dumps/loads works on this system, add equality tests

            print("Completed Test 1/3: dumps/loads - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("An unexpected error occurred.")
            print("Completed Test 1/3: dumps/loads - FAILED")
            self.failedtests["dumps/loads"] = getattr(ex, "errno", str(ex))

    # Tests ujson.loads() raising ValueError on malformed input
    def testloadsinvalid(self):
        try:
            ujson.loads("{not valid json}")
            print("No error raised.")
            print("Completed Test 2/3: loads invalid - FAILED")
            self.failedtests["loads_invalid"] = "No error raised"
        except ValueError:
            print("ValueError raised on malformed JSON, as expected.")
            print("Completed Test 2/3: loads invalid - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("An unexpected error occurred.")
            print("Completed Test 2/3: loads invalid - FAILED")
            self.failedtests["loads_invalid"] = getattr(ex, "errno", str(ex))

    # Tests ujson.dump() and ujson.load() using a uio.StringIO stream
    def testdumpload(self):
        try:
            original = {"hub": "SPIKE Prime", "port": "A", "value": 42}

            stream = uio.StringIO()
            ujson.dump(original, stream)
            print(f"Serialized to stream (via dump()): {stream.getvalue()}")

            stream.seek(0)
            restored = ujson.load(stream)
            print(f"Deserialized from stream (via load()): {restored}")

            stream.close()

            # TODO: After confirming dump/load works on this system, add
            # equality assertions to verify round-trip fidelity.

            print("Completed Test 3/3: dump/load (stream) - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("An unexpected error occurred.")
            print("Completed Test 3/3: dump/load (stream) - FAILED")
            self.failedtests["dump/load"] = getattr(ex, "errno", str(ex))

    def print_results(self):
        self.testdumpsloads()
        self.testloadsinvalid()
        self.testdumpload()
        print(f"\n=== Results: {self.successfultests}/3 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class UMathTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    def test_math(self):
        EPSILON = 0.0001
        if(umath.ceil(87.21) == 88):
            self.successfultests += 1
        else:
            self.failedtests["ceilpos"] = "Failed"

        if(umath.floor(14.61) == 14):
            self.successfultests += 1
        else:
            self.failedtests["floorpos"] = "Failed"

        if(umath.ceil(-87.21) == -87):
            self.successfultests += 1
        else:
            self.failedtests["ceilneg"] = "Failed"
            
        if(umath.floor(-14.61) == -15):
            self.successfultests += 1
        else:
            self.failedtests["floorneg"] = "Failed"
            
        if(umath.trunc(33.22) == 33):
            self.successfultests += 1
        else:
            self.failedtests["truncpos"] = "Failed"

        if(umath.trunc(-33.22) == -33):
            self.successfultests += 1
        else:
            self.failedtests["truncneg"] = "Failed"

        if(umath.fmod(6040, 3) == 1):
            self.successfultests += 1
        else:
            self.failedtests["fmod"] = "Failed"
            
        if(umath.fabs(88273) == 88273):
            self.successfultests += 1
        else:
            self.failedtests["fabspos"] = "Failed"
        
        if(umath.fabs(-27482) == 27482):
            self.successfultests += 1
        else:
            self.failedtests["fabsneg"] = "Failed"
        
        if(umath.fabs(-2742.233) == 2742.233):
            self.successfultests += 1
        else:
            self.failedtests["fabsflt"] = "Failed"

        if(umath.copysign(3928, -182) == -3928):
            self.successfultests += 1
        else:
            self.failedtests["copysign"] = "Failed"

        if(umath.exp(umath.log(1)) == 1):
            self.successfultests += 2
        else:
            self.failedtests["eexp"] = "Failed"
            self.failedtests["ln"] = "Failed"
            
        if(abs(umath.e - 2.718282) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["e"] = "Failed"

        if(umath.pow(19, 7) == 893871739):
            self.successfultests += 1
        else:
            self.failedtests["pow"] = "Failed"

        if(umath.sqrt(242064) == 492):
            self.successfultests += 1
        else:
            self.failedtests["sqrt"] = "Failed"

        if(abs(umath.pi - 3.141593) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["pi"] = "Failed"


        if abs(umath.degrees(umath.pi * 3) - 540) < EPSILON:
            self.successfultests += 1
        else:
            self.failedtests["degrees"] = "Failed"

        if abs(umath.radians(270) - umath.pi * 1.5) < EPSILON:
            self.successfultests += 1
        else:
            self.failedtests["radians"] = "Failed"

        if(abs(umath.sin(umath.pi)) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["sin"] = "Failed"
            
        if(abs(umath.asin(1) - umath.pi * 0.5) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["asin"] = "Failed"

        if(abs(umath.cos(umath.pi) + 1) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["cos"] = "Failed"
            
        if(abs(umath.acos(1)) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["acos"] = "Failed"

        if(abs(umath.tan(umath.pi)) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["tan"] = "Failed"
            
        if(abs(umath.atan(1) - umath.pi * 0.25) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["atan"] = "Failed"

        if(abs(umath.atan2(1, -1) - umath.pi * 0.75) < EPSILON):
            self.successfultests += 1
        else:
            self.failedtests["atan2"] = "Failed"

        infinitenum = float('inf')
        finitenum = 123456
        if(umath.isfinite(finitenum) == True):
            self.successfultests += 1
        else:
            self.failedtests["isfinitefinite"] = "Failed"

        if(umath.isfinite(infinitenum) == False):
            self.successfultests += 1
        else:
            self.failedtests["isfiniteinfinite"] = "Failed"

        if(umath.isinf(finitenum) == False):
            self.successfultests += 1
        else:
            self.failedtests["isinfinitefinite"] = "Failed"

        if(umath.isinf(infinitenum) == True):
            self.successfultests += 1
        else:
            self.failedtests["isinfiniteinfinite"] = "Failed"
        nannum = float("nan")

        if(umath.isnan(nannum) == True):
            self.successfultests += 1
        else:
            self.failedtests["isnannan"] = "Failed"

        if(umath.isnan(finitenum) == False):
            self.successfultests += 1
        else:
            self.failedtests["isnannotnan"] = "Failed"

        frac, integer = umath.modf(87.21)
        if abs(integer - 87.0) < 0.01 and abs(frac - 0.21) < 0.01:
            self.successfultests += 1
            
        else:
            self.failedtests["modf"] = "Failed"

        mantissa, exponent = umath.frexp(64)
        if mantissa == 0.5 and exponent == 7:
            self.successfultests += 1
        else:
            self.failedtests["frexp"] = "Failed"

        result = umath.ldexp(0.5, 7)
        if result == 64:
            self.successfultests += 1
        else:
            self.failedtests["ldexp"] = "Failed"   
    def print_results(self):
        self.test_math()
        print(f"\n=== Results: {self.successfultests}/35 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class URandomTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    def testrandom(self):
        NUM_SAMPLES = 6553
        NUM_BUCKETS = 10
        CHART_WIDTH = 50
        SAMPLE_PEEK = 20

        samples = [random() for _ in range(NUM_SAMPLES)]

        bucket_counts = [0] * NUM_BUCKETS
        for x in samples:
            i = int(x * NUM_BUCKETS)
            if i == NUM_BUCKETS:
                i -= 1
            bucket_counts[i] += 1

        mean = sum(samples) / NUM_SAMPLES
        ideal = NUM_SAMPLES / NUM_BUCKETS
        max_count = max(bucket_counts)

        print("=" * 66)
        print("  random() distribution test   n=" + str(NUM_SAMPLES))
        print("=" * 66)
        print("")
        print("  Range          Count  Bar")
        print("  -------------- -----  " + "-" * CHART_WIDTH)

        for i in range(NUM_BUCKETS):
            lo = i / NUM_BUCKETS
            hi = (i + 1) / NUM_BUCKETS
            count = bucket_counts[i]
            bar = int(round(count / max_count * CHART_WIDTH))
            dev = abs(count - ideal) / ideal
            if dev <= 0.10:
                marker = "#"
            elif dev <= 0.20:
                marker = "+"
            else:
                marker = "."
            lo_str = "{:.2f}".format(lo)
            hi_str = "{:.2f}".format(hi)
            print("  " + "{:.2f}".format(lo) + " - " + "{:.2f}".format(hi) + "  " + str(count) + "  " + marker * bar)

        print("")
        print("  Samples : " + str(NUM_SAMPLES))
        print("  Mean    : " + "{:.6f}".format(mean) + "  (ideal 0.500000)")
        print("  Min     : " + "{:.6f}".format(min(samples)))
        print("  Max     : " + "{:.6f}".format(max(samples)))
        print("  Legend  : # within 10%   + within 20%   . beyond 20%")
        print("")
        print("  First " + str(SAMPLE_PEEK) + " raw values:")

        row = "  "
        for idx, val in enumerate(samples[:SAMPLE_PEEK], 1):
            row += "{:.4f}  ".format(val)
            if idx % 5 == 0:
                print(row)
                row = "  "
        if row.strip():
            print(row)

        print("")
        if(abs(0.5 - mean) < 0.06):
            print("Random Distribution Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Random Distribution Test: FAILED")
            self.failedtests["randomdistribution"] = "Too much error"
    def test_other_rands(self):
        N = 1000
        choicelist = ["apple", "banana", "grape", "orange"]

        randint_vals = [urandom.randint(1, 100) for _ in range(N)]
        randint_mean = sum(randint_vals) / N
        print("Randint mean (ideally 50.5): {:.4f}".format(randint_mean))
        if abs(50.5 - randint_mean) < 3.0:
            print("Randint Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Randint Test: FAILED")
            self.failedtests["randint"] = "Too much error"

        getrandbits_vals = [urandom.getrandbits(7) for _ in range(N)]
        getrandbits_mean = sum(getrandbits_vals) / N
        print("Getrandbits(7) mean (ideally 63.5): {:.4f}".format(getrandbits_mean))
        if abs(63.5 - getrandbits_mean) < 4:
            print("Getrandbits Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Getrandbits Test: FAILED")
            self.failedtests["getrandbits"] = "Too much error"

        randrange_vals = [urandom.randrange(1, 1001, 4) for _ in range(N)]
        randrange_mean = sum(randrange_vals) / N
        print("Randrange(1,1001,4) mean (ideally 501.0): {:.4f}".format(randrange_mean))
        if abs(501.0 - randrange_mean) < 20.0:
            print("Randrange Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Randrange Test: FAILED")
            self.failedtests["randrange"] = "Too much error"

        uniform_vals = [urandom.uniform(1, 10) for _ in range(N)]
        uniform_mean = sum(uniform_vals) / N
        print("Uniform(1,10) mean (ideally 5.5): {:.4f}".format(uniform_mean))
        if abs(5.5 - uniform_mean) < 0.3:
            print("Uniform Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Uniform Test: FAILED")
            self.failedtests["uniform"] = "Too much error"

        choice_counts = {}
        for item in choicelist:
            choice_counts[item] = 0
        for _ in range(N):
            choice_counts[urandom.choice(choicelist)] += 1
        ideal_pct = 100.0 / len(choicelist)
        print("Choice distribution (ideally {:.1f}% each):".format(ideal_pct))
        choice_ok = True
        for item in choicelist:
            pct = choice_counts[item] / N * 100
            print("  " + item + ": {:.1f}%".format(pct))
            if abs(pct - ideal_pct) > 5.0:
                choice_ok = False
        if choice_ok:
            print("Choice Test: SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Choice Test: FAILED")
            self.failedtests["choice"] = "Too much error"
    def print_results(self):
        self.testrandom()
        self.test_other_rands()
        print(f"\n=== Results: {self.successfultests}/6 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class USelectTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}

    def print_results(self):
        # Register the standard input so we can read keyboard presses.
        keyboard = uselect.poll()
        keyboard.register(usys.stdin)
        print("Type a few keys, make sure you get back what character you typed, then press Escape.")
        while True:
            # Check if a key has been pressed.
            if keyboard.poll(0):

                # Read the key and print it.
                key = usys.stdin.read(1)
                if key == '\x1b':
                    print("Escape key pressed. Exiting...")
                    break
                else:
                    print("Pressed:", key)
        result = input("Input Y if the results were accurate:")
        if(result == "Y" or result == "y"):
            print(f"\n=== Results: 1/1 tests passed ===")
            self.successfultests += 1
        else:
            self.failedtests["input"] = "Unsatisfied"
            if self.failedtests:
                print("Failed tests:")
                for key, value in self.failedtests.items():
                    print(f"  {key}: {value}")
class UStructTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}

    def print_results(self):
        packed = ustruct.pack('ii', 42, 100)
        print(f'Packed bytes using pack(): {packed}')

        unpacked = ustruct.unpack('ii', packed)
        print(f'Unpacked values using unpack(): {unpacked}')
        print(unpacked == (42, 100))
        if(unpacked == (42, 100)):
            print("Completed Test 1/2: pack - SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Completed Test 1/2: pack - FAILED")
            self.failedtests["pack"] = "Failed"
        format_string = 'hhl'
        size = ustruct.calcsize(format_string)

        buffer = bytearray(size)

        ustruct.pack_into(format_string, buffer, 0, 5, 10, 15)

        print("Packed buffer using pack_into():", buffer)

        unpackedfrom = ustruct.unpack_from(format_string, buffer, 0)
        print("Unpacked buffer using unpack_from():", unpackedfrom)
        if(unpackedfrom == (5, 10, 15)):
            print("Completed Test 2/2: pack_into - SUCCESSFUL")
            self.successfultests += 1
        else:
            print("Completed Test 2/2: pack_into - FAILED")
            self.failedtests["pack_into"] = "Failed"
        print(f"\n=== Results: {self.successfultests}/2 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
class USysTest:
    def __init__(self, hub, motorclass):
        self.hub = hub
        self.motorclass = motorclass
        self.successfultests = 0
        self.failedtests = {}
    def printVersionDiagnostics(self):
        try:
            print("Hub version information:", pybricksforvers.version)
            print("MicroPython version:", usys.version)
            print("Pybricks version information:", usys.version_info)
            print("MicroPython information:", usys.implementation)
            self.successfultests += 1
            print("Completed Test 1/4: versioninfo - SUCCESSFUL")
        except Exception as ex:
            self.failedtests["versioninfo"] = ex.errno
            print("Completed Test 4/4: versioninfo - FAILED")
    def teststdin(self):
        # Register the standard input so we can read keyboard presses.
        keyboard = uselect.poll()
        keyboard.register(usys.stdin)
        print("Type a few keys, make sure you get back what character you typed, then press Escape.")
        while True:
            # Check if a key has been pressed.
            if keyboard.poll(0):

                # Read the key and print it.
                key = usys.stdin.read(1)
                if key == '\x1b':
                    print("Escape key pressed. Exiting...")
                    break
                else:
                    print("Pressed:", key)
        result = input("Input Y if the results were accurate:")
        if(result == "Y" or result == "y"):
            print("Completed Test 1/4: stdin - SUCCESSFUL")
            self.successfultests += 1
        else:
            self.failedtests["stdin"] = "Unsatisfied"
            print("Completed Test 1/4: stdin - FAILED")
            if self.failedtests:
                print("Failed tests:")
                for key, value in self.failedtests.items():
                    print(f"  {key}: {value}")
    def teststdout(self):
        usys.stdout.flush()
        try:
            usys.stdout.buffer.write(b"stdout worked!\n")
            print("Completed Test 2/4: stdout - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("Completed Test 2/4: stdout - FAILED")
            self.failedtests["stdout"] = ex.errno
    def teststderr(self):
        usys.stdout.flush()
        try:       
            usys.stderr.buffer.write(b"stderr worked!\n")
            print("Completed Test 3/4: stderr - SUCCESSFUL")
            self.successfultests += 1
        except Exception as ex:
            print("Completed Test 3/4: stderr - FAILED")
            self.failedtests["stderr"] = ex.errno

    def print_results(self):
        self.teststdin()
        self.teststdout()
        self.teststderr()
        self.printVersionDiagnostics()
        print(f"\n=== Results: {self.successfultests}/4 tests passed ===")
        if self.failedtests:
            print("Failed tests:")
            for key, value in self.failedtests.items():
                print(f"  {key}: {value}")
diag = OSDiagnostics(hub=PrimeHub(), motorclass=Motor)
diag.testAll()