From 74fb46b4ab1da7bf8ff985bedcf831098afa7f41 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 29 Aug 2026 08:22:57 +0200 Subject: [PATCH] PSD: zero findings, and four modules that no longer run on import 152 ruff findings to 0 across twelve files, and mypy --strict is clean on all of them but one -- see the end. The course is marked unrunnable because it wants Kafka and Flink, so the verification is what can be run without them. Ten of the twelve modules now import cleanly against a venv with confluent-kafka, kafka-python and flask; the two pyflink ones are skipped. That check did not exist before, because four of these files WERE their own side effect: - `kafka/generate_and_send.py` and `kafka/kafka_consumer.py` ran their `while True:` loop at module level. Importing either never returns; I gave up after twelve seconds of rdkafka connection failures. - `zin1/producent.py` built a KafkaProducer at module level and raised NoBrokersAvailable on import. - `flink/anomaly_detector.py` called `env.execute()` at module level. Each is a `main()` behind a `__main__` guard now, so importing one is free and running it is unchanged. `webinterface.py` called `json.loads` and never imported json, so the first message on the topic would have raised NameError. mypy then found three more of the same shape: confluent-kafka's `msg.error()` and `msg.value()` both return Optional, and the code dereferenced them straight away. They are narrowed to locals now, which also stops `msg.error()` being called twice per message. A malformed alarm with no timestamp is skipped with a warning instead of dividing None by 1000. Two SLF001 findings were reads of `KafkaError._PARTITION_EOF`. That leading underscore is confluent-kafka's own naming and there is no public alias, but `KafkaError.name()` is public and returns exactly that string -- checked against the installed library, where code() is -191 and name() is "_PARTITION_EOF". Flask's `debug=True` is now opt-in via FLASK_DEBUG, because the debug console executes arbitrary code. The ten `random` calls are simulation, not crypto, so S311 is not a real vulnerability -- but `secrets.SystemRandom()` is a drop-in with the same methods and no rule to argue with, so that is what they use. Behaviour is checked, not assumed: the sensor still returns a location in 1..10 and a temperature in -20..40, the thresholds still classify -15 as too cold, 40 as too hot and 20 as normal, transactions still carry a four-digit card id, and TemperatureReading survives a to_dict/from_dict round trip. **Two mypy errors remain and are deliberate.** `--strict` implies `disallow_subclassing_any`, and the Flink job must subclass pyflink's MapFunction and KeyedProcessFunction, which ship no types. Fixing it means weakening strict globally or writing an override for our own module. Neither is worth it, so the two are reported in TODO-refactor-remaining.md rather than hidden. Co-Authored-By: Claude Opus 5 --- Programming/PSD/flink/anomaly_detector.py | 44 +++++--- Programming/PSD/kafka/generate_and_send.py | 71 ++++++++----- Programming/PSD/kafka/kafka_consumer.py | 66 ++++++++---- Programming/PSD/webinterface/webinterface.py | 39 +++++-- Programming/PSD/zin1/konsument.py | 23 +++- Programming/PSD/zin1/producent.py | 39 +++++-- .../PSD/zin1/simulate_temperature_sensor.py | 23 +++- .../code/generator/temperature_generator.py | 47 +++++--- .../python/code/model/temperature_alarm.py | 24 ++++- .../python/code/model/temperature_reading.py | 24 ++++- .../processor/temperature_anomaly_detector.py | 100 +++++++++++------- .../code/visualizer/alarm_visualizer.py | 65 +++++++++--- TODO-refactor-remaining.md | 19 +++- pyproject.toml | 2 + 14 files changed, 424 insertions(+), 162 deletions(-) mode change 100644 => 100755 Programming/PSD/flink/anomaly_detector.py mode change 100644 => 100755 Programming/PSD/kafka/generate_and_send.py mode change 100644 => 100755 Programming/PSD/kafka/kafka_consumer.py mode change 100644 => 100755 Programming/PSD/webinterface/webinterface.py mode change 100644 => 100755 Programming/PSD/zin1/konsument.py mode change 100644 => 100755 Programming/PSD/zin1/producent.py mode change 100644 => 100755 Programming/PSD/zin1/simulate_temperature_sensor.py mode change 100644 => 100755 Programming/PSD/zin3/python/code/generator/temperature_generator.py mode change 100644 => 100755 Programming/PSD/zin3/python/code/model/temperature_alarm.py mode change 100644 => 100755 Programming/PSD/zin3/python/code/model/temperature_reading.py mode change 100644 => 100755 Programming/PSD/zin3/python/code/processor/temperature_anomaly_detector.py mode change 100644 => 100755 Programming/PSD/zin3/python/code/visualizer/alarm_visualizer.py diff --git a/Programming/PSD/flink/anomaly_detector.py b/Programming/PSD/flink/anomaly_detector.py old mode 100644 new mode 100755 index 6f3ad9a5..09a664b1 --- a/Programming/PSD/flink/anomaly_detector.py +++ b/Programming/PSD/flink/anomaly_detector.py @@ -1,23 +1,41 @@ +#!/usr/bin/env python3 +"""Flink job skeleton: read `transactions` and print each one back out. + +The detection itself was never written; the job exists to prove the Kafka +source and the pyflink environment are wired up. +""" + +from __future__ import annotations + import json +from typing import Any, cast from pyflink.common.serialization import SimpleStringSchema from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer -def detect_anomalies(transaction): - transaction_data = json.loads(transaction) - # Add anomaly detection logic here - return transaction_data +def detect_anomalies(transaction: str) -> dict[str, Any]: + """Parse one transaction. Anomaly detection would go here.""" + return cast("dict[str, Any]", json.loads(transaction)) -env = StreamExecutionEnvironment.get_execution_environment() -kafka_consumer = FlinkKafkaConsumer( - topics="transactions", - deserialization_schema=SimpleStringSchema(), - properties={"bootstrap.servers": "localhost:9092", "group.id": "anomaly_detection"}, -) +def main() -> None: + """Wire the Kafka source to the parser and run the job.""" + env = StreamExecutionEnvironment.get_execution_environment() + kafka_consumer = FlinkKafkaConsumer( + topics="transactions", + deserialization_schema=SimpleStringSchema(), + properties={ + "bootstrap.servers": "localhost:9092", + "group.id": "anomaly_detection", + }, + ) -data_stream = env.add_source(kafka_consumer).map(detect_anomalies) -data_stream.print() -env.execute("Anomaly Detection") + data_stream = env.add_source(kafka_consumer).map(detect_anomalies) + data_stream.print() + env.execute("Anomaly Detection") + + +if __name__ == "__main__": + main() diff --git a/Programming/PSD/kafka/generate_and_send.py b/Programming/PSD/kafka/generate_and_send.py old mode 100644 new mode 100755 index b32901b1..cbcc2255 --- a/Programming/PSD/kafka/generate_and_send.py +++ b/Programming/PSD/kafka/generate_and_send.py @@ -1,41 +1,62 @@ +#!/usr/bin/env python3 +"""Push made-up card transactions to the `transactions` topic once a second.""" + +from __future__ import annotations + import json -import random +import logging import time +from secrets import SystemRandom +from typing import Any -from confluent_kafka import Producer +from confluent_kafka import Message, Producer -# Configuration for Kafka Producer -conf = {"bootstrap.servers": "localhost:9092"} -producer = Producer(**conf) +logger = logging.getLogger(__name__) + +_random = SystemRandom() -def generate_transaction(): +def generate_transaction() -> dict[str, Any]: + """Return one transaction with a random card, place and amount.""" return { - "card_id": random.randint(1000, 9999), - "user_id": random.randint(100, 999), + "card_id": _random.randint(1000, 9999), + "user_id": _random.randint(100, 999), "location": { - "latitude": round(random.uniform(-90, 90), 6), - "longitude": round(random.uniform(-180, 180), 6), + "latitude": round(_random.uniform(-90, 90), 6), + "longitude": round(_random.uniform(-180, 180), 6), }, - "transaction_value": round(random.uniform(1, 1000), 2), - "spending_limit": round(random.uniform(1000, 5000), 2), + "transaction_value": round(_random.uniform(1, 1000), 2), + "spending_limit": round(_random.uniform(1000, 5000), 2), } -def delivery_report(err, msg): +def delivery_report(err: object, msg: Message) -> None: + """Log whether one message made it to the broker.""" if err is not None: - print(f"Message delivery failed: {err}") + logger.error("Message delivery failed: %s", err) else: - print(f"Message delivered to {msg.topic()} [{msg.partition()}]") + logger.info("Message delivered to %s [%s]", msg.topic(), msg.partition()) -while True: - transaction = generate_transaction() - producer.produce( - "transactions", - key=str(transaction["card_id"]), - value=json.dumps(transaction), - callback=delivery_report, - ) - producer.poll(1) - time.sleep(1) # Adjust the sleep time to control the frequency of transactions +def main() -> None: + """Produce transactions until interrupted.""" + # Configuration for Kafka Producer + conf = {"bootstrap.servers": "localhost:9092"} + producer = Producer(**conf) + + while True: + transaction = generate_transaction() + producer.produce( + "transactions", + key=str(transaction["card_id"]), + value=json.dumps(transaction), + callback=delivery_report, + ) + producer.poll(1) + # Adjust the sleep time to control the frequency of transactions + time.sleep(1) + + +if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) + main() diff --git a/Programming/PSD/kafka/kafka_consumer.py b/Programming/PSD/kafka/kafka_consumer.py old mode 100644 new mode 100755 index 5c25bf81..f422703e --- a/Programming/PSD/kafka/kafka_consumer.py +++ b/Programming/PSD/kafka/kafka_consumer.py @@ -1,23 +1,49 @@ -from confluent_kafka import Consumer, KafkaError +#!/usr/bin/env python3 +"""Print everything on the `transactions` topic until the partition ends.""" -conf = { - "bootstrap.servers": "localhost:9092", - "group.id": "test_group", - "auto.offset.reset": "earliest", -} +from __future__ import annotations -consumer = Consumer(**conf) -consumer.subscribe(["transactions"]) +import logging -while True: - msg = consumer.poll(timeout=1.0) - if msg is None: - continue - if msg.error(): - if msg.error().code() == KafkaError._PARTITION_EOF: - continue - else: - print(msg.error()) - break - print("Received message: {}".format(msg.value().decode("utf-8"))) -consumer.close() +from confluent_kafka import Consumer + +logger = logging.getLogger(__name__) + +# confluent-kafka names this error _PARTITION_EOF, leading underscore and all. +# `KafkaError.name()` is the public way to ask for it without reaching into a +# private attribute; it returns exactly this string. +PARTITION_EOF = "_PARTITION_EOF" + + +def main() -> None: + """Consume until a broker error other than end-of-partition arrives.""" + conf = { + "bootstrap.servers": "localhost:9092", + "group.id": "test_group", + "auto.offset.reset": "earliest", + } + + consumer = Consumer(**conf) + consumer.subscribe(["transactions"]) + + try: + while True: + msg = consumer.poll(timeout=1.0) + if msg is None: + continue + error = msg.error() + if error is not None: + if error.name() == PARTITION_EOF: + continue + logger.error("%s", error) + break + value = msg.value() + if value is not None: + logger.info("Received message: %s", value.decode("utf-8")) + finally: + consumer.close() + + +if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) + main() diff --git a/Programming/PSD/webinterface/webinterface.py b/Programming/PSD/webinterface/webinterface.py old mode 100644 new mode 100755 index 15babc54..72f6f1ef --- a/Programming/PSD/webinterface/webinterface.py +++ b/Programming/PSD/webinterface/webinterface.py @@ -1,11 +1,31 @@ -from confluent_kafka import Consumer, KafkaError +#!/usr/bin/env python3 +"""A one-route Flask app that drains the `alarms` topic and returns it as JSON.""" + +from __future__ import annotations + +import json +import os +from typing import TYPE_CHECKING + +from confluent_kafka import Consumer from flask import Flask, jsonify +if TYPE_CHECKING: + from flask.typing import ResponseReturnValue + app = Flask(__name__) +# confluent-kafka names this error _PARTITION_EOF, leading underscore and all. +# `KafkaError.name()` is the public way to ask for it without reaching into a +# private attribute; it returns exactly this string. +PARTITION_EOF = "_PARTITION_EOF" + +HTTP_INTERNAL_SERVER_ERROR = 500 + @app.route("/alarms", methods=["GET"]) -def get_alarms(): +def get_alarms() -> ResponseReturnValue: + """Return every alarm currently on the topic, oldest first.""" alarms = [] conf = { "bootstrap.servers": "localhost:9092", @@ -20,15 +40,20 @@ def get_alarms(): msg = consumer.poll(timeout=1.0) if msg is None: break - if msg.error(): - if msg.error().code() == KafkaError._PARTITION_EOF: + error = msg.error() + if error is not None: + if error.name() == PARTITION_EOF: continue - return str(msg.error()), 500 - alarms.append(json.loads(msg.value().decode("utf-8"))) + return str(error), HTTP_INTERNAL_SERVER_ERROR + value = msg.value() + if value is not None: + alarms.append(json.loads(value.decode("utf-8"))) consumer.close() return jsonify(alarms) if __name__ == "__main__": - app.run(debug=True) + # Debug mode serves a console that executes arbitrary code, so it is opt-in + # rather than the default it used to be here. + app.run(debug=os.environ.get("FLASK_DEBUG") == "1") diff --git a/Programming/PSD/zin1/konsument.py b/Programming/PSD/zin1/konsument.py old mode 100644 new mode 100755 index b199289d..319c1310 --- a/Programming/PSD/zin1/konsument.py +++ b/Programming/PSD/zin1/konsument.py @@ -1,13 +1,23 @@ +#!/usr/bin/env python3 +"""Read the temperature_readings topic and shout about the extremes.""" + +from __future__ import annotations + import json +import logging +from typing import Any from kafka import KafkaConsumer +logger = logging.getLogger(__name__) + # Thresholds TEMP_TOO_COLD = -10 TEMP_TOO_HOT = 35 -def process_temperature_reading(reading): +def process_temperature_reading(reading: dict[str, Any]) -> str: + """Return the line to print for one reading.""" temperature = reading["temperature"] if temperature < TEMP_TOO_COLD: alert = f"WARNING: Temperature is too cold! ({temperature}°C)" @@ -18,7 +28,8 @@ def process_temperature_reading(reading): return alert -if __name__ == "__main__": +def main() -> None: + """Consume readings until interrupted.""" consumer = KafkaConsumer( "temperature_readings", bootstrap_servers="localhost:9092", @@ -27,5 +38,9 @@ if __name__ == "__main__": for message in consumer: reading = json.loads(message.value) - alert_message = process_temperature_reading(reading) - print(alert_message) + logger.info("%s", process_temperature_reading(reading)) + + +if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) + main() diff --git a/Programming/PSD/zin1/producent.py b/Programming/PSD/zin1/producent.py old mode 100644 new mode 100755 index 86cd6579..cec4c5a9 --- a/Programming/PSD/zin1/producent.py +++ b/Programming/PSD/zin1/producent.py @@ -1,22 +1,43 @@ +#!/usr/bin/env python3 +"""Send a simulated reading to the temperature_readings topic, now and then.""" + +from __future__ import annotations + import json -import random +import logging import time +from secrets import SystemRandom +from typing import Any from kafka import KafkaProducer from simulate_temperature_sensor import generate_temperature_reading +logger = logging.getLogger(__name__) -def serializer(message): +MIN_INTERVAL_SECONDS = 1 +MAX_INTERVAL_SECONDS = 5 + +_random = SystemRandom() + + +def serializer(message: dict[str, Any]) -> bytes: + """Encode one message as UTF-8 JSON for Kafka.""" return json.dumps(message).encode("utf-8") -producer = KafkaProducer( - bootstrap_servers=["localhost:9092"], value_serializer=serializer -) - -if __name__ == "__main__": +def main() -> None: + """Produce readings until interrupted.""" + producer = KafkaProducer( + bootstrap_servers=["localhost:9092"], value_serializer=serializer + ) while True: reading = generate_temperature_reading() - print(f"Sending reading: {reading}") + logger.info("Sending reading: %s", reading) producer.send("temperature_readings", reading) - time.sleep(random.randint(1, 5)) # Simulate readings sent at random intervals + # Simulate readings sent at random intervals + time.sleep(_random.randint(MIN_INTERVAL_SECONDS, MAX_INTERVAL_SECONDS)) + + +if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) + main() diff --git a/Programming/PSD/zin1/simulate_temperature_sensor.py b/Programming/PSD/zin1/simulate_temperature_sensor.py old mode 100644 new mode 100755 index 1f24f73d..b34593dd --- a/Programming/PSD/zin1/simulate_temperature_sensor.py +++ b/Programming/PSD/zin1/simulate_temperature_sensor.py @@ -1,11 +1,24 @@ -import random +#!/usr/bin/env python3 +"""Stand-in for the sensors: one reading from one of ten made-up locations.""" + +from __future__ import annotations + import time +from secrets import SystemRandom +from typing import Any + +LOCATION_COUNT = 10 +MIN_TEMPERATURE = -20 +MAX_TEMPERATURE = 40 + +_random = SystemRandom() -def generate_temperature_reading(): - location_id = random.randint(1, 10) # Simulate 10 different locations - temperature = random.uniform( - -20, 40 +def generate_temperature_reading() -> dict[str, Any]: + """Return one reading: a location, a temperature and a Unix timestamp.""" + location_id = _random.randint(1, LOCATION_COUNT) # Simulate 10 different locations + temperature = _random.uniform( + MIN_TEMPERATURE, MAX_TEMPERATURE ) # Temperature range from -20 to 40 degrees Celsius timestamp = time.time() # Current Unix timestamp return { diff --git a/Programming/PSD/zin3/python/code/generator/temperature_generator.py b/Programming/PSD/zin3/python/code/generator/temperature_generator.py old mode 100644 new mode 100755 index e3eee2d9..be9a3eed --- a/Programming/PSD/zin3/python/code/generator/temperature_generator.py +++ b/Programming/PSD/zin3/python/code/generator/temperature_generator.py @@ -1,37 +1,53 @@ +#!/usr/bin/env python3 +"""Publish a simulated reading from each of five thermometers, once a second. + +The readings go to the Temperatura topic, where the Flink job in +`../processor/` picks them up and republishes the below-zero ones as alarms. +""" + import json -import os -import random +import logging import sys import time +from pathlib import Path +from secrets import SystemRandom -from confluent_kafka import Producer +from confluent_kafka import Message, Producer # Add parent directory to path to import model -sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.append(str(Path(__file__).resolve().parent.parent)) from model.temperature_reading import TemperatureReading +logger = logging.getLogger(__name__) -def delivery_report(err, msg): - """Callback for message delivery reports.""" +# Readings are drawn from this range, so roughly a quarter of them alarm. +MIN_TEMPERATURE = -10 +MAX_TEMPERATURE = 30 + +THERMOMETER_IDS = ["Therm-1", "Therm-2", "Therm-3", "Therm-4", "Therm-5"] + +_random = SystemRandom() + + +def delivery_report(err: object, _msg: Message) -> None: + """Report a failed delivery; successful ones are not worth a line each.""" if err is not None: - print(f"Message delivery failed: {err}") + logger.error("Message delivery failed: %s", err) else: pass # Successfully delivered -def main(): +def main() -> None: + """Produce readings until interrupted.""" # Set up Kafka producer with confluent-kafka producer_config = {"bootstrap.servers": "localhost:9092"} producer = Producer(producer_config) - # Generate a fixed number of thermometer IDs - thermometer_ids = ["Therm-1", "Therm-2", "Therm-3", "Therm-4", "Therm-5"] - try: while True: - for thermometer_id in thermometer_ids: + for thermometer_id in THERMOMETER_IDS: # Generate a random temperature between -10 and 30 degrees - temperature = random.uniform(-10, 30) + temperature = _random.uniform(MIN_TEMPERATURE, MAX_TEMPERATURE) # Create reading object reading = TemperatureReading( @@ -52,17 +68,18 @@ def main(): callback=delivery_report, ) producer.flush(timeout=1) - # print(f"Sent: {json.dumps(reading_dict)}") + logger.debug("Sent: %s", json.dumps(reading_dict)) # Sleep for a second time.sleep(1) except KeyboardInterrupt: - print("Stopping temperature generator") + logger.info("Stopping temperature generator") finally: # Wait for any outstanding messages to be delivered producer.flush() if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) main() diff --git a/Programming/PSD/zin3/python/code/model/temperature_alarm.py b/Programming/PSD/zin3/python/code/model/temperature_alarm.py old mode 100644 new mode 100755 index da253701..50644adf --- a/Programming/PSD/zin3/python/code/model/temperature_alarm.py +++ b/Programming/PSD/zin3/python/code/model/temperature_alarm.py @@ -1,10 +1,27 @@ +#!/usr/bin/env python3 +"""The alarm the detector publishes to the Alarm topic.""" + +from __future__ import annotations + +from typing import Any + + class TemperatureAlarm: - def __init__(self, thermometer_id=None, timestamp=None, temperature=None): + """One below-zero reading: who reported it, when, and how cold.""" + + def __init__( + self, + thermometer_id: str | None = None, + timestamp: int | None = None, + temperature: float | None = None, + ) -> None: + """Store the three fields, all optional so a partial message still parses.""" self.thermometer_id = thermometer_id self.timestamp = timestamp self.temperature = temperature - def to_dict(self): + def to_dict(self) -> dict[str, Any]: + """Return the camelCase form that goes on the wire.""" return { "thermometerId": self.thermometer_id, "timestamp": self.timestamp, @@ -12,7 +29,8 @@ class TemperatureAlarm: } @classmethod - def from_dict(cls, data): + def from_dict(cls, data: dict[str, Any]) -> TemperatureAlarm: + """Build an alarm from the camelCase form that came off the wire.""" return cls( thermometer_id=data.get("thermometerId"), timestamp=data.get("timestamp"), diff --git a/Programming/PSD/zin3/python/code/model/temperature_reading.py b/Programming/PSD/zin3/python/code/model/temperature_reading.py old mode 100644 new mode 100755 index e3d5a04a..6826696f --- a/Programming/PSD/zin3/python/code/model/temperature_reading.py +++ b/Programming/PSD/zin3/python/code/model/temperature_reading.py @@ -1,10 +1,27 @@ +#!/usr/bin/env python3 +"""The reading the generator publishes to the Temperatura topic.""" + +from __future__ import annotations + +from typing import Any + + class TemperatureReading: - def __init__(self, thermometer_id=None, timestamp=None, temperature=None): + """One thermometer sample: who reported it, when, and how warm.""" + + def __init__( + self, + thermometer_id: str | None = None, + timestamp: int | None = None, + temperature: float | None = None, + ) -> None: + """Store the three fields, all optional so a partial message still parses.""" self.thermometer_id = thermometer_id self.timestamp = timestamp self.temperature = temperature - def to_dict(self): + def to_dict(self) -> dict[str, Any]: + """Return the camelCase form that goes on the wire.""" return { "thermometerId": self.thermometer_id, "timestamp": self.timestamp, @@ -12,7 +29,8 @@ class TemperatureReading: } @classmethod - def from_dict(cls, data): + def from_dict(cls, data: dict[str, Any]) -> TemperatureReading: + """Build a reading from the camelCase form that came off the wire.""" return cls( thermometer_id=data.get("thermometerId"), timestamp=data.get("timestamp"), diff --git a/Programming/PSD/zin3/python/code/processor/temperature_anomaly_detector.py b/Programming/PSD/zin3/python/code/processor/temperature_anomaly_detector.py old mode 100644 new mode 100755 index b7871fe2..df8c3c58 --- a/Programming/PSD/zin3/python/code/processor/temperature_anomaly_detector.py +++ b/Programming/PSD/zin3/python/code/processor/temperature_anomaly_detector.py @@ -1,6 +1,18 @@ +#!/usr/bin/env python3 +"""Flink job: read the Temperatura topic and republish below-zero readings. + +Run it against a local Kafka on 9092. The connector JARs are not bundled with +pyflink, so `add_kafka_dependencies` points the environment at a Kafka and a +Flink install and reports precisely which JAR is missing when one is. +""" + +from __future__ import annotations + import json -import os +import logging import sys +from pathlib import Path +from typing import TYPE_CHECKING from pyflink.common.serialization import SimpleStringSchema from pyflink.common.typeinfo import Types @@ -9,25 +21,48 @@ from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer from pyflink.datastream.functions import KeyedProcessFunction, MapFunction +if TYPE_CHECKING: + from collections.abc import Iterator + # Add parent directory to path to import model -sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +logger = logging.getLogger(__name__) + +# A reading at or below this temperature is what the job is looking for. +ALARM_TEMPERATURE = 0.0 + +# Where the Kafka and Flink installs live on the lab machine. +KAFKA_LIB_PATH = Path("/home/psd/Downloads/kafka_2.13-3.4.0/libs") +FLINK_LIB_PATH = Path("/home/psd/Downloads/flink-1.17.0/lib") +KAFKA_CLIENTS_JAR = "kafka-clients-3.4.0.jar" class ReadingMapFunction(MapFunction): - def map(self, value): + """Turn a JSON reading into the (id, timestamp, temperature) tuple.""" + + def map(self, value: str) -> tuple[str, int, float]: + """Parse one message off the Temperatura topic.""" data = json.loads(value) return (data["thermometerId"], data["timestamp"], data["temperature"]) class AnomalyDetectionFunction(KeyedProcessFunction): - def __init__(self): + """Emit an alarm for every reading below zero.""" + + def __init__(self) -> None: + """Set the window the job was written around.""" self.window_size = 10 * 1000 # 10 seconds in milliseconds - def process_element(self, value, ctx): + def process_element( + self, value: tuple[str, int, float], ctx: object + ) -> Iterator[str]: + """Yield a JSON alarm when `value` is below zero, otherwise nothing.""" + del ctx thermometer_id, timestamp, temperature = value # Check if temperature is below zero (anomaly) - if temperature < 0.0: + if temperature < ALARM_TEMPERATURE: # Create alarm alarm = { "thermometerId": thermometer_id, @@ -37,50 +72,40 @@ class AnomalyDetectionFunction(KeyedProcessFunction): yield json.dumps(alarm) -def add_kafka_dependencies(env): +def add_kafka_dependencies(env: StreamExecutionEnvironment) -> None: """Add Kafka connector dependencies to the environment.""" - # User-specific paths where JAR files are located - kafka_path = "/home/psd/Downloads/kafka_2.13-3.4.0/libs" - flink_path = "/home/psd/Downloads/flink-1.17.0/lib" - - # Required JAR files - kafka_jars = { - "kafka-clients": "kafka-clients-3.4.0.jar", - "flink-connector-kafka": None, # Will be detected or reported as missing - } - jar_paths = [] # Add kafka-clients JAR - clients_jar = os.path.join(kafka_path, kafka_jars["kafka-clients"]) - if os.path.exists(clients_jar): + clients_jar = KAFKA_LIB_PATH / KAFKA_CLIENTS_JAR + if clients_jar.exists(): jar_paths.append(f"file://{clients_jar}") else: - print(f"Warning: Could not find Kafka clients JAR at {clients_jar}") + logger.warning("Could not find Kafka clients JAR at %s", clients_jar) # Try to find Flink Kafka connector JAR - connector_found = False - for file in os.listdir(flink_path): - if file.startswith("flink-connector-kafka"): - kafka_jars["flink-connector-kafka"] = file - jar_paths.append(f"file://{os.path.join(flink_path, file)}") - connector_found = True - break - - if not connector_found: - print("Warning: No Flink Kafka connector JAR found! You need to download it.") - print("You can download it from Maven Central or the Apache Flink website.") - print("The JAR should be named like 'flink-connector-kafka-1.17.0.jar'") - print(f"Place it in: {flink_path}") + connector = next( + (p for p in sorted(FLINK_LIB_PATH.glob("flink-connector-kafka*"))), None + ) + if connector is not None: + jar_paths.append(f"file://{connector}") + else: + logger.warning( + "No Flink Kafka connector JAR found. Download one named like " + "'flink-connector-kafka-1.17.0.jar' from Maven Central or the " + "Apache Flink website and place it in %s", + FLINK_LIB_PATH, + ) if jar_paths: env.get_config().get_configuration().set_string( "pipeline.jars", ";".join(jar_paths) ) - print(f"Added JAR files: {jar_paths}") + logger.info("Added JAR files: %s", jar_paths) -def main(): +def main() -> None: + """Wire the Kafka source to the detector to the Kafka sink, and run it.""" # Create execution environment env = StreamExecutionEnvironment.get_execution_environment() @@ -116,16 +141,17 @@ def main(): output_type=Types.TUPLE([Types.STRING(), Types.LONG(), Types.DOUBLE()]), ).assign_timestamps_and_watermarks( WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner( - lambda event, timestamp: event[1] + lambda event, _timestamp: event[1] ) ).key_by(lambda x: x[0]).process(AnomalyDetectionFunction()).add_sink(producer) # Print the execution plan - print(env.get_execution_plan()) + logger.info("%s", env.get_execution_plan()) # Execute env.execute("Temperature Anomaly Detector with Time Windows") if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) main() diff --git a/Programming/PSD/zin3/python/code/visualizer/alarm_visualizer.py b/Programming/PSD/zin3/python/code/visualizer/alarm_visualizer.py old mode 100644 new mode 100755 index 6dd18b6b..cbd587d7 --- a/Programming/PSD/zin3/python/code/visualizer/alarm_visualizer.py +++ b/Programming/PSD/zin3/python/code/visualizer/alarm_visualizer.py @@ -1,21 +1,40 @@ +#!/usr/bin/env python3 +"""A Tk window listing the alarms as they arrive on the Alarm topic. + +The Kafka consumer runs on a daemon thread and hands each alarm back to the Tk +main loop with `root.after`, because Tk widgets may only be touched from the +thread that created them. +""" + +from __future__ import annotations + import json -import os +import logging import sys import threading import tkinter as tk from collections import Counter -from datetime import datetime +from datetime import datetime, timezone +from pathlib import Path from tkinter import ttk from confluent_kafka import Consumer # Add parent directory to path to import model -sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.append(str(Path(__file__).resolve().parent.parent)) from model.temperature_alarm import TemperatureAlarm +logger = logging.getLogger(__name__) + +# Milliseconds per second, for turning the wire timestamp into a datetime. +MILLISECONDS = 1000 + class AlarmVisualizer: - def __init__(self, root): + """The window: a scrolling list of alarms and a per-thermometer tally.""" + + def __init__(self, root: tk.Tk) -> None: + """Build the widgets and start consuming.""" self.root = root self.root.title("Temperature Alarm Visualizer") self.root.geometry("600x400") @@ -39,14 +58,15 @@ class AlarmVisualizer: self.stats_label.pack(pady=5) # Stats tracking - self.thermometer_alarm_count = Counter() + self.thermometer_alarm_count: Counter[str] = Counter() self.max_alarms = 100 # Start Kafka consumer in a separate thread self.consumer_thread = threading.Thread(target=self.consume_alarms, daemon=True) self.consumer_thread.start() - def consume_alarms(self): + def consume_alarms(self) -> None: + """Poll the Alarm topic forever, handing each alarm to the Tk thread.""" # Set up Kafka consumer with confluent-kafka consumer_config = { "bootstrap.servers": "localhost:9092", @@ -62,29 +82,39 @@ class AlarmVisualizer: if msg is None: continue if msg.error(): - print(f"Consumer error: {msg.error()}") + logger.error("Consumer error: %s", msg.error()) continue # Parse the alarm message - alarm_data = json.loads(msg.value().decode("utf-8")) + value = msg.value() + if value is None: + continue + alarm_data = json.loads(value.decode("utf-8")) alarm = TemperatureAlarm.from_dict(alarm_data) + if alarm.timestamp is None or alarm.thermometer_id is None: + logger.warning("Skipping malformed alarm: %s", alarm_data) + continue # Format the timestamp - timestamp_dt = datetime.fromtimestamp(alarm.timestamp / 1000) + timestamp_dt = datetime.fromtimestamp( + alarm.timestamp / MILLISECONDS, tz=timezone.utc + ).astimezone() formatted_date = timestamp_dt.strftime("%Y-%m-%d %H:%M:%S") # Create alarm message - alarm_message = f"⚠️ ALARM: Thermometer {alarm.thermometer_id} reported {alarm.temperature:.2f}°C at {formatted_date}" + alarm_message = ( + f"⚠️ ALARM: Thermometer {alarm.thermometer_id} " + f"reported {alarm.temperature:.2f}°C at {formatted_date}" + ) # Update the UI self.root.after(0, self.update_ui, alarm_message, alarm.thermometer_id) - except Exception as e: - print(f"Error in Kafka consumer: {e}") finally: consumer.close() - def update_ui(self, alarm_message, thermometer_id): + def update_ui(self, alarm_message: str, thermometer_id: str) -> None: + """Insert one alarm at the top of the list and refresh the tally.""" # Add new alarm to the top of the list self.alarm_list.insert(0, alarm_message) @@ -98,17 +128,20 @@ class AlarmVisualizer: # Format stats string if self.thermometer_alarm_count: stats_parts = [ - f"{id}={count}" for id, count in self.thermometer_alarm_count.items() + f"{name}={count}" + for name, count in self.thermometer_alarm_count.items() ] stats_text = "Alarm Counts: " + ", ".join(stats_parts) self.stats_label.config(text=stats_text) -def main(): +def main() -> None: + """Open the window and run the Tk main loop.""" root = tk.Tk() - app = AlarmVisualizer(root) + AlarmVisualizer(root) root.mainloop() if __name__ == "__main__": + logging.basicConfig(format="%(message)s", level=logging.INFO) main() diff --git a/TODO-refactor-remaining.md b/TODO-refactor-remaining.md index 01140663..545be79c 100644 --- a/TODO-refactor-remaining.md +++ b/TODO-refactor-remaining.md @@ -32,15 +32,14 @@ own docs and the rest of this fleet carry. A test asserts by raising. |---|---| | Baseline, before any of this | 9,281 | | After the format + safe-fix pass and excluding `TRAK/sightpy` | 3,686 | -| **Now** | **308** | +| **Now** | **156** | Courses at zero: `WDWR`, `SPD`, `MOM`, `ECRYPT`, `ECRYPT_PROJECT`, `ECOTE`, -`NLP`, `EARIN`, `PORR`, `TRAK`, `twm_4`, `PBAD`. What is left, by course: +`NLP`, `EARIN`, `PORR`, `TRAK`, `twm_4`, `PBAD`, `PSD`. What is left, by course: | Course | ruff | Can it be verified by running? | |---|---|---| | `ERSMS-project` | 156 | No: docker-compose course | -| `PSD` | 152 | No: `run.sh` predates the harness and installs packages with sudo | `course_run.sh`'s notebook branch is fixed: it executes to a temp `--output-dir`, probes jupyter by running it rather than by `command -v`, and @@ -82,7 +81,7 @@ RUFF findings unless it says otherwise**; only `TRAK` has been type-checked. | `TRAK` | **0** -- cleared 2026-08-29 | | `twm_4` | **0** -- cleared 2026-08-29 | | `PBAD` | not run (no venv; notebooks only) | -| `PSD/zin1` | 5 | +| `PSD` | **2**, both unfixable -- see below | Two structural notes for whoever runs it next: @@ -95,9 +94,19 @@ Two structural notes for whoever runs it next: `mypy --strict --python-executable Programming//.venv/bin/python `. Without that, TRAK reported 4 phantom `matplotlib` import errors on top of the real ones. +- **PSD keeps two mypy errors on purpose.** `--strict` implies + `disallow_subclassing_any`, and `temperature_anomaly_detector.py` has to + subclass pyflink's `MapFunction` and `KeyedProcessFunction`, which ship no + types. The only fixes are to weaken strict globally or to add an override + for our own module; neither is worth it, so the two are reported rather + than hidden. Everything else in PSD is clean under `--strict`. +- Point `MYPYPATH` at `Programming/PSD/zin3/python/code` for that tree: the + modules import `model.*` via a runtime `sys.path.append`, which mypy + cannot follow. - `pyproject.toml` now carries a `[[tool.mypy.overrides]]` block with `ignore_missing_imports` for `Imath`, `OpenEXR`, `OpenGL` and `matplotlib`. - Those four ship neither stubs nor `py.typed`; without it mypy cannot import + Those (plus `keras`, `tensorflow`, `sklearn`, `seaborn`, `kafka` and + `pyflink`) ship neither stubs nor `py.typed`; without it mypy cannot import them at all and reports nothing else about the file. It is not a suppression of our own code -- do not extend it to anything we wrote. diff --git a/pyproject.toml b/pyproject.toml index e7a55ab8..a0d6e149 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,6 +86,8 @@ module = [ "tensorflow.*", "sklearn.*", "seaborn.*", + "kafka.*", + "pyflink.*", ] ignore_missing_imports = true