Merge pull request #201 from hotosm/feature/tm-stats

Tasking Manager stats

API/main.py

@@ -29,6 +29,7 @@
 from .data_quality import router as data_quality_router
 from .trainings import router as training_router
 from .organization import router as organization_router
+from .tasking_manager import router as tm_router
 from .raw_data import router as raw_data_router
 
 # This is used for local setup for auth login
@@ -57,6 +58,7 @@
 app.include_router(data_quality_router)
 app.include_router(training_router)
 app.include_router(organization_router)
+app.include_router(tm_router)
 app.include_router(raw_data_router)
 
 

API/tasking_manager.py

@@ -0,0 +1,67 @@
+# Copyright (C) 2021 Humanitarian OpenStreetmap Team
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+# Humanitarian OpenStreetmap Team
+# 1100 13th Street NW Suite 800 Washington, D.C. 20005
+# <info@hotosm.org>
+
+"""[Router Responsible for Organizational data API ]
+"""
+from fastapi import APIRouter, Depends
+
+from .auth import login_required
+
+from src.galaxy.tasking_manager.models import ValidatorStatsRequest
+from src.galaxy.app import TaskingManager
+
+from fastapi.responses import StreamingResponse
+
+from datetime import datetime
+
+
+router = APIRouter(prefix="/tasking-manager")
+
+@router.post("/validators")
+def get_validator_stats(request: ValidatorStatsRequest):
+    tm = TaskingManager(request)
+    csv_stream = tm.get_validators_stats()
+
+    response = StreamingResponse(csv_stream)
+    name =f"ValidatorStats_{datetime.now().isoformat()}"
+    response.headers["Content-Disposition"] = f"attachment; filename={name}.csv"
+
+    return response
+
+
+@router.get("/teams")
+def get_teams():
+    csv_stream = TaskingManager().list_teams()
+
+    response = StreamingResponse(csv_stream)
+    name =f"Teams_{datetime.now().isoformat()}"
+    response.headers["Content-Disposition"] = f"attachment; filename={name}.csv"
+
+    return response
+
+
+@router.get("/teams/individual")
+def get_teams():
+    csv_stream = TaskingManager().list_teams_metadata()
+
+    response = StreamingResponse(csv_stream)
+    name =f"Teams_{datetime.now().isoformat()}"
+    response.headers["Content-Disposition"] = f"attachment; filename={name}.csv"
+
+    return response

src/galaxy/app.py

@@ -35,6 +35,7 @@
 from json import loads as json_loads
 from geojson import Feature, FeatureCollection, Point
 from io import StringIO
+from csv import DictWriter
 from .config import config
 import logging
 import orjson
@@ -280,6 +281,58 @@ def get_time_spent_mapping_and_validating_per_user(self):
             return time_spent_mapping_result, time_spent_validating_result
         return [],[]   
 
+    def get_validators_stats(self):
+        query = generate_tm_validators_stats_query(self.cur, self.params)
+        result = [dict(r) for r in self.database.executequery(query)]
+
+        indexes = ['user_id', 'username']
+        columns = ['project_id', 'country', 'total_tasks', 'tasks_mapped', 'tasks_validated']
+
+        df = pandas.DataFrame(result)
+        out = pandas.pivot_table(df,
+            values='cnt',
+            index=indexes,
+            columns=columns,
+            fill_value=0
+        ).swaplevel(0, 1).reset_index()
+
+        stream = StringIO()
+        out.to_csv(stream)
+
+        return iter(stream.getvalue())
+
+    def list_teams(self):
+        query = generate_tm_teams_list()
+        results_dicts = [dict(r) for r in self.database.executequery(query)]
+
+        stream = StringIO()
+
+        csv_keys: List[str] = list(results_dicts[0].keys())
+        writer = DictWriter(stream, fieldnames=csv_keys)
+        writer.writeheader()
+
+        [writer.writerow(row) for row in results_dicts]
+
+        return iter(stream.getvalue())
+
+    def list_teams_metadata(self):
+        query = generate_list_teams_metadata()
+        results_dicts = [dict(r) for r in self.database.executequery(query)]
+
+        results_dicts = [{**r, "function": TeamMemberFunction(r["function"]).name.lower()}
+            for r in results_dicts]
+
+        stream = StringIO()
+
+        csv_keys: List[str] = list(results_dicts[0].keys())
+        writer = DictWriter(stream, fieldnames=csv_keys)
+        writer.writeheader()
+
+        [writer.writerow(row) for row in results_dicts]
+
+        return iter(stream.getvalue())
+
+
 class Mapathon:
     """Class for mapathon detail report and summary report this is the class that self connects to database and provide you summary and detail report."""
 
@@ -624,7 +677,6 @@ def get_trainingslist(self, params: TrainingParams):
         # print(Trainings_list)
         return Trainings_list
 
-
 class OrganizationHashtags:
     """[Class responsible for Organization Hashtag data API]
     """
@@ -647,7 +699,7 @@ def get_report_as_csv(self, filelocation):
             result = Output(self.query, self.con).to_CSV(filelocation)
             return result
         except Exception as err:
-            return err
+            return err  
 
 class RawData:
     """Class responsible for the Rawdata Extraction from available sources , Currently Works for Underpass source Current Snapshot

src/galaxy/query_builder/builder.py

@@ -580,6 +580,59 @@ def generate_organization_hashtag_reports(cur,params):
     # print(query)
     return query
 
+
+def generate_tm_validators_stats_query(cur, params):
+    stmt = """WITH t1 as (SELECT user_id, project_id, count(id) AS cnt from task_history
+    where action_text = 'VALIDATED' AND date_part('year', action_date) = %s group by user_id, project_id order by project_id)
+    """
+
+    sub_query = cur.mogrify(sql.SQL(stmt), (params.year,)).decode()
+
+    query = f"""
+    {sub_query}
+    SELECT t1.user_id,
+        u.username,
+        t1.project_id,
+        t1.cnt,
+        p.total_tasks,
+        p.tasks_mapped,
+        p.tasks_validated,
+        unnest(p.country) AS country
+    from t1, projects as p, users as u
+    where t1.project_id = p.id AND u.id = t1.user_id
+    ORDER BY u.username, t1.project_id
+    """
+
+    return query
+
+
+def generate_tm_teams_list():
+    query = """with vt AS (SELECT distinct team_id as id from project_teams where role = 1 order by id),
+            mu AS (SELECT tm.team_id, ARRAY_AGG(users.username) AS managers from team_members AS tm, vt, users WHERE users.id = tm.user_id AND tm.team_id = vt.id AND tm.function = 1 GROUP BY tm.team_id),
+            uc AS (SELECT tm.team_id, count(tm.user_id) AS members_count from team_members AS tm, vt WHERE tm.team_id = vt.id GROUP BY tm.team_id)
+            SELECT t.id, t.organisation_id, orgs.name AS organisation_name, t.name AS team_name, mu.managers, uc.members_count from teams AS t, mu, uc, organisations AS orgs where orgs.id = t.organisation_id AND t.id = mu.team_id AND t.id = uc.team_id"""
+
+    return query
+
+
+def generate_list_teams_metadata():
+    query = """
+        with vt AS (SELECT distinct team_id as id from project_teams where role = 1 order by id),
+        m AS (SELECT tm.team_id, tm.user_id, users.username, tm.function FROM team_members AS tm, vt, users WHERE users.id = tm.user_id AND tm.team_id = vt.id)
+        SELECT m.team_id AS team_id,
+            t.name AS team_name,
+            orgs.id AS organisation_id,
+            orgs.name AS organisation_name,
+            m.user_id,
+            m.username,
+            m.function from m, teams as t, organisations as orgs
+        where
+            orgs.id = t.organisation_id AND
+            t.id = m.team_id
+        ORDER BY team_id, function, username;
+    """
+
+    return query
 def raw_historical_data_extraction_query(cur,conn,params):
     geometry_dump = dumps(dict(params.geometry))
     geom_filter = f"ST_intersects(ST_GEOMFROMGEOJSON('{geometry_dump}'), geom)"
@@ -849,4 +902,4 @@ def raw_currentdata_extraction_query(params,c_id,geometry_dump,geom_area,ogr_exp
 def check_last_updated_rawdata():
     query = f"""select NOW()-importdate as last_updated from planet_osm_replication_status"""
     return query
-
+

src/galaxy/tasking_manager/models.py

@@ -0,0 +1,8 @@
+from ..validation.models import BaseModel
+from typing import Optional
+
+
+class ValidatorStatsRequest(BaseModel):
+    year: int = 2012
+    country: Optional[str] = None
+    organisation: Optional[str] = None

src/galaxy/validation/models.py

@@ -413,6 +413,12 @@ class OrganizationHashtag(BaseModel):
     total_unique_contributors : int
     total_new_road_meters : int 
 
+
+class TeamMemberFunction(Enum):
+    """Describes the function a member can hold within a team"""
+
+    MANAGER = 1
+    MEMBER = 2
 class RawDataOutputType ( Enum):
     GEOJSON ="GeoJSON"
     KML = "KML"