fix: address pylint warnings

python/cloud-demo/README.md

@@ -17,7 +17,7 @@ Python 3 is required
 
 ```
 pip3 install -r requirements.txt 
-python3 cloud-demo.py \
+python3 cloud_demo.py \
   --url <url_from_cloud> \
   --api-key <api_key_from_cloud> \
   --signals Speed,SteeringWheel_Position,Accelerator_PedalPosition

python/cloud-demo/cloud-demo.py → python/cloud-demo/cloud_demo.py

@@ -7,21 +7,18 @@
 from lib import arguments
 from lib.broker import Broker
 
-"""
-Simple program designed to be used with our cloud demo.
-It is expected that you have followed the steps at
-https://demo.remotivelabs.com and started a broker + uploaded
-a recording.
+# Simple program designed to be used with our cloud demo.
+# It is expected that you have followed the steps at
+# https://demo.remotivelabs.com and started a broker + uploaded
+# a recording.
 
-Once you complete the stages in our cloud-demo you will get all
-information required to run this program. It will look something like:
+# Once you complete the stages in our cloud-demo you will get all
+# information required to run this program. It will look something like:
 
-python3 cloud-demo.py \
-     --url <broker_url> \
-     --api-key <api_key> \
-     --signals VehicleSpeed,ChassisAcceleratorPedalposition
-
-"""
+# python3 cloud_demo.py \
+#      --url <broker_url> \
+#      --api-key <api_key> \
+#      --signals VehicleSpeed,ChassisAcceleratorPedalposition
 
 expected_available_signals = [
     "VehicleSpeed",
@@ -77,5 +74,5 @@ def sleep(seconds: float) -> None:
 
 
 if __name__ == "__main__":
-    args = arguments.parse(sys.argv[1:])
+    args = arguments.parse()
     main(args)

python/cloud-demo/lib/arguments.py

@@ -3,7 +3,7 @@
 import argparse
 
 
-def parse(argv: list[str]) -> argparse.Namespace:
+def parse() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Provide address to RemotiveBroker")
 
     parser.add_argument(
@@ -38,7 +38,7 @@ def parse(argv: list[str]) -> argparse.Namespace:
         required=False,
         help="Comma separated list of signal names to subscribe on",
         default="Speed",
-        type=lambda s: [item for item in s.split(",")],
+        type=lambda s: list(s.split(",")),
     )
 
     args = parser.parse_args()

python/cloud-demo/lib/broker.py

@@ -9,6 +9,7 @@
 from typing_extensions import Self
 
 
+# pylint: disable=R0902
 class Broker:
     def __init__(self, url: str, api_key: str | None = None, access_token: str | None = None) -> None:
         self.url = url

python/playback-record/playback.py

@@ -1,8 +1,7 @@
 from __future__ import annotations
 
 import argparse
-import signal
-import sys
+import signal as sig
 import time
 from threading import Event
 from typing import Any, Optional
@@ -172,7 +171,7 @@ def exit_handler(url: str, x_api_key: str | None, access_token: str | None) -> N
     stop_playback(url, x_api_key, access_token)
 
 
-def main(argv: list[str]) -> None:
+def main() -> None:
     parser = argparse.ArgumentParser(description="Provide address to Beambroker")
     parser.add_argument(
         "-url",
@@ -217,7 +216,7 @@ def main(argv: list[str]) -> None:
 
 def run(url: str, configure: str, x_api_key: Optional[str] = None, access_token: Optional[str] = None) -> None:
     # To do a clean exit of the script on CTRL+C
-    signal.signal(signal.SIGINT, lambda signum, frame: exit_handler(url, x_api_key, access_token))
+    sig.signal(sig.SIGINT, lambda signum, frame: exit_handler(url, x_api_key, access_token))
 
     # Setting up stubs and configuration
     intercept_channel = br.create_channel(url, x_api_key, access_token)
@@ -302,4 +301,4 @@ def run(url: str, configure: str, x_api_key: Optional[str] = None, access_token:
 
 
 if __name__ == "__main__":
-    main(sys.argv[1:])
+    main()

python/pytest/test_sample.py

@@ -7,7 +7,7 @@
 _SERVER_APIKEY = None
 
 
-class Broker:
+class Broker:  # pylint: disable=R0903
     def __init__(self) -> None:
         self.channel = br.create_channel(_SERVER_URL, _SERVER_APIKEY)
         self.network_stub = br.network_api_pb2_grpc.NetworkServiceStub(self.channel)
@@ -22,15 +22,15 @@ def broker() -> Broker:
     return Broker()
 
 
-def test_check_license(broker: Broker) -> None:
+def test_check_license(local_broker: Broker) -> None:
     """Check valid license"""
-    br.check_license(broker.system_stub)
+    br.check_license(local_broker.system_stub)
 
 
-def test_server_info(broker: Broker) -> None:
+def test_server_info(local_broker: Broker) -> None:
     """Validate server information"""
 
-    conf = broker.system_stub.GetConfiguration(br.common_pb2.Empty())
+    conf = local_broker.system_stub.GetConfiguration(br.common_pb2.Empty())
 
     # Major version should be 1
     assert conf.serverVersion.startswith("v1.")
@@ -40,18 +40,18 @@ def test_server_info(broker: Broker) -> None:
     assert conf.networkInfo[0].namespace.name == "mynamespace"
 
 
-def test_list_signals(broker: Broker) -> None:
+def test_list_signals(local_broker: Broker) -> None:
     """List and valitade signals."""
 
     ns = br.common_pb2.NameSpace(name="mynamespace")
-    signals = broker.system_stub.ListSignals(ns)
+    signals = local_broker.system_stub.ListSignals(ns)
     assert len(signals.frame) == 1
 
 
-def test_meta_fields(broker: Broker) -> None:
+def test_meta_fields(local_broker: Broker) -> None:
     """Validate signal meta information."""
 
-    sc = br.SignalCreator(broker.system_stub)
+    sc = br.SignalCreator(local_broker.system_stub)
     meta_signal = sc.get_meta("mysignal1", "mynamespace")
     frame = sc.frame_by_signal("mysignal1", "mynamespace")
     assert frame.name == "myframe1"

python/reflector-ecu/reflector.py

@@ -10,7 +10,6 @@
 import grpc
 import remotivelabs.broker.sync as br
 
-signal_creator = None
 q: queue.Queue[Any] = queue.Queue()
 
 
@@ -38,13 +37,15 @@ def read_signals(stub: br.network_api_pb2_grpc.NetworkServiceStub, signal: br.co
         sys.exit()
 
 
-def ecu_a(stub: br.network_api_pb2_grpc.NetworkServiceStub, pause: int) -> None:
+def ecu_a(stub: br.network_api_pb2_grpc.NetworkServiceStub, signal_creator: br.SignalCreator, pause: int) -> None:
     """Publishes a value, read other value (published by ecu_B)
 
     Parameters
     ----------
     stub : NetworkServiceStub
         Object instance of class
+    signal_creator: SignalCreator
+        Object instance of class
     pause : int
         Amount of time to pause, in seconds
 
@@ -113,7 +114,7 @@ def read_on_timer(stub: br.network_api_pb2_grpc.NetworkServiceStub, signals: br.
 
 def get_value_pair(signal: br.network_api_pb2.Signal) -> Tuple[str, Any]:
     if signal.raw != b"":
-        raise Exception(f"not a valid signal, probably a frame {signal}")
+        raise ValueError(f"not a valid signal, probably a frame {signal}")
     if signal.HasField("integer"):
         return ("integer", signal.integer)
     if signal.HasField("double"):
@@ -123,7 +124,7 @@ def get_value_pair(signal: br.network_api_pb2.Signal) -> Tuple[str, Any]:
     if signal.HasField("empty"):
         return ("empty", signal.empty)
 
-    raise Exception(f"not a valid signal {signal}")
+    raise ValueError(f"not a valid signal {signal}")
 
 
 def act_on_signal(
@@ -134,6 +135,7 @@ def act_on_signal(
     fun: Callable[[Any], None],
     on_subcribed: Callable[[Any], None] | None = None,
 ) -> None:
+    # pylint: disable=R0913
     sub_info = br.network_api_pb2.SubscriberConfig(
         clientId=client_id,
         signals=br.network_api_pb2.SignalIds(signalId=sub_signals),
@@ -157,7 +159,7 @@ def act_on_signal(
     print("subscription terminated")
 
 
-def main(argv: list[str]) -> None:
+def main() -> None:
     parser = argparse.ArgumentParser(description="Provide address to Beambroker")
 
     parser.add_argument(
@@ -195,6 +197,7 @@ def double_and_publish(
     client_id: br.common_pb2.ClientId,
     trigger: Any,
     signals: br.network_api_pb2.Signals,
+    signal_creator: br.SignalCreator,
 ) -> None:
     if signal_creator is None:
         return
@@ -216,14 +219,14 @@ def double_and_publish(
             )
 
 
-def all_siblings(name: str, namespace_name: str) -> Sequence[br.common_pb2.SignalId]:
+def all_siblings(name: str, namespace_name: str, signal_creator: br.SignalCreator) -> Sequence[br.common_pb2.SignalId]:
     if signal_creator is None:
         return []
     frame_name = signal_creator.frame_by_signal(name, namespace_name)
     return signal_creator.signals_in_frame(frame_name.name, frame_name.namespace.name)
 
 
-def some_function_to_calculate_crc(a: Any, b: Any, c: Any) -> int:
+def some_function_to_calculate_crc(a: Any, b: Any, c: Any) -> int:  # pylint: disable=W0613
     return 1
 
 
@@ -250,7 +253,6 @@ def run(url: str, x_api_key: Optional[str] = None, access_token: Optional[str] =
     br.upload_folder(system_stub, "configuration_can")
     br.reload_configuration(system_stub)
 
-    global signal_creator
     signal_creator = br.SignalCreator(system_stub)
 
     # ecu a, we do this with lambda refering to modify_signal_publish_frame.
@@ -328,4 +330,4 @@ def modify_signals_publish_frame(
 
 
 if __name__ == "__main__":
-    main(sys.argv[1:])
+    main()