server-methods.main() - Code Metrics - Inspection of "Allow Asynchronous Method Calls (#90)" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5b4bf6...e09c16 )

by Olivier

created 2019-10-12 15:12 UTC

server-methods.main() B

↳ Parent: server-methods

Complexity

Conditions

Size

Total Lines	66
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	42
nop	0
dl	0
loc	66
rs	8.872
c	0
b	0
f	0

How to fix Long Method

import asyncio
import logging

from asyncua import ua, uamethod, Server


# method to be exposed through server
def func(parent, variant):
    print("func method call with parameters: ", variant.Value)
    ret = False
    if variant.Value % 2 == 0:
        ret = True
    return [ua.Variant(ret, ua.VariantType.Boolean)]


# method to be exposed through server
async def func_async(parent, variant):
    if variant.Value % 2 == 0:
        print("Sleeping asynchronously for 1 second")
        await asyncio.sleep(1)
    else:
        print("Not sleeping!")


# method to be exposed through server
# uses a decorator to automatically convert to and from variants


@uamethod
def multiply(parent, x, y):
    print("multiply method call with parameters: ", x, y)
    return x * y


@uamethod
async def multiply_async(parent, x, y):
    sleep_time = x * y
    print(f"Sleeping asynchronously for {x * y} seconds")
    await asyncio.sleep(sleep_time)


async def main():
    # optional: setup logging
    logging.basicConfig(level=logging.WARN)
    # logger = logging.getLogger("asyncua.address_space")
    # logger.setLevel(logging.DEBUG)
    # logger = logging.getLogger("asyncua.internal_server")
    # logger.setLevel(logging.DEBUG)
    # logger = logging.getLogger("asyncua.binary_server_asyncio")
    # logger.setLevel(logging.DEBUG)
    # logger = logging.getLogger("asyncua.uaprocessor")
    # logger.setLevel(logging.DEBUG)
    # logger = logging.getLogger("asyncua.subscription_service")
    # logger.setLevel(logging.DEBUG)

    # now setup our server
    server = Server()
    await server.init()
    # server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
    server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
    server.set_server_name("FreeOpcUa Example Server")

    # setup our own namespace
    uri = "http://examples.freeopcua.github.io"
    idx = await server.register_namespace(uri)

    # get Objects node, this is where we should put our custom stuff
    objects = server.get_objects_node()

    # populating our address space
    await objects.add_folder(idx, "myEmptyFolder")
    myobj = await objects.add_object(idx, "MyObject")
    myvar = await myobj.add_variable(idx, "MyVariable", 6.7)
    await myvar.set_writable()  # Set MyVariable to be writable by clients
    myarrayvar = await myobj.add_variable(idx, "myarrayvar", [6.7, 7.9])
    await myobj.add_variable(
        idx, "myStronglytTypedVariable", ua.Variant([], ua.VariantType.UInt32)
    )
    await myobj.add_property(idx, "myproperty", "I am a property")
    await myobj.add_method(idx, "mymethod", func, [ua.VariantType.Int64], [ua.VariantType.Boolean])

    inargx = ua.Argument()
    inargx.Name = "x"
    inargx.DataType = ua.NodeId(ua.ObjectIds.Int64)
    inargx.ValueRank = -1
    inargx.ArrayDimensions = []
    inargx.Description = ua.LocalizedText("First number x")
    inargy = ua.Argument()
    inargy.Name = "y"
    inargy.DataType = ua.NodeId(ua.ObjectIds.Int64)
    inargy.ValueRank = -1
    inargy.ArrayDimensions = []
    inargy.Description = ua.LocalizedText("Second number y")
    outarg = ua.Argument()
    outarg.Name = "Result"
    outarg.DataType = ua.NodeId(ua.ObjectIds.Int64)
    outarg.ValueRank = -1
    outarg.ArrayDimensions = []
    outarg.Description = ua.LocalizedText("Multiplication result")

    await myobj.add_method(idx, "multiply", multiply, [inargx, inargy], [outarg])
    await myobj.add_method(idx, "multiply_async", multiply_async, [inargx, inargy], [])
    await myobj.add_method(idx, "func_async", func_async, [ua.VariantType.Int64], [])

    async with server:
        while True:
            await asyncio.sleep(1)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # loop.set_debug(True)
    loop.run_until_complete(main())
    loop.close()


1			import asyncio
2			import logging
3
4			from asyncua import ua, uamethod, Server
5
6
7			# method to be exposed through server
8			def func(parent, variant):
9			print("func method call with parameters: ", variant.Value)
10			ret = False
11			if variant.Value % 2 == 0:
12			ret = True
13			return [ua.Variant(ret, ua.VariantType.Boolean)]
14
15
16			# method to be exposed through server
17			async def func_async(parent, variant):
18			if variant.Value % 2 == 0:
19			print("Sleeping asynchronously for 1 second")
20			await asyncio.sleep(1)
21			else:
22			print("Not sleeping!")
23
24
25			# method to be exposed through server
26			# uses a decorator to automatically convert to and from variants
27
28
29			@uamethod
30			def multiply(parent, x, y):
31			print("multiply method call with parameters: ", x, y)
32			return x * y
33
34
35			@uamethod
36			async def multiply_async(parent, x, y):
37			sleep_time = x * y
38			print(f"Sleeping asynchronously for {x * y} seconds")
39			await asyncio.sleep(sleep_time)
40
41
42			async def main():
43			# optional: setup logging
44			logging.basicConfig(level=logging.WARN)
45			# logger = logging.getLogger("asyncua.address_space")
46			# logger.setLevel(logging.DEBUG)
47			# logger = logging.getLogger("asyncua.internal_server")
48			# logger.setLevel(logging.DEBUG)
49			# logger = logging.getLogger("asyncua.binary_server_asyncio")
50			# logger.setLevel(logging.DEBUG)
51			# logger = logging.getLogger("asyncua.uaprocessor")
52			# logger.setLevel(logging.DEBUG)
53			# logger = logging.getLogger("asyncua.subscription_service")
54			# logger.setLevel(logging.DEBUG)
55
56			# now setup our server
57			server = Server()
58			await server.init()
59			# server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
60			server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
61			server.set_server_name("FreeOpcUa Example Server")
62
63			# setup our own namespace
64			uri = "http://examples.freeopcua.github.io"
65			idx = await server.register_namespace(uri)
66
67			# get Objects node, this is where we should put our custom stuff
68			objects = server.get_objects_node()
69
70			# populating our address space
71			await objects.add_folder(idx, "myEmptyFolder")
72			myobj = await objects.add_object(idx, "MyObject")
73			myvar = await myobj.add_variable(idx, "MyVariable", 6.7)
74			await myvar.set_writable() # Set MyVariable to be writable by clients
75			myarrayvar = await myobj.add_variable(idx, "myarrayvar", [6.7, 7.9])
76			await myobj.add_variable(
77			idx, "myStronglytTypedVariable", ua.Variant([], ua.VariantType.UInt32)
78			)
79			await myobj.add_property(idx, "myproperty", "I am a property")
80			await myobj.add_method(idx, "mymethod", func, [ua.VariantType.Int64], [ua.VariantType.Boolean])
81
82			inargx = ua.Argument()
83			inargx.Name = "x"
84			inargx.DataType = ua.NodeId(ua.ObjectIds.Int64)
85			inargx.ValueRank = -1
86			inargx.ArrayDimensions = []
87			inargx.Description = ua.LocalizedText("First number x")
88			inargy = ua.Argument()
89			inargy.Name = "y"
90			inargy.DataType = ua.NodeId(ua.ObjectIds.Int64)
91			inargy.ValueRank = -1
92			inargy.ArrayDimensions = []
93			inargy.Description = ua.LocalizedText("Second number y")
94			outarg = ua.Argument()
95			outarg.Name = "Result"
96			outarg.DataType = ua.NodeId(ua.ObjectIds.Int64)
97			outarg.ValueRank = -1
98			outarg.ArrayDimensions = []
99			outarg.Description = ua.LocalizedText("Multiplication result")
100
101			await myobj.add_method(idx, "multiply", multiply, [inargx, inargy], [outarg])
102			await myobj.add_method(idx, "multiply_async", multiply_async, [inargx, inargy], [])
103			await myobj.add_method(idx, "func_async", func_async, [ua.VariantType.Int64], [])
104
105			async with server:
106			while True:
107			await asyncio.sleep(1)
108
109
110			if __name__ == "__main__":
111			loop = asyncio.get_event_loop()
112			# loop.set_debug(True)
113			loop.run_until_complete(main())
114			loop.close()
115

FreeOpcUa / opcua-asyncio

Push — master ( 5b4bf6...e09c16 )

server-methods.main() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like