display player type in Sociogram in RGB

cutt/analysis.py

@@ -126,6 +126,7 @@ def graph_json(
             return G
         return JSONResponse({"nodes": nodes, "edges": edges})
 
+    playertypes = playertype(request)
     with Session(engine) as session:
         players = session.exec(
             select(P)
@@ -140,7 +141,18 @@ def graph_json(
             )
         for p in players:
             player_map[p.id] = p.display_name
-            nodes.append({"id": p.display_name, "label": p.display_name})
+            playertype_colour = "#%02x%02x%02x" % (
+                int(playertypes[p.id]["handler"] * 255),
+                int(playertypes[p.id]["combi"] * 255),
+                int(playertypes[p.id]["cutter"] * 255),
+            )
+            nodes.append(
+                {
+                    "id": p.display_name,
+                    "label": p.display_name,
+                    "data": {"playertype": playertype_colour},
+                }
+            )
 
         subquery = (
             select(C.user, func.max(C.time).label("latest"))
@@ -207,7 +219,8 @@ def graph_json(
     )
     in_degrees = G.in_degree(weight="weight")
     nodes = [
-        dict(node, **{"data": {"inDegree": in_degrees[node["id"]]}}) for node in nodes
+        dict(node, **{"data": {"inDegree": in_degrees[node["id"]], **node["data"]}})
+        for node in nodes
     ]
     if networkx_graph:
         return G
@@ -434,6 +447,52 @@ def mvp(
     ]
 
 
+def playertype(request: Annotated[TeamScopedRequest, Security(verify_team_scope)]):
+    with Session(engine) as session:
+        players = session.exec(
+            select(P)
+            .join(PlayerTeamLink)
+            .join(Team)
+            .where(Team.id == request.team_id, P.disabled == False)
+        ).all()
+        if not players:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+        player_map = {p.id: p for p in players}
+        subquery = (
+            select(PT.user, func.max(PT.time).label("latest"))
+            .where(PT.team == request.team_id)
+            .group_by(PT.user)
+            .subquery()
+        )
+        statement2 = select(PT).join(
+            subquery, (PT.user == subquery.c.user) & (PT.time == subquery.c.latest)
+        )
+        playertypes = {}
+        for pt in session.exec(statement2):
+            for i, p_id in enumerate(pt.handlers):
+                if p_id not in player_map:
+                    continue
+                playertypes[p_id] = playertypes.get(p_id, []) + [-1]
+            for i, p_id in enumerate(pt.combis):
+                if p_id not in player_map:
+                    continue
+                playertypes[p_id] = playertypes.get(p_id, []) + [0]
+            for i, p_id in enumerate(pt.cutters):
+                if p_id not in player_map:
+                    continue
+                playertypes[p_id] = playertypes.get(p_id, []) + [1]
+    playertype_analysis = {}
+    for p_id, v in playertypes.items():
+        v = np.array(v)
+        playertype_analysis[p_id] = {
+            "mean": np.mean(v),
+            "handler": (v == -1).sum() / len(v),
+            "combi": (v == 0).sum() / len(v),
+            "cutter": (v == 1).sum() / len(v),
+        }
+    return playertype_analysis
+
+
 async def turnout(
     request: Annotated[
         TeamScopedRequest, Security(verify_team_scope, scopes=["analysis"])
@@ -504,6 +563,9 @@ analysis_router.add_api_route(
     name="MVPs",
     description="Request Most Valuable Players stats",
 )
+analysis_router.add_api_route(
+    "/playertype/{team_id}", endpoint=playertype, methods=["GET"]
+)
 analysis_router.add_api_route("/turnout/{team_id}", endpoint=turnout, methods=["GET"])
 analysis_router.add_api_route(
     "/times/{team_id}", endpoint=last_submissions, methods=["GET"]