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