Corteva_Project

Invesxon Project/Weather_data_DDL.sql

@@ -0,0 +1,16 @@
+-- public.weather_data definition
+
+-- Drop table
+
+-- DROP TABLE public.weather_data;
+
+CREATE TABLE public.weather_data (
+	id serial4 NOT NULL,
+	station_id varchar(50) NOT NULL,
+	"date" date NOT NULL,
+	max_temp int4 NULL,
+	min_temp int4 NULL,
+	precipitation int4 NULL,
+	CONSTRAINT weather_data_pkey PRIMARY KEY (id),
+	CONSTRAINT weather_data_station_id_date_key UNIQUE (station_id, date)
+);

Invesxon Project/app.py

@@ -0,0 +1,87 @@
+from flask import Flask, request, jsonify
+from flask_restful import Api, Resource
+from flask_sqlalchemy import SQLAlchemy
+from flasgger import Swagger
+
+app = Flask(__name__)
+app.config['SQLALCHEMY_DATABASE_URI'] = "postgresql://postgres:Welcome10%40@localhost:5432/postgres"
+app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
+db = SQLAlchemy(app)
+api = Api(app)
+swagger = Swagger(app)
+
+class WeatherData(db.Model):
+    __tablename__ = 'weather_data'
+    id = db.Column(db.Integer, primary_key=True)
+    station_id = db.Column(db.String, nullable=False)
+    date = db.Column(db.Date, nullable=False)
+    max_temp = db.Column(db.Integer)
+    min_temp = db.Column(db.Integer)
+    precipitation = db.Column(db.Integer)
+    __table_args__ = (db.UniqueConstraint('station_id', 'date', name='_station_date_uc'),)
+
+class YearlyWeatherStats(db.Model):
+    __tablename__ = 'yearly_weather_stats'
+    id = db.Column(db.Integer, primary_key=True)
+    station_id = db.Column(db.String, nullable=False)
+    year = db.Column(db.Integer, nullable=False)
+    avg_max_temp = db.Column(db.Numeric)
+    avg_min_temp = db.Column(db.Numeric)
+    total_precipitation = db.Column(db.Numeric)
+    __table_args__ = (db.UniqueConstraint('station_id', 'year', name='_station_year_uc'),)
+
+class WeatherDataResource(Resource):
+    def get(self):
+        station_id = request.args.get('station_id')
+        date = request.args.get('date')
+        page = request.args.get('page', 1, type=int)
+        per_page = request.args.get('per_page', 10, type=int)
+
+        query = WeatherData.query
+        if station_id:
+            query = query.filter_by(station_id=station_id)
+        if date:
+            query = query.filter_by(date=date)
+        result = query.paginate(page=page, per_page=per_page)
+
+        return jsonify({
+            'total': result.total,
+            'pages': result.pages,
+            'page': result.page,
+            'per_page': result.per_page,
+            'items': [item.as_dict() for item in result.items]
+        })
+
+class WeatherStatsResource(Resource):
+    def get(self):
+        station_id = request.args.get('station_id')
+        year = request.args.get('year')
+        page = request.args.get('page', 1, type=int)
+        per_page = request.args.get('per_page', 10, type=int)
+
+        query = YearlyWeatherStats.query
+        if station_id:
+            query = query.filter_by(station_id=station_id)
+        if year:
+            query = query.filter_by(year=year)
+        result = query.paginate(page=page, per_page=per_page)
+
+        return jsonify({
+            'total': result.total,
+            'pages': result.pages,
+            'page': result.page,
+            'per_page': result.per_page,
+            'items': [item.as_dict() for item in result.items]
+        })
+
+def to_dict(self):
+    return {c.name: getattr(self, c.name) for c in self.__table__.columns}
+
+WeatherData.as_dict = to_dict
+YearlyWeatherStats.as_dict = to_dict
+
+api.add_resource(WeatherDataResource, '/api/weather')
+api.add_resource(WeatherStatsResource, '/api/weather/stats')
+
+if __name__ == "__main__":
+    app.run(debug=True)

Invesxon Project/dataingestion.py

@@ -0,0 +1,79 @@
+import os
+import glob
+import logging
+from datetime import datetime
+from sqlalchemy import create_engine, Column, Integer, String, Date, UniqueConstraint
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import psycopg2
+import pandas as pd
+
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Database setup
+DATABASE_URL = "postgresql://postgres:Welcome10%40@localhost:5432/postgres"
+engine = create_engine(DATABASE_URL)
+Session = sessionmaker(bind=engine)
+session = Session()
+Base = declarative_base()
+
+class WeatherData(Base):
+    __tablename__ = 'weather_data'
+    id = Column(Integer, primary_key=True)
+    station_id = Column(String, nullable=False)
+    date = Column(Date, nullable=False)
+    max_temp = Column(Integer)
+    min_temp = Column(Integer)
+    precipitation = Column(Integer)
+    __table_args__ = (UniqueConstraint('station_id', 'date', name='_station_date_uc'),)
+
+Base.metadata.create_all(engine)
+
+def ingest_data(file_path):
+    try:
+        logger.info(f"Starting data ingestion from {file_path}")
+        start_time = datetime.now()
+        record_count = 0
+        with open(file_path, 'r') as file:
+            for line in file:
+                date_str, max_temp, min_temp, precipitation = line.strip().split('\t')
+                if int(max_temp) == -9999: max_temp = None
+                if int(min_temp) == -9999: min_temp = None
+                if int(precipitation) == -9999: precipitation = None
+
+                weather_record = WeatherData(
+                    station_id=os.path.basename(file_path).split('.')[0],
+                    date=datetime.strptime(date_str, '%Y%m%d').date(),
+                    max_temp=max_temp,
+                    min_temp=min_temp,
+                    precipitation=precipitation
+                )
+                session.merge(weather_record)
+                record_count += 1
+
+        session.commit()
+        end_time = datetime.now()
+        logger.info(f"Completed data ingestion from {file_path} with {record_count} records in {end_time - start_time}")
+    except:
+        print("This data is already exists")
+
+if __name__ == "__main__":
+    data_files = glob.glob("wx_data/*.txt")
+    for file_path in data_files:
+        ingest_data(file_path)
+
+
+
+try:
+    conn = psycopg2.connect("dbname=postgres user=postgres password=Welcome10@ host=localhost port=5432")
+    print("Connection successful")
+except Exception as e:
+    print(f"Error connecting to the database: {e}")
+
+con=conn.cursor()
+# df=pd.read_sql("with new_table as (select *,date_part('year',date) as Year from weather_data) select station_id,year,avg(max_temp) as avg_maxtemp,avg(min_temp) as avg_mintemp,sum(precipitation) as total_precipitation  from new_table group by station_id,year",conn)
+con.execute("create table if not exists postgres.public.mytab as select station_id,date_part('year',date) as year,avg(max_temp) as avg_maxtemp,avg(min_temp) as avg_mintemp,sum(precipitation) as total_precipitation  from weather_data group by station_id,year")
+conn.commit()