Merge pull request #11 from rogers-obrien-rad/feature/9/schedule

Feature/9/schedule

README.md

@@ -1,4 +1,4 @@
-[![test suite](https://github.com/rogers-obrien-rad/package-template/actions/workflows/tests.yml/badge.svg)](https://github.com/rogers-obrien-rad/package-template/actions/workflows/tests.yml)
+[![test suite](https://github.com/rogers-obrien-rad/smartpm-python-sdk/actions/workflows/tests.yml/badge.svg)](https://github.com/rogers-obrien-rad/smartpm-python-sdk/actions/workflows/tests.yml)
 
 ![ro_logo](https://github.com/rogers-obrien-rad/general-template/blob/main/images/ro_logo.png)
 
@@ -8,7 +8,7 @@ _Python wrapper for the SmartPM API_
 # Installation
 
 # Snippets
-
+Python code snippets showcasing the various methods for each endpoint can be found [here](https://github.com/rogers-obrien-rad/smartpm-python-sdk/tree/main/snippets).
 
 # Resources
 * [SmartPM Platform Login](https://live.smartpmtech.com/auth/login)

smartpm/endpoints/schedule.py

@@ -0,0 +1,144 @@
+from smartpm.client import SmartPMClient
+from smartpm.decorators import api_wrapper, utility
+from smartpm.logging_config import logger
+
+class Schedule:
+    def __init__(self, client: SmartPMClient):
+        self.client = client
+
+    @api_wrapper
+    def get_schedule_quality(self, project_id, scenario_id, import_log_id=None, quality_profile_id=None):
+        """
+        Get the schedule quality for a specific project and scenario: https://developers.smartpmtech.com/#operation/get-schedule-quality
+
+        Parameters
+        ----------
+        project_id : str
+            The Project ID containing the scenario for which you would like to pull the schedule quality for
+        scenario_id : str
+            The Scenario ID for which you would like to pull the schedule quality for
+        import_log_id : str, default None
+            The schedule id for which you would like to see the schedule quality for, if not specified, the latest date will be used
+        quality_profile_id : str, default None
+            The quality profile you would like to use, if not specified the default profile for the project will be used
+
+        Returns
+        -------
+        <response.json> : dict
+            Schedule quality data as a JSON object
+        """
+        logger.debug(f"Fetching schedule quality for project_id: {project_id}, scenario_id: {scenario_id}, import_log_id: {import_log_id}, quality_profile_id: {quality_profile_id}")
+        params = {}
+        if import_log_id:
+            params['importLogId'] = import_log_id
+        if quality_profile_id:
+            params['qualityProfileId'] = quality_profile_id
+
+        endpoint = f'v1/projects/{project_id}/scenarios/{scenario_id}/schedule-quality'
+        response = self.client._get(endpoint=endpoint, params=params)
+
+        return response
+
+    @utility
+    def get_metric_by_name(self, schedule_quality_data, metric_name):
+        """
+        Get a specific metric by its name from the schedule quality data.
+
+        Parameters
+        ----------
+        schedule_quality_data : dict
+            The schedule quality data as a JSON object
+        metric_name : str
+            The name of the metric to retrieve
+
+        Returns
+        -------
+        metric : dict
+            The metric data if found, otherwise None
+        """
+        metrics = schedule_quality_data.get('metrics', [])
+        for metric in metrics:
+            if metric.get('name') == metric_name:
+                logger.debug(f"Found {metric_name}")
+                return metric
+
+        logger.warning(f"Could not find metric {metric_name}")
+        return None
+
+    @utility
+    def get_schedule_grade(self, schedule_quality_data):
+        """
+        Get the overall grade from the schedule quality data.
+
+        Parameters
+        ----------
+        schedule_quality_data : dict
+            The schedule quality data as a JSON object
+
+        Returns
+        -------
+        dict or None
+            The grade data if found, otherwise None
+        """
+        return schedule_quality_data.get('grade')
+
+    @api_wrapper
+    def get_schedule_compression(self, project_id, scenario_id, data_date=None):
+        """
+        Get the schedule compression for a specific project and scenario: https://live.smartpmtech.com/public/v1/projects/{projectId}/scenarios/{scenarioId}/schedule-compression
+
+        Parameters
+        ----------
+        project_id : str
+            The Project ID containing the scenario for which you would like to pull the schedule compression for
+        scenario_id : str
+            The Scenario ID for which you would like to pull the schedule compression for
+        data_date : str, optional
+            The schedule data date you would like to retrieve schedule compression for
+
+        Returns
+        -------
+        <response.json> : dict
+            Schedule compression data as a JSON object
+        """
+        logger.debug(f"Fetching schedule compression for project_id: {project_id}, scenario_id: {scenario_id}, data_date: {data_date}")
+        params = {}
+        if data_date:
+            params['dataDate'] = data_date
+
+        endpoint = f'v1/projects/{project_id}/scenarios/{scenario_id}/schedule-compression'
+        response = self.client._get(endpoint=endpoint, params=params)
+        return response
+
+    @api_wrapper
+    def get_all_quality_profiles(self):
+        """
+        Get all quality profiles: https://developers.smartpmtech.com/#operation/get-schedule-quality (scroll down - no direct link)
+
+        Returns
+        -------
+        <response.json> : list of dict
+            quality profiles data as a JSON object
+        """
+        endpoint = 'v1/quality-profiles'
+        return self.client._get(endpoint=endpoint)
+
+    @api_wrapper
+    def get_quality_profile(self, quality_profile_id):
+        """
+        Get a specific quality profile: https://developers.smartpmtech.com/#operation/get-quality-profile
+
+        Parameters
+        ----------
+        quality_profile_id : str
+            The quality profile you want to pull configuration for
+
+        Returns
+        -------
+        <response.json> : dict
+            Quality profile data as a JSON object
+        """
+        logger.debug(f"Fetching quality profile for quality_profile_id: {quality_profile_id}")
+        endpoint = f'v1/quality-profiles/{quality_profile_id}'
+        response = self.client._get(endpoint=endpoint)
+        return response

snippets/explore_schedule.py

@@ -0,0 +1,81 @@
+import os
+import sys
+import json
+
+# Add the package root directory to the sys.path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../')))
+
+from dotenv import load_dotenv
+load_dotenv()  # Load environment variables from .env file
+
+from smartpm.client import SmartPMClient
+from smartpm.endpoints.projects import Projects # import projects to get project IDs
+from smartpm.endpoints.scenarios import Scenarios
+from smartpm.endpoints.schedule import Schedule
+
+API_KEY = os.getenv("API_KEY")
+COMPANY_ID = os.getenv("COMPANY_ID")
+
+def main():
+    # Setup SDK
+    client = SmartPMClient(API_KEY, COMPANY_ID)
+    projects_api = Projects(client)
+    scenarios_api = Scenarios(client)
+    schedule_api = Schedule(client)
+
+    # Get Schedule Quality
+    # --------------------
+    # Find project by name
+    name_to_find = "212096 - 401 FIRST STREET (College Station)" # replace with your project name
+    project = projects_api.find_project_by_name(name=name_to_find)
+    project_id = project["id"]
+
+    # Find scenario by name
+    scenario_to_find = "Full Schedule" # replace with your scenario name
+    matching_scenarios = scenarios_api.find_scenario_by_name(
+        project_id=project_id,
+        scenario_name=scenario_to_find
+    )
+    scenario_id = matching_scenarios[-1].get("id")
+
+    print("Get Schedule Quality")
+    schedule_quality_data = schedule_api.get_schedule_quality(
+        project_id=project_id,
+        scenario_id=scenario_id
+    )
+    print(json.dumps(schedule_quality_data, indent=4))
+    # --------------------
+
+    # Get Schedule Quality Metrics
+    # ----------------------------
+    print("Get Schedule Grade")
+    grade_data = schedule_api.get_schedule_grade(schedule_quality_data=schedule_quality_data)
+    print(f"Grade is {grade_data.get('mark')} ({round(grade_data.get('score'), 1)}%) - {grade_data.get('indicator')}")
+
+    print("Get Schedule Metrics")
+    print("Critical Path Percent")
+    critical_path_data = schedule_api.get_metric_by_name(
+        schedule_quality_data=schedule_quality_data,
+        metric_name="CRITICAL_PATH_PERCENT"
+    )
+    print(json.dumps(critical_path_data, indent=4))
+    print("Finish to Start")
+    finish_to_start_data = schedule_api.get_metric_by_name(
+        schedule_quality_data=schedule_quality_data,
+        metric_name="RELATIONSHIPS_FINISH_TO_START"
+    )
+    print(json.dumps(finish_to_start_data, indent=4))
+    # ----------------------------
+
+    # Get Schedule Compression
+    # ------------------------
+    print("Get Schedule Compression")
+    schedule_compression_data = schedule_api.get_schedule_compression(
+        project_id=project_id,
+        scenario_id=scenario_id
+    )
+    print(json.dumps(schedule_compression_data, indent=4))
+    # ------------------------
+
+if __name__ == "__main__":
+    main()

tests/test_activities.py

@@ -0,0 +1,104 @@
+import pytest
+import os
+import sys
+import logging
+import json
+
+# Add the package root directory to the sys.path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../')))
+
+from dotenv import load_dotenv
+load_dotenv()  # Load environment variables from .env file
+
+from smartpm.client import SmartPMClient
+from smartpm.endpoints.scenarios import Scenarios
+from smartpm.endpoints.projects import Projects
+from smartpm.endpoints.activity import Activity
+
+API_KEY = os.getenv("API_KEY")
+COMPANY_ID = os.getenv("COMPANY_ID")
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+@pytest.fixture
+def client():
+    return SmartPMClient(API_KEY, COMPANY_ID)
+
+@pytest.fixture
+def activity(client):
+    return Activity(client)
+
+@pytest.fixture
+def scenarios(client):
+    return Scenarios(client)
+
+@pytest.fixture
+def projects(client):
+    return Projects(client)
+
+def test_get_activities(activity, scenarios, projects):
+    """Test retrieving activities for a specific scenario by its ID."""
+    # Get a list of projects to use a valid project ID
+    all_projects = projects.get_projects()
+    logger.info("Number of projects: %d", len(all_projects))
+
+    assert isinstance(all_projects, list)
+    assert len(all_projects) > 0
+
+    first_project = all_projects[0]
+    project_id = first_project['id']
+
+    # Retrieve scenarios for the first project
+    all_scenarios = scenarios.get_scenarios(project_id)
+    logger.info("Project ID: %s", project_id)
+
+    assert isinstance(all_scenarios, list)
+    assert len(all_scenarios) > 0
+
+    first_scenario = all_scenarios[0]
+    scenario_id = first_scenario['id']
+
+    # Retrieve activities for the first scenario
+    activities = activity.get_activities(project_id, scenario_id)
+    logger.info("Scenario ID: %s", scenario_id)
+
+    # Pretty-print the activities
+    pretty_activities = json.dumps(activities, indent=4)
+    logger.info("Activities: %s", pretty_activities)
+
+    assert isinstance(activities, list)
+
+def test_count_activities_by_completion(activity, scenarios, projects):
+    """Test counting complete and incomplete activities for a specific scenario by its ID."""
+    # Get a list of projects to use a valid project ID
+    all_projects = projects.get_projects()
+    logger.info("Number of projects: %d", len(all_projects))
+
+    assert isinstance(all_projects, list)
+    assert len(all_projects) > 0
+
+    first_project = all_projects[0]
+    project_id = first_project['id']
+
+    # Retrieve scenarios for the first project
+    all_scenarios = scenarios.get_scenarios(project_id)
+    logger.info("Project ID: %s", project_id)
+
+    assert isinstance(all_scenarios, list)
+    assert len(all_scenarios) > 0
+
+    first_scenario = all_scenarios[0]
+    scenario_id = first_scenario['id']
+
+    # Count complete and incomplete activities for the first scenario
+    completion_counts = activity.count_activities_by_completion(project_id, scenario_id)
+    logger.info("Scenario ID: %s", scenario_id)
+    logger.info("Completion Counts: %s", completion_counts)
+
+    assert isinstance(completion_counts, dict)
+    assert 'complete' in completion_counts
+    assert 'incomplete' in completion_counts
+    assert isinstance(completion_counts['complete'], int)
+    assert isinstance(completion_counts['incomplete'], int)